Activate Policy Nexus source credential lane

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

tests/test_policy_nexus_forgejo_source_provision.py

@@ -25,6 +25,25 @@ class PolicyNexusForgejoSourceProvisionTests(unittest.TestCase):
         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):
@@ -48,6 +67,28 @@ class PolicyNexusForgejoSourceProvisionTests(unittest.TestCase):
             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]] = []
 
@@ -57,6 +98,8 @@ class PolicyNexusForgejoSourceProvisionTests(unittest.TestCase):
                 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,