"""Fail-closed integration tests for the parentless public source boundary.""" from __future__ import annotations import subprocess import tempfile import unittest from pathlib import Path REPOSITORY_ROOT = Path(__file__).resolve().parents[2] EXPORTER = REPOSITORY_ROOT / "scripts" / "export-public-source.mjs" VALIDATOR = REPOSITORY_ROOT / "scripts" / "validate-public-source.mjs" class PublicSourceExportTests(unittest.TestCase): def _repository( self, root: Path, *, license_present: bool = True, managed_validation_workflow: str | None = None, ) -> Path: repository = root / "private-source" repository.mkdir() files: dict[str, str | bytes] = { "README.md": "# Product\n", "SECURITY.md": "# Security\n", "CONTRIBUTING.md": "# Contributing\n", "docker-compose.yml": "services: {}\n", "VERSION": "1.2.1\n", "backend/pyproject.toml": "[project]\nname = 'fixture-api'\n", "frontend/package.json": '{"name":"fixture-web"}\n', "node-agent/pyproject.toml": "[project]\nname = 'fixture-agent'\n", "runtime-worker/pyproject.toml": "[project]\nname = 'fixture-worker'\n", } if license_present: files["LICENSE"] = (REPOSITORY_ROOT / "LICENSE").read_bytes() if managed_validation_workflow is not None: files[".gitea/workflows/managed-validation.yml"] = managed_validation_workflow for relative_path, contents in files.items(): destination = repository / relative_path destination.parent.mkdir(parents=True, exist_ok=True) if isinstance(contents, bytes): destination.write_bytes(contents) else: destination.write_text(contents, encoding="utf-8") (repository / "public-source.allowlist").write_text( "\n".join(sorted(files)) + "\n", encoding="utf-8" ) subprocess.run(["git", "init", "-q"], cwd=repository, check=True) subprocess.run(["git", "add", "."], cwd=repository, check=True) subprocess.run( [ "git", "-c", "user.name=ModelForge test", "-c", "user.email=test@example.invalid", "commit", "-qm", "fixture", ], cwd=repository, check=True, ) return repository def _export(self, repository: Path, output: Path) -> subprocess.CompletedProcess[str]: return subprocess.run( ["node", str(EXPORTER), "--repository", str(repository), "--output", str(output)], capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) def _validate(self, output: Path) -> subprocess.CompletedProcess[str]: return subprocess.run( ["node", str(VALIDATOR)], cwd=output, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) def test_export_requires_an_explicit_tracked_license(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) repository = self._repository(root, license_present=False) result = self._export(repository, root / "public") self.assertNotEqual(result.returncode, 0) self.assertIn("canonical LICENSE is missing", result.stderr) def test_manifest_detects_content_tampering_and_unexpected_files(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) output = root / "public" result = self._export(self._repository(root), output) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) validation = self._validate(output) self.assertEqual(validation.returncode, 0, validation.stdout + validation.stderr) original_readme = (output / "README.md").read_bytes() (output / "README.md").write_text("tampered\n", encoding="utf-8") tampered = self._validate(output) self.assertNotEqual(tampered.returncode, 0) self.assertIn("Manifest", tampered.stderr) (output / "README.md").write_bytes(original_readme) (output / "not-reviewed.txt").write_text("extra\n", encoding="utf-8") unexpected = self._validate(output) self.assertNotEqual(unexpected.returncode, 0) self.assertIn("Unexpected public source files", unexpected.stderr) def test_dirty_source_tree_cannot_claim_the_head_revision(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) repository = self._repository(root) (repository / "README.md").write_text("changed after commit\n", encoding="utf-8") result = self._export(repository, root / "public") self.assertNotEqual(result.returncode, 0) self.assertIn("commit the exact source tree", result.stderr) def test_sanitized_destination_collisions_fail_before_writing(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) repository = self._repository(root) private_name = "Tow" + "er.txt" public_name = "GPU " + "Node.txt" (repository / private_name).write_text("one\n", encoding="utf-8") (repository / public_name).write_text("two\n", encoding="utf-8") with (repository / "public-source.allowlist").open("a", encoding="utf-8") as allowlist: allowlist.write(f"{private_name}\n{public_name}\n") subprocess.run(["git", "add", "."], cwd=repository, check=True) subprocess.run( [ "git", "-c", "user.name=ModelForge test", "-c", "user.email=test@example.invalid", "commit", "-qm", "collision fixture", ], cwd=repository, check=True, ) result = self._export(repository, root / "public") self.assertNotEqual(result.returncode, 0) self.assertIn("finding", result.stderr) def test_public_managed_validation_requires_explicit_owner_dispatch(self) -> None: workflow = "name: Managed validation\n\non:\n pull_request:\n workflow_dispatch:\n" with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) repository = self._repository( root, managed_validation_workflow=workflow, ) output = root / "public" result = self._export(repository, output) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) canonical = ( repository / ".gitea/workflows/managed-validation.yml" ).read_text(encoding="utf-8") exported = ( output / ".gitea/workflows/managed-validation.yml" ).read_text(encoding="utf-8") self.assertIn(" pull_request:\n", canonical) self.assertNotIn(" pull_request:\n", exported) self.assertIn(" workflow_dispatch:\n", exported) self.assertIn("fork PRs never reach private runners", exported) def test_changed_managed_validation_trigger_fails_closed(self) -> None: workflow = "name: Managed validation\n\non:\n workflow_dispatch:\n" with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) repository = self._repository( root, managed_validation_workflow=workflow, ) result = self._export(repository, root / "public") self.assertNotEqual(result.returncode, 0) self.assertIn("has no expected pull_request trigger", result.stderr) if __name__ == "__main__": unittest.main()