sysadminandClaude Opus 4.8
a8bcbdbcf2
Add preflight_contract helpers with explicit re-resolve error messages, document the read-only tool set and consumption rules, and extend tests for close_pr sequencing and task replacement. Validation: ./venv/bin/python -m pytest tests/test_preflight_read_survival.py tests/test_mcp_server.py::TestPreflightVerification -q Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
159 lines
7.5 KiB
Python
159 lines
7.5 KiB
Python
"""#469/#470: capability preflight lifetime across safe read-only 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")
|
|
|
|
def test_close_pr_sequence_with_interleaved_reads(self):
|
|
"""resolve(close_pr) → whoami/view reads → close_pr gate (#470)."""
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check(
|
|
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
|
)
|
|
# Simulate live-state revalidation between resolve and mutation.
|
|
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_fresh_capability_resolve_replaces_prior_task(self):
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check(
|
|
"capability", resolved_role="author", resolved_task="create_issue"
|
|
)
|
|
mcp_server.record_preflight_check(
|
|
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
|
)
|
|
self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
|
|
mcp_server.verify_preflight_purity(task="close_pr")
|
|
|
|
def test_missing_capability_error_names_re_resolve(self):
|
|
mcp_server.record_preflight_check("whoami")
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
mcp_server.verify_preflight_purity(task="close_pr")
|
|
msg = str(ctx.exception)
|
|
self.assertIn("gitea_resolve_task_capability", msg)
|
|
self.assertIn('task="close_pr"', msg)
|
|
|
|
def test_task_mismatch_error_names_re_resolve(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")
|
|
msg = str(ctx.exception)
|
|
self.assertIn("task mismatch", msg)
|
|
self.assertIn('task="close_pr"', msg)
|
|
|
|
def test_consumed_capability_error_names_re_resolve(self):
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check(
|
|
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
|
)
|
|
mcp_server.verify_preflight_purity(task="close_pr")
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
mcp_server.verify_preflight_purity(task="close_pr")
|
|
self.assertIn('task="close_pr"', str(ctx.exception))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |