Add attended Policy Nexus source bootstrap

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a058f3-8ba0-7692-a042-9a870fc3d663

tests/test_policy_nexus_forgejo_source_provision.py

@@ -0,0 +1,85 @@
+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_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_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 == "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()