"""Tests for the session hard-stop after a terminal review mutation (#332/#693). Extends the single-terminal-decision machinery (#211): after a live REQUEST_CHANGES on a PR head the same-head path must stop; after an APPROVE only the merge sequence for that same PR may continue; #693 allows a *different* PR to be mark_ready/review in the same durable lock (cross-PR isolation); duplicate REQUEST_CHANGES at an unchanged head is suppressed. """ import os import unittest from unittest.mock import patch import mcp_server HEAD_SHA = "a" * 40 def _lock(mutations=None, correction=False, correction_pr=None): mutations = list(mutations or []) if correction and correction_pr is None and mutations: correction_pr = mutations[-1].get("pr_number") return { "task": "review_pr", "remote": "prgs", "session_pid": os.getpid(), "session_profile": "prgs-reviewer", "session_profile_lock": "", "final_review_decision_ready": False, "ready_pr_number": None, "ready_action": None, "ready_expected_head_sha": None, "ready_remote": None, "ready_org": None, "ready_repo": None, "live_mutations": mutations, "correction_authorized": correction, "correction_reason": "test" if correction else None, "correction_pr_number": correction_pr if correction else None, "correction_head_sha": None, } def _seed(mutations=None, correction=False): import review_workflow_load review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT) mcp_server._save_review_decision_lock(_lock(mutations, correction)) mcp_server.gitea_load_review_workflow() APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1, "review_state": "approve"} RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2, "review_state": "request_changes"} class TestTerminalHardStopReasons(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def test_no_lock_no_reasons(self): mcp_server._save_review_decision_lock(None) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(5, "merge"), []) def test_no_terminal_mutation_no_reasons(self): _seed() self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(5, "merge"), []) def test_request_changes_blocks_same_pr_allows_other_pr_review(self): _seed([RC_A]) # Same PR: still hard-stopped for mark_ready/review/merge. for op in ("merge", "mark_ready", "review"): reasons = mcp_server.terminal_review_hard_stop_reasons(5, op) self.assertTrue(reasons, f"{op}/5") self.assertIn("request_changes on PR #5", reasons[0]) self.assertIn("#332", reasons[0]) # #693: different PR may mark_ready / review (not merge via hard-stop alone). self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"), []) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(6, "review"), []) # Merge of an unapproved other PR remains blocked by hard-stop path # (last terminal is not approve of PR 6). self.assertTrue( mcp_server.terminal_review_hard_stop_reasons(6, "merge")) def test_approve_allows_same_pr_merge_and_other_pr_review(self): _seed([APPROVED_A]) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(5, "merge"), []) self.assertTrue( mcp_server.terminal_review_hard_stop_reasons(6, "merge")) # #693 cross-PR isolation: other PR formal review path is open. self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"), []) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(6, "review"), []) def test_correction_reopens_review_path_only(self): _seed([RC_A], correction=True) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), []) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(5, "review"), []) # scoped correction never opens a cross-PR merge self.assertTrue( mcp_server.terminal_review_hard_stop_reasons(6, "merge")) # unscoped/cross-PR correction must not lift PR 6 via correction alone # (PR 6 is allowed by isolation, not by correction scope) class TestMergeHardStopWiring(unittest.TestCase): def tearDown(self): import review_workflow_load review_workflow_load.clear_review_workflow_load() mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def test_merge_blocked_after_request_changes(self): _seed([RC_A]) result = mcp_server.gitea_merge_pr( pr_number=6, confirmation="MERGE PR 6", remote="prgs") self.assertFalse(result["performed"]) self.assertTrue( any("#332" in r for r in result["reasons"]), result["reasons"]) def test_merge_of_other_pr_blocked_after_approve(self): _seed([APPROVED_A]) result = mcp_server.gitea_merge_pr( pr_number=6, confirmation="MERGE PR 6", remote="prgs") self.assertFalse(result["performed"]) self.assertTrue( any("#332" in r for r in result["reasons"]), result["reasons"]) class TestMarkFinalHardStopWiring(unittest.TestCase): def tearDown(self): import review_workflow_load review_workflow_load.clear_review_workflow_load() mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def test_mark_ready_same_pr_blocked_after_terminal_mutation(self): _seed([RC_A]) result = mcp_server.gitea_mark_final_review_decision( pr_number=5, action="approve", remote="prgs", expected_head_sha=HEAD_SHA) self.assertFalse(result["marked_ready"]) self.assertTrue( any("#332" in r for r in result["reasons"]), result["reasons"]) def test_mark_ready_other_pr_allowed_after_foreign_terminal(self): """#693: foreign open-PR terminal must not block mark_final on PR B.""" _seed([RC_A]) no_lease = {"block": False, "reasons": [], "mutation_allowed": True} with patch("mcp_server._list_pr_lease_comments", return_value=[]), \ patch("mcp_server._pr_work_lease_reviewer_block", return_value=no_lease), \ patch.object(mcp_server, "gitea_check_pr_eligibility", return_value={"head_sha": HEAD_SHA, "eligible": True}): result = mcp_server.gitea_mark_final_review_decision( pr_number=6, action="approve", remote="prgs", expected_head_sha=HEAD_SHA) self.assertTrue(result["marked_ready"], result.get("reasons")) def _feedback(blocking, stale=False, success=True): return { "success": success, "has_blocking_change_requests": blocking, "review_feedback_stale": stale, "current_head_sha": HEAD_SHA, } def _mark(action, pr_number=6, **kwargs): kwargs.setdefault("expected_head_sha", HEAD_SHA) no_lease_block = {"block": False, "reasons": [], "mutation_allowed": True} with patch("mcp_server._list_pr_lease_comments", return_value=[]), \ patch("mcp_server._pr_work_lease_reviewer_block", return_value=no_lease_block): return mcp_server.gitea_mark_final_review_decision( pr_number=pr_number, action=action, remote="prgs", **kwargs ) class TestDuplicateRequestChangesSuppression(unittest.TestCase): def tearDown(self): import review_workflow_load review_workflow_load.clear_review_workflow_load() mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def test_duplicate_request_changes_blocked_at_same_head(self): _seed() with patch.object(mcp_server, "gitea_get_pr_review_feedback", return_value=_feedback(blocking=True, stale=False)): result = _mark("request_changes") self.assertFalse(result["marked_ready"]) self.assertTrue( any("duplicate" in r for r in result["reasons"]), result["reasons"]) def test_request_changes_allowed_when_prior_blocker_stale(self): _seed() with patch.object(mcp_server, "gitea_get_pr_review_feedback", return_value=_feedback(blocking=True, stale=True)): result = _mark("request_changes") self.assertTrue(result["marked_ready"], result.get("reasons")) def test_request_changes_allowed_when_no_blocker(self): _seed() with patch.object(mcp_server, "gitea_get_pr_review_feedback", return_value=_feedback(blocking=False)): result = _mark("request_changes") self.assertTrue(result["marked_ready"], result.get("reasons")) def test_request_changes_fails_closed_when_feedback_unavailable(self): _seed() with patch.object(mcp_server, "gitea_get_pr_review_feedback", return_value=_feedback(blocking=False, success=False)): result = _mark("request_changes") self.assertFalse(result["marked_ready"]) self.assertTrue( any("could not verify" in r for r in result["reasons"]), result["reasons"]) def test_approve_does_not_fetch_feedback(self): _seed() with patch.object(mcp_server, "gitea_get_pr_review_feedback", side_effect=AssertionError("must not be called")): result = _mark("approve") self.assertTrue(result["marked_ready"], result.get("reasons")) if __name__ == "__main__": unittest.main()