"""Stale #332 review-decision lock cleanup (#594). Proves: a. active terminal locks still block unrelated terminal reviews b. merged/closed PR locks can be cleared canonically c. cleanup cannot bypass review gates on open PRs d. after moot cleanup, mark_ready for a different PR can proceed (PR #592-style after PR #586-style stale approve) """ from __future__ import annotations import os import unittest from unittest.mock import patch import mcp_server import stale_review_decision_lock as srdl from mcp_server import ( gitea_cleanup_stale_review_decision_lock, gitea_mark_final_review_decision, ) HEAD = "a" * 40 FAKE_AUTH = "Basic dGVzdDp0ZXN0" REVIEWER_ENV = { "GITEA_PROFILE_NAME": "prgs-reviewer", "GITEA_ALLOWED_OPERATIONS": ( "gitea.read,gitea.pr.review,gitea.pr.approve," "gitea.pr.request_changes,gitea.pr.comment" ), "GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer", } REVIEWER_PROFILE = { "profile_name": "prgs-reviewer", "allowed_operations": [ "gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.request_changes", "gitea.pr.comment", ], "forbidden_operations": [ "gitea.pr.merge", "gitea.pr.create", "gitea.branch.push", ], } def _reviewer_profile_patch(): return patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE) def _lock(mutations=None, correction=False): 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": None, "ready_action": None, "ready_expected_head_sha": None, "ready_remote": None, "ready_org": None, "ready_repo": None, "live_mutations": list(mutations or []), "correction_authorized": correction, "correction_reason": None, } APPROVED_586 = { "pr_number": 586, "action": "approve", "review_id": 395, "review_state": "approve", } APPROVED_OPEN = { "pr_number": 700, "action": "approve", "review_id": 1, "review_state": "approve", } RC_OPEN = { "pr_number": 701, "action": "request_changes", "review_id": 2, "review_state": "request_changes", } 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() def _merged_pr(pr_number=586, sha="6913ac9" + "0" * 33): return { "number": pr_number, "state": "closed", "merged": True, "merged_at": "2026-07-09T18:20:03Z", "merge_commit_sha": sha, } def _open_pr(pr_number=700): return { "number": pr_number, "state": "open", "merged": False, "merged_at": None, "merge_commit_sha": None, } # --------------------------------------------------------------------------- # # Pure assessment # --------------------------------------------------------------------------- # class TestAssessStaleLock(unittest.TestCase): def test_no_lock(self): a = srdl.assess_stale_review_decision_lock(None) self.assertFalse(a["cleanup_allowed"]) self.assertFalse(a["is_moot"]) def test_no_terminal_mutations(self): a = srdl.assess_stale_review_decision_lock(_lock([])) self.assertFalse(a["cleanup_allowed"]) def test_open_pr_not_cleanable(self): a = srdl.assess_stale_review_decision_lock( _lock([APPROVED_OPEN]), pr_live=_open_pr(700) ) self.assertFalse(a["is_moot"]) self.assertFalse(a["cleanup_allowed"]) self.assertTrue(any("open PR" in r for r in a["reasons"])) def test_merged_pr_is_moot_and_cleanable(self): a = srdl.assess_stale_review_decision_lock( _lock([APPROVED_586]), pr_live=_merged_pr(586) ) self.assertTrue(a["is_moot"]) self.assertTrue(a["cleanup_allowed"]) self.assertEqual(a["last_terminal_pr"], 586) def test_closed_unmerged_is_moot(self): a = srdl.assess_stale_review_decision_lock( _lock([RC_OPEN]), pr_live={"number": 701, "state": "closed", "merged": False}, ) self.assertTrue(a["is_moot"]) self.assertTrue(a["cleanup_allowed"]) def test_profile_mismatch_blocks(self): a = srdl.assess_stale_review_decision_lock( _lock([APPROVED_586]), pr_live=_merged_pr(586), active_profile_identity="other-profile", ) self.assertFalse(a["cleanup_allowed"]) self.assertFalse(a["profile_match"]) def test_lookup_error_fail_closed(self): a = srdl.assess_stale_review_decision_lock( _lock([APPROVED_586]), pr_lookup_error="timeout", ) self.assertFalse(a["cleanup_allowed"]) self.assertTrue(any("timeout" in r for r in a["reasons"])) # --------------------------------------------------------------------------- # # Hard-stop still blocks active locks # --------------------------------------------------------------------------- # class TestActiveHardStopPreserved(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def test_open_approve_still_blocks_other_pr_mark_ready(self): _seed([APPROVED_OPEN]) reasons = mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready") self.assertTrue(reasons) self.assertIn("#332", reasons[0]) self.assertIn("gitea_cleanup_stale_review_decision_lock", reasons[0]) def test_request_changes_still_blocks_everything(self): _seed([RC_OPEN]) for op in ("merge", "mark_ready", "review"): self.assertTrue( mcp_server.terminal_review_hard_stop_reasons(592, op), msg=op, ) # --------------------------------------------------------------------------- # # MCP cleanup tool # --------------------------------------------------------------------------- # class TestCleanupTool(unittest.TestCase): def setUp(self): self._env = patch.dict(os.environ, REVIEWER_ENV, clear=False) self._env.start() def tearDown(self): self._env.stop() mcp_server._save_review_decision_lock(None) mcp_server.review_workflow_load.clear_review_workflow_load() def test_read_only_reports_moot_without_clearing(self): _seed([APPROVED_586]) with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \ patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \ patch.object( mcp_server, "_authenticated_username", return_value="sysadmin" ), \ _reviewer_profile_patch(), \ patch.object(mcp_server, "_profile_operation_gate", return_value=[]): r = gitea_cleanup_stale_review_decision_lock( apply=False, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertTrue(r["success"]) self.assertTrue(r["is_moot"]) self.assertTrue(r["cleanup_allowed"]) self.assertFalse(r["cleanup_performed"]) self.assertIsNotNone(mcp_server._load_review_decision_lock()) def test_apply_clears_moot_lock(self): _seed([APPROVED_586]) posts = [] def _api(method, url, auth, payload=None): if method == "GET" and "/pulls/586" in url: return _merged_pr() if method == "POST" and "/comments" in url: posts.append(payload) return {"id": 9999} return {} with patch.object(mcp_server, "api_request", side_effect=_api), \ patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \ patch.object( mcp_server, "_authenticated_username", return_value="sysadmin" ), \ _reviewer_profile_patch(), \ patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \ patch.object( mcp_server, "_audited", side_effect=lambda *a, **k: __import__( "contextlib" ).nullcontext(), ): r = gitea_cleanup_stale_review_decision_lock( apply=True, expected_terminal_pr=586, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertTrue(r["cleanup_performed"], r) self.assertTrue(r["audit"]["applied"]) self.assertIsNone(mcp_server._load_review_decision_lock()) self.assertEqual(r.get("audit_comment_id"), 9999) self.assertTrue(posts) def test_apply_refuses_open_pr(self): _seed([APPROVED_OPEN]) with patch.object(mcp_server, "api_request", return_value=_open_pr(700)), \ patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \ patch.object( mcp_server, "_authenticated_username", return_value="sysadmin" ), \ _reviewer_profile_patch(), \ patch.object(mcp_server, "_profile_operation_gate", return_value=[]): r = gitea_cleanup_stale_review_decision_lock( apply=True, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(r["cleanup_performed"]) self.assertFalse(r["cleanup_allowed"]) self.assertIsNotNone(mcp_server._load_review_decision_lock()) def test_expected_terminal_pr_pin(self): _seed([APPROVED_586]) with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \ patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \ patch.object( mcp_server, "_authenticated_username", return_value="sysadmin" ), \ _reviewer_profile_patch(), \ patch.object(mcp_server, "_profile_operation_gate", return_value=[]): r = gitea_cleanup_stale_review_decision_lock( apply=True, expected_terminal_pr=999, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(r["cleanup_performed"]) self.assertTrue(any("expected_terminal_pr" in x for x in r["reasons"])) def test_after_moot_cleanup_other_pr_mark_ready_unblocked(self): """PR #592-style: after clearing merged #586 lock, new mark_ready works.""" _seed([APPROVED_586]) with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \ patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \ patch.object( mcp_server, "_authenticated_username", return_value="sysadmin" ), \ _reviewer_profile_patch(), \ patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \ patch.object( mcp_server, "_audited", side_effect=lambda *a, **k: __import__( "contextlib" ).nullcontext(), ): cleaned = gitea_cleanup_stale_review_decision_lock( apply=True, expected_terminal_pr=586, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", post_audit_comment=False, ) self.assertTrue(cleaned["cleanup_performed"], cleaned) # Hard-stop gone; re-init clean lock like a fresh review_pr resolve. mcp_server.init_review_decision_lock(remote="prgs", task="review_pr") mcp_server.gitea_load_review_workflow() self.assertEqual( mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"), [], ) no_lease = {"block": False, "reasons": [], "mutation_allowed": True} with patch.object(mcp_server, "_list_pr_lease_comments", return_value=[]), \ patch.object( mcp_server, "_pr_work_lease_reviewer_block", return_value=no_lease ), \ patch.object( mcp_server, "gitea_check_pr_eligibility", return_value={ "eligible": True, "reasons": [], "authenticated_user": "sysadmin", "pr_author": "jcwalker3", "self_author": False, }, ), \ patch.object( mcp_server, "_authenticated_username", return_value="sysadmin" ): marked = gitea_mark_final_review_decision( pr_number=592, action="approve", expected_head_sha=HEAD, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertTrue(marked.get("marked_ready"), marked) if __name__ == "__main__": unittest.main()