diff --git a/tests/test_audit.py b/tests/test_audit.py index adb9242..11cecd3 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -286,6 +286,21 @@ class TestSimpleToolAudit(_AuditWiringBase): class TestGatedToolAudit(_AuditWiringBase): + def setUp(self): + super().setUp() + from tests.test_mcp_server import _install_owned_reviewer_lease + import reviewer_pr_lease + + self._lease_patch = _install_owned_reviewer_lease(8) + self._lease_patch.start() + self._auth_identity_patch = patch( + "mcp_server._authenticated_username", return_value="reviewer-bot" + ) + self._auth_identity_patch.start() + self.addCleanup(self._auth_identity_patch.stop) + self.addCleanup(self._lease_patch.stop) + self.addCleanup(reviewer_pr_lease.clear_session_lease) + def _pr(self, author, state="open", sha="abc123", mergeable=True): return {"user": {"login": author}, "state": state, "head": {"sha": sha}, "mergeable": mergeable} @@ -344,11 +359,22 @@ class TestGatedToolAudit(_AuditWiringBase): GITEA_ALLOWED_OPERATIONS="read,review,approve") with patch.dict(os.environ, env, clear=True): from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision + from tests.test_mcp_server import _install_owned_reviewer_lease + import reviewer_pr_lease + init_review_decision_lock("prgs", "review_pr") gitea_mark_final_review_decision(8, "approve", remote="prgs") - r = gitea_submit_pr_review(pr_number=8, action="approve", - body="LGTM", remote="prgs", - final_review_decision_ready=True) + lease_patch = _install_owned_reviewer_lease(8) + lease_patch.start() + try: + r = gitea_submit_pr_review( + pr_number=8, action="approve", + body="LGTM", remote="prgs", + final_review_decision_ready=True, + ) + finally: + lease_patch.stop() + reviewer_pr_lease.clear_session_lease() self.assertTrue(r["performed"]) recs = self._records() self.assertEqual(len(recs), 1) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index aabb7e7..d04cf38 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -79,6 +79,64 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"): return [_formal_review(reviewer, "APPROVED", sha=sha)] +_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease" + + +def _reviewer_lease_comment( + pr_number, + *, + session_id=_DEFAULT_LEASE_SESSION, + head_sha="abc123", + reviewer="reviewer-bot", +): + from datetime import datetime, timezone + + import reviewer_pr_lease + + body = reviewer_pr_lease.format_lease_body( + repo="Scaled-Tech-Consulting/Gitea-Tools", + pr_number=pr_number, + issue_number=407, + reviewer_identity=reviewer, + profile="gitea-reviewer", + session_id=session_id, + worktree="branches/review-test", + phase="claimed", + candidate_head=head_sha, + target_branch="master", + target_branch_sha="b" * 40, + last_activity=datetime.now(timezone.utc), + ) + return {"id": 9001, "body": body, "user": {"login": reviewer}} + + +def _install_owned_reviewer_lease( + pr_number, + *, + session_id=_DEFAULT_LEASE_SESSION, + head_sha="abc123", +): + import reviewer_pr_lease + + reviewer_pr_lease.clear_session_lease() + reviewer_pr_lease.record_session_lease({ + "pr_number": pr_number, + "session_id": session_id, + "candidate_head": head_sha, + "target_branch": "master", + }) + return patch( + "mcp_server._fetch_pr_comments", + return_value=[ + _reviewer_lease_comment( + pr_number, + session_id=session_id, + head_sha=head_sha, + ) + ], + ) + + # Issue-write tools are profile-gated (#69). ISSUE_WRITE_ENV = { "GITEA_ALLOWED_OPERATIONS": ( @@ -539,6 +597,19 @@ class TestViewPR(unittest.TestCase): class TestMergePR(unittest.TestCase): """Gated merge workflow (#16). gitea_merge_pr is the only merge path.""" + def setUp(self): + import reviewer_pr_lease + + self._lease_patch = _install_owned_reviewer_lease(8) + self._lease_patch.start() + self._auth_identity_patch = patch( + "mcp_server._authenticated_username", return_value="reviewer-bot" + ) + self._auth_identity_patch.start() + self.addCleanup(self._auth_identity_patch.stop) + self.addCleanup(self._lease_patch.stop) + self.addCleanup(reviewer_pr_lease.clear_session_lease) + def _pr(self, author, state="open", sha="abc123", mergeable=True): return { "user": {"login": author}, @@ -799,9 +870,11 @@ class TestMergePR(unittest.TestCase): pr_number=8, confirmation=self._confirm(8), expected_head_sha="deadbeef", remote="prgs") self.assertFalse(r["performed"]) - self.assertIn( - "expected head SHA does not match current PR head (fail closed)", - r["reasons"]) + self.assertTrue(any( + "expected head SHA does not match current PR head (fail closed)" in reason + or "PR head changed during lease" in reason + for reason in r["reasons"] + )) self._assert_no_merge_call(mock_api) @patch("mcp_server.api_request") @@ -1680,7 +1753,20 @@ class TestReviewDecisionValidationGate(unittest.TestCase): } def setUp(self): + import reviewer_pr_lease + init_review_decision_lock("prgs", "review_pr") + self._lease_patch = _install_owned_reviewer_lease( + self.PR, head_sha=self.SHA, + ) + self._lease_patch.start() + self._auth_identity_patch = patch( + "mcp_server._authenticated_username", return_value="reviewer-bot" + ) + self._auth_identity_patch.start() + self.addCleanup(self._auth_identity_patch.stop) + self.addCleanup(self._lease_patch.stop) + self.addCleanup(reviewer_pr_lease.clear_session_lease) def _env(self): return patch.dict(os.environ, { @@ -1775,8 +1861,19 @@ class TestSubmitPrReview(unittest.TestCase): """Gated review-mutation tool (#15).""" def setUp(self): + import reviewer_pr_lease + init_review_decision_lock("prgs", "review_pr") gitea_mark_final_review_decision(8, "approve", remote="prgs") + self._lease_patch = _install_owned_reviewer_lease(8) + self._lease_patch.start() + self._auth_identity_patch = patch( + "mcp_server._authenticated_username", return_value="reviewer-bot" + ) + self._auth_identity_patch.start() + self.addCleanup(self._auth_identity_patch.stop) + self.addCleanup(self._lease_patch.stop) + self.addCleanup(reviewer_pr_lease.clear_session_lease) def _pr(self, author, state="open", sha="abc123", mergeable=True): return { @@ -2014,9 +2111,11 @@ class TestSubmitPrReview(unittest.TestCase): final_review_decision_ready=True, ) self.assertFalse(r["performed"]) - self.assertIn( - "expected head SHA does not match current PR head (fail closed)", - r["reasons"]) + self.assertTrue(any( + "expected head SHA does not match current PR head (fail closed)" in reason + or "PR head changed during lease" in reason + for reason in r["reasons"] + )) self._assert_no_mutation(mock_api) def test_head_sha_match_allows(self): @@ -2079,9 +2178,9 @@ class TestSubmitPrReview(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} with patch.dict(os.environ, env, clear=True): - gitea_mark_final_review_decision(5, "approve", remote="prgs") + gitea_mark_final_review_decision(8, "approve", remote="prgs") r = gitea_submit_pr_review( - pr_number=5, action="approve", remote="prgs", + pr_number=8, action="approve", remote="prgs", final_review_decision_ready=True, ) self.assertFalse(r["performed"]) @@ -2300,6 +2399,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase): self.assertEqual(res["cleanup_status"].get(1), "not present") def test_merge_pr_with_closes_removes_label(self): + import reviewer_pr_lease + + lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123") + lease_patch.start() + self.addCleanup(lease_patch.stop) + self.addCleanup(reviewer_pr_lease.clear_session_lease) + def api_side_effect(method, url, auth, payload=None): if method == "GET" and "/user" in url: return {"login": "merger"} @@ -2334,6 +2440,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase): self.assertEqual(res["cleanup_status"].get(123), "released") def test_merge_pr_with_branch_name_removes_label(self): + import reviewer_pr_lease + + lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123") + lease_patch.start() + self.addCleanup(lease_patch.stop) + self.addCleanup(reviewer_pr_lease.clear_session_lease) + def api_side_effect(method, url, auth, payload=None): if method == "GET" and "/user" in url: return {"login": "merger"} @@ -3015,10 +3128,12 @@ class TestVerifyMutationAuthority(unittest.TestCase): # profile; the active profile resolves as reviewer — side-channel # override rejected even with a matching in-process authority. self._authority() - with patch.dict(os.environ, - {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}): - with self.assertRaises(RuntimeError) as ctx: - mcp_server.verify_mutation_authority("prgs") + with patch("mcp_server.gitea_config.is_runtime_switching_enabled", + return_value=False): + with patch.dict(os.environ, + {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}): + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_mutation_authority("prgs") self.assertIn("side-channel override rejected", str(ctx.exception)) def test_foreign_pid_authority_is_not_trusted(self): diff --git a/tests/test_pr_queue_inventory.py b/tests/test_pr_queue_inventory.py index aeb8bc4..1e5a3c5 100644 --- a/tests/test_pr_queue_inventory.py +++ b/tests/test_pr_queue_inventory.py @@ -156,9 +156,22 @@ class TestPRQueueInventory(unittest.TestCase): ] from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision - init_review_decision_lock("prgs", "review_pr") - gitea_mark_final_review_decision(1, "approve", remote="prgs") - result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True) + from tests.test_mcp_server import _install_owned_reviewer_lease + import reviewer_pr_lease + + with patch("mcp_server._authenticated_username", return_value="reviewer1"): + init_review_decision_lock("prgs", "review_pr") + gitea_mark_final_review_decision(1, "approve", remote="prgs") + lease_patch = _install_owned_reviewer_lease( + 1, head_sha="abc1", session_id="inventory-review-lease", + ) + lease_patch.start() + self.addCleanup(lease_patch.stop) + self.addCleanup(reviewer_pr_lease.clear_session_lease) + result = gitea_review_pr( + pr_number=1, event="APPROVE", remote="prgs", + final_review_decision_ready=True, + ) self.assertTrue(result["success"]) self.assertIn("=== PR Queue Inventory ===", result["message"]) self.assertIn("Repository:", result["message"])