Merge pull request 'fix: preserve capability preflight across safe reads (Closes #469)' (#502) from fix/issue-469-preflight-read-survival into master

2026-07-08 03:01:48 -05:00
parent f89949cea9 3bce0a55fb
commit ce61e424f6
2 changed files with 198 additions and 30 deletions
@@ -0,0 +1,106 @@
+"""#469: capability preflight survives interleaved read-only whoami calls."""
+
+import os
+import unittest
+
+import gitea_mcp_server as mcp_server
+
+
+class TestPreflightReadSurvival(unittest.TestCase):
+    def setUp(self):
+        self.orig_whoami = mcp_server._preflight_whoami_called
+        self.orig_capability = mcp_server._preflight_capability_called
+        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
+        self.orig_capability_violation = mcp_server._preflight_capability_violation
+        self.orig_resolved_role = mcp_server._preflight_resolved_role
+        self.orig_resolved_task = mcp_server._preflight_resolved_task
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_resolved_role = None
+        mcp_server._preflight_resolved_task = None
+        mcp_server._process_start_porcelain = ""
+        mcp_server._preflight_whoami_baseline_porcelain = None
+        mcp_server._preflight_capability_baseline_porcelain = None
+
+    def tearDown(self):
+        mcp_server._preflight_whoami_called = self.orig_whoami
+        mcp_server._preflight_capability_called = self.orig_capability
+        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
+        mcp_server._preflight_capability_violation = self.orig_capability_violation
+        mcp_server._preflight_resolved_role = self.orig_resolved_role
+        mcp_server._preflight_resolved_task = self.orig_resolved_task
+        mcp_server._process_start_porcelain = self.orig_process_start
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+
+    def test_interleaved_whoami_preserves_capability(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reconciler", resolved_task="close_pr"
+        )
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+
+        mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+    def test_missing_capability_still_fails_closed(self):
+        mcp_server.record_preflight_check("whoami")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_task_mismatch_fails_closed(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn("task mismatch", str(ctx.exception))
+
+    def test_capability_consumed_after_mutation_gate(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        mcp_server.verify_preflight_purity(task="create_issue")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="create_issue")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_whoami_recovery_after_violation_clears_capability(self):
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reviewer", resolved_task="review_pr"
+        )
+        mcp_server.verify_preflight_purity(task="review_pr")
+
+
+if __name__ == "__main__":
+    unittest.main()