railiance-platform/tests/test_policy_nexus_forgejo_source_provision.py

from __future__ import annotations

import importlib.util
from pathlib import Path
import subprocess
import unittest
from unittest import mock


SCRIPT = (
    Path(__file__).resolve().parents[1]
    / "scripts"
    / "provision-policy-nexus-forgejo-source.py"
)
SPEC = importlib.util.spec_from_file_location("policy_nexus_source_provision", SCRIPT)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)


class PolicyNexusForgejoSourceProvisionTests(unittest.TestCase):
    def test_contract_is_exactly_read_repository_and_code_only(self) -> None:
        self.assertEqual(MODULE.SCOPES, ["read:repository"])
        self.assertNotIn("repo.code", MODULE.NON_CODE_UNITS)
        self.assertIn("repo.actions", MODULE.NON_CODE_UNITS)
        self.assertIn("repo.packages", MODULE.NON_CODE_UNITS)

    def test_token_creation_accepts_target_user_basic_auth(self) -> None:
        with mock.patch.object(MODULE.urllib.request, "urlopen") as urlopen:
            response = mock.MagicMock()
            response.status = 201
            response.read.return_value = (
                b'{"id": 1, "sha1": "candidate", "scopes": ["read:repository"]}'
            )
            urlopen.return_value.__enter__.return_value = response
            MODULE.api_request(
                "",
                "POST",
                "/users/policy-nexus-source/tokens",
                payload={"name": "candidate", "scopes": ["read:repository"]},
                expected=(201,),
                authorization="Basic non-production",
            )
        request = urlopen.call_args.args[0]
        self.assertEqual(request.get_header("Authorization"), "Basic non-production")

    def test_broader_returned_scope_is_rejected_before_network_checks(self) -> None:
        with mock.patch.object(MODULE, "api_request") as request:
            with self.assertRaises(MODULE.ProvisionError):
                MODULE.verify_source_token("candidate", ["read:repository", "write:repository"])
        request.assert_not_called()

    def test_diagnostic_redacts_unexpected_exception_details(self) -> None:
        with self.subTest("unexpected"):
            with mock.patch.object(MODULE, "DIAGNOSTIC_PATH") as path:
                MODULE.write_diagnostic("provision", RuntimeError("sensitive detail"))
            payload = path.write_text.call_args.args[0]
            self.assertNotIn("sensitive detail", payload)
            self.assertIn("unexpected internal error", payload)

        with self.subTest("bounded provision error"):
            with mock.patch.object(MODULE, "DIAGNOSTIC_PATH") as path:
                MODULE.write_diagnostic(
                    "provision",
                    MODULE.ProvisionError("PUT /bounded/path returned HTTP 403"),
                )
            payload = path.write_text.call_args.args[0]
            self.assertIn("PUT /bounded/path returned HTTP 403", payload)

    def test_warden_failure_classification_is_allowlisted(self) -> None:
        result = subprocess.CompletedProcess(
            ["warden"],
            2,
            stdout=b"opaque provider material\n",
            stderr=b"fetch failed (exit 2) - check caller auth and the path\n",
        )
        classification = MODULE.classify_warden_failure(result)
        self.assertEqual(classification, "caller-auth")
        self.assertNotIn("opaque provider material", classification)

    def test_warden_failure_summary_redacts_token_like_strings(self) -> None:
        result = subprocess.CompletedProcess(
            ["warden"],
            2,
            stdout=b"route failed for hvs.NONPRODUCTION_SECRET_SENTINEL\n",
            stderr=b"invalid invocation\n",
        )
        summary = MODULE.sanitize_warden_failure(result)
        self.assertIn("invalid invocation", summary)
        self.assertNotIn("NONPRODUCTION_SECRET_SENTINEL", summary)

    def test_actions_failure_revokes_pat_and_removes_new_kv_value(self) -> None:
        calls: list[tuple[str, str]] = []

        def fake_api(token, method, path, **kwargs):
            calls.append((method, path))
            if path == "/users/policy-nexus-source/tokens?limit=100":
                return 200, []
            if path.endswith("/actions/secrets?limit=100"):
                return 200, []
            if method == "PATCH" and path == "/admin/users/policy-nexus-source":
                return 200, {"login": "policy-nexus-source", "restricted": True}
            if method == "POST" and path == "/users/policy-nexus-source/tokens":
                return 201, {
                    "id": 73,
                    "name": "candidate",
                    "sha1": "source-token-value",
                    "scopes": ["read:repository"],
                }
            if method == "PUT" and path.endswith("/actions/secrets/FORGEJO_SOURCE_TOKEN"):
                raise MODULE.ProvisionError("injected Actions failure")
            if method == "DELETE" and path == "/users/policy-nexus-source/tokens/73":
                return 204, None
            raise AssertionError((method, path, kwargs))

        bao_calls: list[tuple[list[str], bytes | None]] = []

        def fake_bao(args, *, stdin=None):
            bao_calls.append((args, stdin))
            if args[:3] == ["kv", "get", "-field=FORGEJO_SOURCE_TOKEN"]:
                if stdin is None and len(bao_calls) == 1:
                    return subprocess.CompletedProcess(args, 2, b"", b"")
                return subprocess.CompletedProcess(args, 0, b"source-token-value\n", b"")
            return subprocess.CompletedProcess(args, 0, b"", b"")

        with (
            mock.patch.object(MODULE, "ensure_user"),
            mock.patch.object(MODULE, "ensure_team", return_value=7),
            mock.patch.object(MODULE, "verify_effective_permission"),
            mock.patch.object(MODULE, "verify_source_token"),
            mock.patch.object(MODULE, "api_request", side_effect=fake_api),
            mock.patch.object(MODULE, "bao", side_effect=fake_bao),
        ):
            with self.assertRaises(MODULE.ProvisionError):
                MODULE.provision("admin-token")

        self.assertIn(("DELETE", "/users/policy-nexus-source/tokens/73"), calls)
        self.assertTrue(any(args == ["kv", "delete", MODULE.BAO_PATH] for args, _ in bao_calls))
        put = next((args, stdin) for args, stdin in bao_calls if args[:2] == ["kv", "put"])
        self.assertEqual(put[1], b"source-token-value")
        self.assertNotIn("source-token-value", " ".join(put[0]))


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