RedFlag/.publication/test_surface_gate.py

import json
from pathlib import Path
import subprocess
import tempfile
import unittest


HERE = Path(__file__).resolve().parent
GATE = HERE / "surface_gate.py"


class SurfaceGateTests(unittest.TestCase):
    def setUp(self):
        self.work = tempfile.TemporaryDirectory()
        self.repo = Path(self.work.name)
        subprocess.run(["git", "init", "-q", self.repo], check=True)
        subprocess.run(["git", "-C", self.repo, "config", "user.name", "Test"], check=True)
        subprocess.run(
            ["git", "-C", self.repo, "config", "user.email", "test@example.test"],
            check=True,
        )

    def tearDown(self):
        self.work.cleanup()

    def candidate(self, files, admitted=None, exclude=None):
        for name, content in files.items():
            path = self.repo / name
            path.parent.mkdir(parents=True, exist_ok=True)
            if isinstance(content, bytes):
                path.write_bytes(content)
            else:
                path.write_text(content)

        policy_dir = self.repo / ".publication"
        policy_dir.mkdir(exist_ok=True)
        policy_paths = {".publication/paths.txt", ".publication/surface.json"}
        admitted = set(files) if admitted is None else set(admitted)
        admitted.update(policy_paths)
        (policy_dir / "paths.txt").write_text("".join(f"{p}\n" for p in sorted(admitted)))
        policy = {
            "schema_version": 2,
            "repository": "Fimeg/RedFlag",
            "path_manifest": ".publication/paths.txt",
            "exclude": exclude or [],
            "review_required": [],
            "forbidden": [],
            "forbidden_classes": [],
            "max_blob_bytes": 2097152,
            "exceptions": [],
        }
        (policy_dir / "surface.json").write_text(json.dumps(policy, indent=2) + "\n")
        subprocess.run(["git", "-C", self.repo, "add", "."], check=True)
        subprocess.run(
            ["git", "-C", self.repo, "commit", "-q", "-m", "test: make candidate"],
            check=True,
        )
        return subprocess.check_output(
            ["git", "-C", self.repo, "rev-parse", "HEAD"], text=True
        ).strip()

    def run_gate(self, sha, *extra):
        return subprocess.run(
            [
                "python3",
                GATE,
                "--repo",
                self.repo,
                "--sha",
                sha,
                "--manifest",
                self.repo / ".publication/surface.json",
                "--skip-history",
                *extra,
            ],
            capture_output=True,
            text=True,
        )

    def test_explicitly_admitted_file_passes(self):
        sha = self.candidate({"product/main.go": "package main\n"})
        result = self.run_gate(sha)
        self.assertEqual(0, result.returncode, result.stdout + result.stderr)
        self.assertIn("Verdict: PASS", result.stdout)

    def test_unlisted_file_under_product_directory_fails(self):
        sha = self.candidate(
            {"product/main.go": "package main\n", "product/private.txt": "not admitted\n"},
            admitted={"product/main.go"},
        )
        result = self.run_gate(sha)
        self.assertEqual(1, result.returncode)
        self.assertIn("unlisted-path", result.stdout)
        self.assertIn("product/private.txt", result.stdout)

    def test_excluded_file_cannot_appear(self):
        sha = self.candidate(
            {"product/main.go": "package main\n", "private/note.txt": "no\n"},
            exclude=["private/*"],
        )
        result = self.run_gate(sha)
        self.assertEqual(1, result.returncode)
        self.assertIn("excluded-path", result.stdout)
        self.assertIn("private/note.txt", result.stdout)

    def test_missing_manifest_entry_fails_closed(self):
        sha = self.candidate(
            {"product/main.go": "package main\n"},
            admitted={"product/main.go", "product/missing.go"},
        )
        result = self.run_gate(sha)
        self.assertEqual(1, result.returncode)
        self.assertIn("missing-path", result.stdout)
        self.assertIn("product/missing.go", result.stdout)

    def test_approved_candidate_inventory_is_deterministic(self):
        sha = self.candidate({"a.txt": "a\n", "nested/b.txt": "b\n"})
        first = self.repo / "first.manifest"
        second = self.repo / "second.manifest"
        one = self.run_gate(sha, "--inventory-out", first)
        two = self.run_gate(sha, "--inventory-out", second)
        self.assertEqual(0, one.returncode, one.stdout + one.stderr)
        self.assertEqual(0, two.returncode, two.stdout + two.stderr)
        self.assertEqual(first.read_bytes(), second.read_bytes())
        paths = [line.split(" ", 3)[3] for line in first.read_text().splitlines()]
        self.assertEqual(sorted(paths), paths)
        self.assertEqual(
            [".publication/paths.txt", ".publication/surface.json", "a.txt", "nested/b.txt"],
            paths,
        )

    def test_unlisted_binary_path_fails_before_content_scan(self):
        sha = self.candidate(
            {"product/main.go": "package main\n", "product/payload.bin": b"\0\xff\0\xff"},
            admitted={"product/main.go"},
        )
        result = self.run_gate(sha)
        self.assertEqual(1, result.returncode)
        self.assertIn("unlisted-path", result.stdout)
        self.assertIn("product/payload.bin", result.stdout)

    def test_deleted_unlisted_file_in_history_still_fails(self):
        self.candidate({"product/main.go": "package main\n", "private/note.txt": "no\n"})
        (self.repo / "private/note.txt").unlink()
        sha = self.candidate({"product/main.go": "package main\n"})
        result = self.run_gate(sha)
        self.assertEqual(1, result.returncode)
        self.assertIn("unlisted-path", result.stdout)
        self.assertIn("reachable commit", result.stdout)

    def test_deleted_excluded_binary_in_history_still_fails(self):
        self.candidate({"product/main.go": "package main\n", "product/old.png": b"\0\xff"})
        (self.repo / "product/old.png").unlink()
        sha = self.candidate({"product/main.go": "package main\n"}, exclude=["product/old.png"])
        result = self.run_gate(sha)
        self.assertEqual(1, result.returncode)
        self.assertIn("excluded-path", result.stdout)
        self.assertIn("reachable commit", result.stdout)

    def test_edits_to_admitted_files_preserve_valid_history(self):
        self.candidate({"product/main.go": "package main\n"})
        sha = self.candidate({"product/main.go": "package main\n// changed\n"})
        result = self.run_gate(sha)
        self.assertEqual(0, result.returncode, result.stdout + result.stderr)
        self.assertIn("2 reachable commits", result.stdout)


if __name__ == "__main__":
    unittest.main()