feat(preflight): block workspace edits before identity/capability verification (Closes #210)
This commit is contained in:
@@ -2813,3 +2813,87 @@ class TestIssueLocking(unittest.TestCase):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pre-flight ordering and workspace edit block (#210)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestPreflightVerification(unittest.TestCase):
|
||||
"""Assert that pre-flight verification gates order correctly."""
|
||||
|
||||
def setUp(self):
|
||||
# Save real global variables
|
||||
import mcp_server
|
||||
self.orig_whoami_called = mcp_server._preflight_whoami_called
|
||||
self.orig_capability_called = 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
|
||||
|
||||
# Reset state for each test
|
||||
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
|
||||
|
||||
def tearDown(self):
|
||||
# Restore real global variables
|
||||
import mcp_server
|
||||
mcp_server._preflight_whoami_called = self.orig_whoami_called
|
||||
mcp_server._preflight_capability_called = self.orig_capability_called
|
||||
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
|
||||
if "GITEA_TEST_FORCE_DIRTY" in os.environ:
|
||||
del os.environ["GITEA_TEST_FORCE_DIRTY"]
|
||||
|
||||
def test_preflight_whoami_violation(self):
|
||||
import mcp_server
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server._preflight_capability_called = True
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertTrue(mcp_server._preflight_whoami_violation)
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
|
||||
|
||||
def test_preflight_capability_violation(self):
|
||||
import mcp_server
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server._preflight_whoami_called = True
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
self.assertTrue(mcp_server._preflight_capability_violation)
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
|
||||
|
||||
def test_preflight_not_called_fails_closed(self):
|
||||
import mcp_server
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
|
||||
|
||||
mcp_server._preflight_whoami_called = True
|
||||
with self.assertRaises(RuntimeError) as ctx2:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))
|
||||
|
||||
def test_preflight_reviewer_mutation_violation(self):
|
||||
import mcp_server
|
||||
mcp_server._preflight_whoami_called = True
|
||||
mcp_server._preflight_capability_called = True
|
||||
mcp_server._preflight_resolved_role = "reviewer"
|
||||
|
||||
# When dirty, reviewer edits are blocked
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
|
||||
|
||||
# When clean, reviewer is allowed
|
||||
del os.environ["GITEA_TEST_FORCE_DIRTY"]
|
||||
mcp_server.verify_preflight_purity()
|
||||
|
||||
@@ -789,6 +789,7 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: 700 passed, 6 skipped",
|
||||
"- Mutations: one PR opened",
|
||||
"- Workspace mutations: none",
|
||||
"- Current status: PR open",
|
||||
"- Blockers: none",
|
||||
"- Next: review PR #999",
|
||||
@@ -909,6 +910,29 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
self.assertIn("assess_controller_handoff", skill)
|
||||
self.assertIn("issue #182", skill)
|
||||
|
||||
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
|
||||
# 1. Workspace mutations: none is rejected when local_edits is True
|
||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #196",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #203",
|
||||
"- No review/merge: confirmed",
|
||||
])
|
||||
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
|
||||
self.assertEqual(res["verdict"], "incomplete")
|
||||
self.assertIn("Workspace mutations", res["missing_fields"])
|
||||
|
||||
# 2. Workspace mutations: edited files is allowed when local_edits is True
|
||||
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
|
||||
"- Selected issue: #196",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #203",
|
||||
"- No review/merge: confirmed",
|
||||
])
|
||||
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
|
||||
self.assertEqual(res2["verdict"], "complete")
|
||||
|
||||
|
||||
|
||||
class TestReviewMutationFinalReport(unittest.TestCase):
|
||||
"""Final reports must list exactly one live review mutation."""
|
||||
|
||||
Reference in New Issue
Block a user