prj-forgejo-org-refactor/tests/test_access_consumers.py

import sys
import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
from access_consumers import classify, runtime_consumers, scan_repo, target_repo  # noqa: E402


class ClassifyTest(unittest.TestCase):
    def test_kinds(self):
        cases = [
            ("posts/x.yaml", "/coulomb/pr-hall-of-helix/raw/branch/main/a.png", "url: x", "raw-fetch"),
            ("src/client.py", "/api/v1/repos/coulomb/x", "u = ...", "api-client"),
            (".forgejo/workflows/image.yaml", "/coulomb/x", "uses: x", "ci-action"),
            ("go.mod", "/coulomb/x", "require x", "package-consumer"),
            ("k8s/app.yaml", "/coulomb/x.git", "  repoURL: https://...", "gitops-source"),
            ("chart/values.yaml", "/coulomb/x", "  repository: forgejo...", "image-pull"),
            ("README.md", "/coulomb/x", "see", "doc-link"),
            ("docs/evidence/run.json", "/coulomb/x", "x", "record"),
            ("Makefile", "/coulomb/x", "push", "build-tooling"),
            ("pyproject.toml", "/api/packages/coulomb/pypi/simple", "url: x", "package-consumer"),
            ("client.py", "/api/v1/packages/coulomb", "u = ...", "package-consumer"),
        ]
        for f, p, line, want in cases:
            self.assertEqual(classify(f, p, line), want, f)

    def test_target_repo(self):
        self.assertEqual(target_repo("/coulomb/key-cape.git"), "coulomb/key-cape")
        self.assertEqual(target_repo("/api/v1/repos/coulomb/state-hub/issues"), "coulomb/state-hub")
        self.assertEqual(target_repo("/v2/coulomb/flex-auth/manifests/x"), "coulomb/flex-auth")
        self.assertEqual(target_repo("/coulomb/activity-core@sha256:ab"), "coulomb/activity-core")
        self.assertIsNone(target_repo("/user/login"))
        self.assertIsNone(target_repo("/api/packages/coulomb/pypi/simple"))
        self.assertIsNone(target_repo("/api/v1/packages/coulomb"))
        self.assertIsNone(target_repo("/api/v1/orgs/coulomb/repos"))

    def test_missing_pull_secret_does_not_prove_anonymous_access(self):
        rt = {"applications": [], "workloads": [{
            "namespace": "ns", "kind": "Deployment", "name": "api", "image_pull_secret_names": [],
            "images": ["forgejo.coulomb.social/coulomb/x@sha256:00"]}]}
        (c,) = runtime_consumers(rt)
        self.assertEqual(c["class"], "cluster-image-pull")
        self.assertTrue(c["credential"].startswith("unknown"))
        self.assertEqual(c["target"], "coulomb/x")

    def test_scan_excludes_sensitive_and_binary_files_without_values(self):
        with tempfile.TemporaryDirectory() as d:
            repo = Path(d)
            (repo / "client.py").write_text('url = "https://user:hidden@forgejo.coulomb.social/coulomb/valid?token=hidden"\n')
            (repo / ".env").write_text("forgejo.coulomb.social/coulomb/private-value\n")
            (repo / "secret.sops.yaml").write_text("forgejo.coulomb.social/coulomb/encrypted-value\n")
            (repo / "binary").write_bytes(b"\0forgejo.coulomb.social/coulomb/binary-value")
            (repo / "link.py").symlink_to(repo / "client.py")
            os.mkfifo(repo / "pipe")
            diagnostics = []
            hits = scan_repo(repo, diagnostics)
            self.assertEqual([h["target"] for h in hits], ["coulomb/valid"])
            serialized = json.dumps(hits)
            for value in ["hidden", "private-value", "encrypted-value", "binary-value"]:
                self.assertNotIn(value, serialized)
            self.assertEqual({x["reason"] for x in diagnostics},
                             {"sensitive-or-encrypted", "binary-content", "symlink-not-followed", "non-regular-file"})

    def test_cli_is_repeatable_and_keeps_legacy_host_separate(self):
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            repo = root / "reader"
            repo.mkdir()
            subprocess.run(["git", "init", "-q", str(repo)], check=True)
            (repo / "client.py").write_text(
                'a = "https://forgejo.coulomb.social/coulomb/x"\n'
                'b = "https://gitea.coulomb.social/coulomb/x"\n')
            runtime = root / "runtime.json"
            runtime.write_text(json.dumps({"applications": [], "workloads": []}))
            assignments = root / "assignments.csv"
            assignments.write_text("repository,proposed_org\nx,netkingdom\n")
            output = root / "output.json"
            command = [sys.executable, str(Path(__file__).resolve().parents[1] / "tools/access_consumers.py"),
                       "--root", str(root), "--runtime", str(runtime),
                       "--assignments", str(assignments), "--output", str(output)]
            subprocess.run(command, check=True, capture_output=True)
            first = output.read_bytes()
            subprocess.run(command, check=True, capture_output=True)
            self.assertEqual(first, output.read_bytes())
            report = json.loads(first)
            self.assertEqual({c["host"] for c in report["consumers"]},
                             {"forgejo.coulomb.social", "gitea.coulomb.social"})
            self.assertEqual(report["scan_error_count"], 0)

    def test_unreadable_file_is_reported_without_raw_error(self):
        with tempfile.TemporaryDirectory() as d:
            repo = Path(d)
            (repo / "client.py").write_text("placeholder")
            diagnostics = []
            with patch.object(Path, "read_text", side_effect=OSError("private error detail")):
                self.assertEqual(scan_repo(repo, diagnostics), [])
            self.assertEqual(diagnostics, [{"repository": repo.name,
                                           "path": "client.py", "reason": "unreadable-file"}])


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