"""Head-scoped #332 review-decision locks (#620). Proves: * REQUEST_CHANGES on head A does not block mark_ready/submit on head B * APPROVE on head B is allowed after RC on head A (same open PR) * same-head duplicate terminal remains blocked * open-PR cleanup remains forbidden; assessment reports stale_by_head * historical mutations keep their head_sha (preserved ledger) * PR #619-style: old RC at 036b78e…, current head 2429d9d…, approve allowed """ from __future__ import annotations import os import unittest from unittest.mock import patch import mcp_server import stale_review_decision_lock as srdl HEAD_A = "036b78e31ea5d036452b3a52e0e086b01e0c763f" HEAD_B = "2429d9d2e826e519eb8eb4ade45987c00838aefb" HEAD_C = "c" * 40 def _lock(mutations=None, correction=False, ready_pr=None, ready_head=None, ready_action=None): return { "task": "review_pr", "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "session_pid": os.getpid(), "session_profile": "prgs-reviewer", "session_profile_lock": "prgs-reviewer", "profile_identity": "prgs-reviewer", "final_review_decision_ready": False, "ready_pr_number": ready_pr, "ready_action": ready_action, "ready_expected_head_sha": ready_head, "ready_remote": "prgs" if ready_pr else None, "ready_org": "Scaled-Tech-Consulting" if ready_pr else None, "ready_repo": "Gitea-Tools" if ready_pr else None, "live_mutations": list(mutations or []), "correction_authorized": correction, "correction_reason": None, } RC_619_LEGACY = { "pr_number": 619, "action": "request_changes", "review_id": 410, "review_state": "request_changes", # no head_sha — pre-#620 durable ledger shape } RC_619_HEAD_A = { "pr_number": 619, "action": "request_changes", "review_id": 410, "review_state": "request_changes", "head_sha": HEAD_A, } APPROVE_619_HEAD_B = { "pr_number": 619, "action": "approve", "review_id": 411, "review_state": "approve", "head_sha": HEAD_B, } def _seed(mutations=None, **kwargs): import review_workflow_load review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT) mcp_server._save_review_decision_lock(_lock(mutations, **kwargs)) mcp_server.gitea_load_review_workflow() def _open_pr(pr_number=619, head=HEAD_B): return { "number": pr_number, "state": "open", "merged": False, "merged_at": None, "head": {"sha": head}, } def _no_lease(): return {"block": False, "reasons": [], "mutation_allowed": True} def _feedback(blocking=False, stale=False, success=True): return { "success": success, "has_blocking_change_requests": blocking, "review_feedback_stale": stale, "current_head_sha": HEAD_B, } class TestPureHeadScopeHelpers(unittest.TestCase): def test_mutation_head_prefers_recorded_sha(self): m = {"pr_number": 1, "head_sha": HEAD_A} self.assertEqual(srdl.mutation_head_sha(m), HEAD_A) def test_legacy_mutation_uses_ready_expected_for_same_pr(self): lock = _lock( [RC_619_LEGACY], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) self.assertEqual(srdl.mutation_head_sha(RC_619_LEGACY, lock), HEAD_A) def test_legacy_mutation_ignores_ready_for_other_pr(self): lock = _lock([RC_619_LEGACY], ready_pr=1, ready_head=HEAD_A) self.assertIsNone(srdl.mutation_head_sha(RC_619_LEGACY, lock)) def test_prior_blocks_same_head_not_other_head(self): lock = _lock([RC_619_HEAD_A]) self.assertTrue( srdl.prior_live_mutations_block_boundary( lock, pr_number=619, expected_head_sha=HEAD_A ) ) self.assertFalse( srdl.prior_live_mutations_block_boundary( lock, pr_number=619, expected_head_sha=HEAD_B ) ) def test_backfill_stamps_legacy_terminals(self): lock = _lock( [dict(RC_619_LEGACY)], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) srdl.backfill_terminal_heads_from_ready(lock) self.assertEqual(lock["live_mutations"][0]["head_sha"], HEAD_A) class TestHardStopHeadScope(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def test_rc_head_a_allows_mark_ready_head_b(self): _seed( [RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons( 619, "mark_ready", expected_head_sha=HEAD_B ), [], ) def test_rc_head_a_blocks_mark_ready_same_head(self): _seed( [RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) reasons = mcp_server.terminal_review_hard_stop_reasons( 619, "mark_ready", expected_head_sha=HEAD_A ) self.assertTrue(reasons) self.assertIn("#332", reasons[0]) def test_rc_head_a_blocks_other_pr(self): _seed([RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A) reasons = mcp_server.terminal_review_hard_stop_reasons( 700, "mark_ready", expected_head_sha=HEAD_B ) self.assertTrue(reasons) def test_legacy_619_ledger_allows_new_head(self): """PR #619-style durable lock without mutation head_sha.""" _seed( [RC_619_LEGACY], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) self.assertEqual( mcp_server.terminal_review_hard_stop_reasons( 619, "mark_ready", expected_head_sha=HEAD_B ), [], ) class TestMarkFinalAndRecord(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def _mark(self, pr, action, head, feedback=None): 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_get_pr_review_feedback", return_value=feedback or _feedback(blocking=False, stale=True), ), patch.object( mcp_server, "gitea_check_pr_eligibility", return_value={"eligible": True, "head_sha": head}, ): return mcp_server.gitea_mark_final_review_decision( pr_number=pr, action=action, expected_head_sha=head, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) def test_rc_then_rc_on_new_head(self): _seed( [RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) # Prior RC at head A is unresolved but head moved → feedback stale. res = self._mark( 619, "request_changes", HEAD_B, feedback=_feedback(blocking=True, stale=True), ) self.assertTrue(res.get("marked_ready"), res) lock = mcp_server._load_review_decision_lock() # Historical head A preserved on mutation after backfill. heads = { m.get("head_sha") for m in lock["live_mutations"] if m.get("action") == "request_changes" } self.assertIn(HEAD_A, heads) self.assertEqual(lock["ready_expected_head_sha"], HEAD_B) def test_rc_then_approve_on_new_head(self): _seed( [RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) res = self._mark(619, "approve", HEAD_B) self.assertTrue(res.get("marked_ready"), res) self.assertTrue(res.get("head_scoped")) def test_same_head_rc_still_blocked_by_hard_stop(self): _seed( [RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) res = self._mark(619, "approve", HEAD_A) self.assertFalse(res.get("marked_ready")) self.assertTrue(any("#332" in r for r in res.get("reasons") or [])) def test_pr619_style_legacy_lock_approve_on_current_head(self): _seed( [RC_619_LEGACY], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) res = self._mark(619, "approve", HEAD_B) self.assertTrue(res.get("marked_ready"), res) lock = mcp_server._load_review_decision_lock() # Backfill should have stamped HEAD_A onto the legacy mutation. self.assertEqual(lock["live_mutations"][0].get("head_sha"), HEAD_A) self.assertEqual(lock["ready_expected_head_sha"], HEAD_B) def test_record_live_mutation_stores_head(self): _seed(ready_pr=619, ready_head=HEAD_B, ready_action="approve") lock = mcp_server._load_review_decision_lock() lock["final_review_decision_ready"] = True lock["ready_pr_number"] = 619 lock["ready_expected_head_sha"] = HEAD_B mcp_server._save_review_decision_lock(lock) mcp_server.record_live_review_mutation(619, "approve", review_id=99) stored = mcp_server._load_review_decision_lock()["live_mutations"][-1] self.assertEqual(stored["head_sha"], HEAD_B) self.assertEqual(stored["pr_number"], 619) def test_submit_gate_allows_new_head_after_prior_terminal(self): """check_review_decision_gate must not block different-head submit.""" mutations = [RC_619_HEAD_A] _seed( mutations, ready_pr=619, ready_head=HEAD_B, ready_action="approve", ) lock = mcp_server._load_review_decision_lock() lock["final_review_decision_ready"] = True lock["ready_pr_number"] = 619 lock["ready_action"] = "approve" lock["ready_expected_head_sha"] = HEAD_B lock["ready_remote"] = "prgs" lock["ready_org"] = "Scaled-Tech-Consulting" lock["ready_repo"] = "Gitea-Tools" mcp_server._save_review_decision_lock(lock) reasons = mcp_server.check_review_decision_gate( 619, "approve", final_review_decision_ready=True, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertEqual(reasons, [], reasons) def test_submit_gate_blocks_same_head_duplicate(self): mutations = [RC_619_HEAD_A] _seed( mutations, ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) lock = mcp_server._load_review_decision_lock() lock["final_review_decision_ready"] = True lock["ready_pr_number"] = 619 lock["ready_action"] = "request_changes" lock["ready_expected_head_sha"] = HEAD_A lock["ready_remote"] = "prgs" lock["ready_org"] = "Scaled-Tech-Consulting" lock["ready_repo"] = "Gitea-Tools" mcp_server._save_review_decision_lock(lock) reasons = mcp_server.check_review_decision_gate( 619, "request_changes", final_review_decision_ready=True, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertTrue(reasons) class TestAssessmentOpenPrHead(unittest.TestCase): def test_open_pr_stale_by_head_no_cleanup(self): lock = _lock( [RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A, ready_action="request_changes", ) a = srdl.assess_stale_review_decision_lock( lock, pr_live=_open_pr(619, HEAD_B) ) self.assertTrue(a["has_lock"]) self.assertFalse(a["is_moot"]) self.assertFalse(a["cleanup_allowed"]) self.assertTrue(a["stale_by_head"]) self.assertTrue(a["fresh_review_on_current_head_allowed"]) self.assertEqual(a["locked_head_sha"], HEAD_A) self.assertEqual(a["current_pr_head_sha"], HEAD_B) def test_open_pr_same_head_no_fresh(self): lock = _lock([RC_619_HEAD_A], ready_pr=619, ready_head=HEAD_A) a = srdl.assess_stale_review_decision_lock( lock, pr_live=_open_pr(619, HEAD_A) ) self.assertFalse(a["stale_by_head"]) self.assertFalse(a["fresh_review_on_current_head_allowed"]) self.assertFalse(a["cleanup_allowed"]) def test_historical_mutation_preserved_in_summary(self): lock = _lock( [RC_619_HEAD_A, APPROVE_619_HEAD_B], ready_pr=619, ready_head=HEAD_B, ready_action="approve", ) summary = srdl.lock_summary(lock) self.assertEqual(summary["live_mutations_count"], 2) self.assertEqual(summary["last_terminal"]["head_sha"], HEAD_B) self.assertEqual(summary["locked_head_sha"], HEAD_B) if __name__ == "__main__": unittest.main()