"""Fail-closed checks for the production delivery workflow. This test intentionally uses only the Python standard library so it can run before project dependencies are installed. The small helpers parse indentation-delimited YAML blocks that are relevant to the policy; they are not a general YAML parser. """ from __future__ import annotations import re import subprocess import unittest from pathlib import Path REPOSITORY_ROOT = Path(__file__).resolve().parents[2] DEPLOY_WORKFLOW = REPOSITORY_ROOT / ".gitea" / "workflows" / "unraid-deploy.yml" VALIDATION_WORKFLOW = REPOSITORY_ROOT / ".gitea" / "workflows" / "managed-validation.yml" def _indented_block(text: str, header: str) -> str: """Return a YAML block beginning at an exact header line.""" lines = text.splitlines() try: start = lines.index(header) except ValueError as exc: raise AssertionError(f"Missing YAML header: {header!r}") from exc indent = len(header) - len(header.lstrip()) end = len(lines) for index in range(start + 1, len(lines)): line = lines[index] if not line.strip() or line.lstrip().startswith("#"): continue current_indent = len(line) - len(line.lstrip()) if current_indent <= indent: end = index break return "\n".join(lines[start:end]) def _named_step(text: str, name: str) -> str: return _indented_block(text, f" - name: {name}") def _run_script(step: str) -> str: """Extract one inline or literal run body without normalizing shell content.""" lines = step.splitlines() if " run: |" not in lines: inline = next((line for line in lines if line.startswith(" run: ")), None) if inline is None: raise AssertionError("Named step does not contain a run command") return inline.removeprefix(" run: ") + "\n" start = lines.index(" run: |") + 1 script_lines = lines[start:] if any(line and not line.startswith(" ") for line in script_lines): raise AssertionError("Unexpected indentation in literal run block") return "\n".join(line[10:] if line else "" for line in script_lines) + "\n" class ProductionDeployPolicyTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.deploy = DEPLOY_WORKFLOW.read_text(encoding="utf-8") cls.validation = VALIDATION_WORKFLOW.read_text(encoding="utf-8") def test_deployment_has_no_automatic_trigger(self) -> None: trigger = _indented_block(self.deploy, "on:") trigger_keys = re.findall(r"^ ([a-z_]+):", trigger, flags=re.MULTILINE) self.assertEqual(trigger_keys, ["workflow_dispatch"]) self.assertNotRegex(trigger, r"(?m)^\s+push:") def test_dispatch_inputs_are_required_and_fail_safe(self) -> None: inputs = _indented_block(self.deploy, " inputs:") input_names = re.findall(r"^ ([a-z_]+):", inputs, flags=re.MULTILINE) self.assertEqual(input_names, ["release_tag", "release_commit", "action"]) for input_name in ("release_tag", "release_commit"): block = _indented_block(self.deploy, f" {input_name}:") self.assertRegex(block, r"(?m)^ required: true$") self.assertRegex(block, r"(?m)^ type: string$") action = _indented_block(self.deploy, " action:") self.assertRegex(action, r"(?m)^ required: true$") self.assertRegex(action, r"(?m)^ default: VERIFY_ONLY$") self.assertRegex(action, r"(?m)^ type: choice$") options = re.findall(r"^ - (\S+)$", action, flags=re.MULTILINE) self.assertEqual(options, ["VERIFY_ONLY", "DEPLOY_STABLE_TO_PRODUCTION"]) def test_inputs_are_validated_before_checkout(self) -> None: validation = _named_step(self.deploy, "Validate immutable dispatch inputs") checkout = _named_step(self.deploy, "Check out the exact release commit with full history") self.assertLess(self.deploy.index(validation), self.deploy.index(checkout)) self.assertIn('"refs/heads/master"', validation) self.assertIn( r"^v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$", validation, ) self.assertIn(r"^[0-9a-f]{40}$", validation) self.assertIn("VERIFY_ONLY|DEPLOY_STABLE_TO_PRODUCTION", validation) tag_pattern = re.search(r'RELEASE_TAG\}" =~ (\^.*\$) \]\]', validation) commit_pattern = re.search(r'RELEASE_COMMIT\}" =~ (\^.*\$) \]\]', validation) self.assertIsNotNone(tag_pattern) self.assertIsNotNone(commit_pattern) assert tag_pattern is not None and commit_pattern is not None for valid_tag in ("v0.0.0", "v1.2.1", "v10.20.30"): self.assertRegex(valid_tag, tag_pattern.group(1)) for invalid_tag in ("1.2.1", "v01.2.1", "v1.2.1-rc.1", "v1.2.1+build", "master"): self.assertNotRegex(invalid_tag, tag_pattern.group(1)) self.assertRegex("0" * 40, commit_pattern.group(1)) for invalid_commit in ("0" * 39, "A" * 40, "g" * 40, "master"): self.assertNotRegex(invalid_commit, commit_pattern.group(1)) def test_checkout_is_pinned_to_the_validated_commit_with_full_history(self) -> None: checkout = _named_step(self.deploy, "Check out the exact release commit with full history") self.assertRegex(checkout, r"actions/checkout@[0-9a-f]{40}") self.assertIn("ref: ${{ inputs.release_commit }}", checkout) self.assertIn("fetch-depth: 0", checkout) self.assertIn("persist-credentials: true", checkout) def test_release_provenance_is_verified_and_exported(self) -> None: provenance = _named_step(self.deploy, "Verify stable release provenance") required_fragments = ( "git fetch --force --no-recurse-submodules origin", 'git cat-file -t "${tag_ref}"', 'git rev-parse "${tag_ref}^{tag}"', 'git rev-parse "${tag_ref}^{commit}"', "git rev-parse HEAD", "< VERSION", "git merge-base --is-ancestor", "refs/remotes/origin/master", "DEPLOY_COMMIT=%s", '>> "${GITHUB_ENV}"', ) for fragment in required_fragments: with self.subTest(fragment=fragment): self.assertIn(fragment, provenance) self.assertNotIn("GITHUB_SHA", self.deploy) def test_all_production_shell_blocks_parse(self) -> None: names = ( "Validate immutable dispatch inputs", "Verify stable release provenance", "Verification-only result", "Deploy verified stable release to production", ) for name in names: script = _run_script(_named_step(self.deploy, name)) result = subprocess.run( ["bash", "-n"], input=script.encode("utf-8"), capture_output=True, check=False, ) with self.subTest(step=name): self.assertEqual(result.returncode, 0, result.stderr.decode("utf-8", "replace")) def test_deployment_requires_the_explicit_action_and_verified_commit(self) -> None: deploy = _named_step(self.deploy, "Deploy verified stable release to production") self.assertIn("inputs.action == 'DEPLOY_STABLE_TO_PRODUCTION'", deploy) self.assertIn('test -n "${DEPLOY_COMMIT:-}"', deploy) self.assertIn('"${GITHUB_REPOSITORY}" "${DEPLOY_COMMIT}"', deploy) self.assertEqual(self.deploy.count("/opt/gitea-deploy/deploy.py deploy"), 1) verify_only = _named_step(self.deploy, "Verification-only result") self.assertIn("inputs.action == 'VERIFY_ONLY'", verify_only) def test_managed_validation_runs_policy_before_profile_work(self) -> None: policy_step = _named_step(self.validation, "Production delivery policy") self.assertIn("python3 -m unittest discover -s .gitea/tests", policy_step) self.assertNotRegex(policy_step, r"(?m)^ if:") self.assertLess( self.validation.index(policy_step), self.validation.index(" - name: Validate the requested profile"), ) def test_security_profile_runs_fail_closed_scanners(self) -> None: step = _named_step(self.validation, "Security — secrets and vulnerable dependencies") self.assertIn("env.PROFILE == 'security'", step) self.assertIn("env.PROFILE == 'full'", step) script = _run_script(step) self.assertIn("gitleaks git .", script) self.assertIn("pip_audit --strict --skip-editable", script) self.assertIn("npm audit --audit-level=high --omit=dev", script) self.assertNotRegex(script, r"(?m)(?:\|\|\s*true|continue-on-error)") toolchain = _named_step(self.validation, "Toolchain") toolchain_script = _run_script(toolchain) self.assertIn("command -v gitleaks", toolchain_script) self.assertIn("exit 1", toolchain_script) def test_compose_validation_renders_the_production_projection(self) -> None: step = _named_step(self.validation, "Compose projections") script = _run_script(step) self.assertIn( "for overlay in backup dr gpu node-agent node-recovery production runtime-worker", script, ) for variable in ( "MODELFORGE_POSTGRES_ADMIN_PASSWORD", "MODELFORGE_MIGRATION_DATABASE_URL", "MODELFORGE_RUNTIME_DATABASE_URL", "MODELFORGE_OPERATOR_API_KEY", "MODELFORGE_BACKUP_ENCRYPTION_KEY", "MODELFORGE_CORS_ORIGINS", "VITE_API_BASE_URL", ): with self.subTest(variable=variable): self.assertIn(f"export {variable}=", script) if __name__ == "__main__": unittest.main()