"""Post-merge moot reviewer-lease handling (#515). Covers: a. Reviewer-lease acquisition/adoption is refused on an already-merged/closed PR (fail closed, no mutation). b. The post-merge moot cleanup path is safe and idempotent. c. An active foreign lease on an *open* PR is never force-cleaned. """ import os import sys import unittest from datetime import datetime, timezone from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import reviewer_pr_lease as leases # noqa: E402 from mcp_server import ( # noqa: E402 gitea_acquire_reviewer_pr_lease, gitea_cleanup_post_merge_moot_lease, ) FAKE_AUTH = "Basic dGVzdDp0ZXN0" MERGER_ENV = { "GITEA_PROFILE_NAME": "prgs-merger", "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment", } PR = 487 ISSUE = 485 SESSION = "97274-676d20a825c4" def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed", candidate_head="a" * 40): body = leases.format_lease_body( repo="Scaled-Tech-Consulting/Gitea-Tools", pr_number=pr_number, issue_number=ISSUE, reviewer_identity="sysadmin", profile="prgs-reviewer", session_id=session_id, worktree="branches/review-pr487", phase=phase, candidate_head=candidate_head, target_branch="master", target_branch_sha="b" * 40, last_activity=datetime.now(timezone.utc), ) return {"id": 6603, "body": body, "user": {"login": "sysadmin"}} # --------------------------------------------------------------------------- # # Pure logic # --------------------------------------------------------------------------- # class TestAcquireRefusedOnMergedPR(unittest.TestCase): def test_acquire_refused_when_pr_merged_or_closed(self): result = leases.assess_acquire_lease( [_lease_comment()], pr_number=PR, reviewer_identity="sysadmin", profile="prgs-merger", session_id="new-session", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=ISSUE, worktree="branches/merge-pr487", candidate_head="a" * 40, target_branch="master", target_branch_sha="b" * 40, pr_merged_or_closed=True, ) self.assertFalse(result["acquire_allowed"]) self.assertTrue(result["post_merge_moot"]) self.assertIsNone(result["lease_body"]) self.assertTrue(any("post_merge_moot" in r for r in result["reasons"])) def test_acquire_still_allowed_on_open_pr_without_flag(self): result = leases.assess_acquire_lease( [], pr_number=PR, reviewer_identity="sysadmin", profile="prgs-reviewer", session_id="s1", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=ISSUE, worktree="branches/review-pr487", candidate_head="a" * 40, target_branch="master", target_branch_sha="b" * 40, ) self.assertTrue(result["acquire_allowed"]) self.assertFalse(result["post_merge_moot"]) class TestPostMergeMootAssessment(unittest.TestCase): def test_merged_pr_with_active_lease_is_moot_and_cleanable(self): a = leases.assess_post_merge_moot_lease( [_lease_comment()], pr_number=PR, pr_merged=True, pr_state="closed", merge_commit_sha="c" * 40, ) self.assertTrue(a["pr_merged_or_closed"]) self.assertTrue(a["is_moot"]) self.assertTrue(a["cleanup_allowed"]) self.assertIsNotNone(a["release_body"]) self.assertIn("phase: released", a["release_body"]) self.assertIn("blocker: post-merge-moot", a["release_body"]) def test_open_pr_active_lease_never_cleaned(self): a = leases.assess_post_merge_moot_lease( [_lease_comment()], pr_number=PR, pr_merged=False, pr_state="open", ) self.assertFalse(a["pr_merged_or_closed"]) self.assertFalse(a["is_moot"]) self.assertFalse(a["cleanup_allowed"]) self.assertIsNone(a["release_body"]) self.assertTrue(any("still open" in r for r in a["reasons"])) def test_merged_pr_without_lease_nothing_to_clean(self): a = leases.assess_post_merge_moot_lease( [], pr_number=PR, pr_merged=True, pr_state="closed") self.assertTrue(a["pr_merged_or_closed"]) self.assertFalse(a["is_moot"]) self.assertFalse(a["cleanup_allowed"]) self.assertTrue(any("nothing to clean" in r for r in a["reasons"])) def test_cleanup_is_idempotent(self): """After the released marker is posted, a re-assess finds nothing to clean.""" first = leases.assess_post_merge_moot_lease( [_lease_comment()], pr_number=PR, pr_merged=True, pr_state="closed") self.assertTrue(first["cleanup_allowed"]) released_comment = { "id": 7000, "body": first["release_body"], "user": {"login": "sysadmin"}} second = leases.assess_post_merge_moot_lease( [_lease_comment(), released_comment], pr_number=PR, pr_merged=True, pr_state="closed") self.assertFalse(second["is_moot"]) self.assertFalse(second["cleanup_allowed"]) # --------------------------------------------------------------------------- # # Server tools # --------------------------------------------------------------------------- # def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999): """Build an api_request side effect keyed on method + url.""" calls = {"post": []} def _side(method, url, auth=None, payload=None, *a, **k): m = (method or "").upper() if m == "POST": calls["post"].append({"url": url, "payload": payload}) return {"id": posted_id} if "/comments" in url: return list(comments) if "/pulls/" in url: pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40} if pr_merged: pr["merged"] = True pr["merged_at"] = "2026-07-08T07:46:04Z" return pr if "/issues/" in url: return {"state": "closed" if pr_merged else "open", "number": ISSUE} return {} return _side, calls class TestAcquireToolRefusesMergedPR(unittest.TestCase): def setUp(self): leases.clear_session_lease() @patch("mcp_server.verify_preflight_purity", return_value=None) @patch("mcp_server._authenticated_username", return_value="sysadmin") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.api_request") def test_acquire_tool_fails_closed_on_merged_pr_without_posting( self, mock_api, _auth, _user, _purity): side, calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=[]) mock_api.side_effect = side with patch.dict(os.environ, MERGER_ENV, clear=True): result = gitea_acquire_reviewer_pr_lease( pr_number=PR, worktree="branches/merge-pr487", candidate_head="a" * 40, issue_number=ISSUE, remote="prgs", ) self.assertFalse(result["success"]) self.assertFalse(result["acquired"]) self.assertTrue(result.get("post_merge_moot")) self.assertEqual(calls["post"], [], "must not post a lease comment") class TestCleanupTool(unittest.TestCase): def setUp(self): leases.clear_session_lease() @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.api_request") def test_read_only_reports_moot_without_mutating(self, mock_api, _auth): side, calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=[_lease_comment()]) mock_api.side_effect = side with patch.dict(os.environ, MERGER_ENV, clear=True): result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=False, remote="prgs") self.assertTrue(result["success"]) self.assertTrue(result["pr_merged_or_closed"]) self.assertTrue(result["lease_moot"]) self.assertFalse(result["cleanup_performed"]) self.assertTrue(result["no_merge_or_adoption"]) self.assertEqual(result["mode"], "read_only") self.assertEqual(calls["post"], []) @patch("mcp_server.verify_preflight_purity", return_value=None) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.api_request") def test_apply_posts_released_marker_on_merged_pr( self, mock_api, _auth, _purity): side, calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=[_lease_comment()]) mock_api.side_effect = side with patch.dict(os.environ, MERGER_ENV, clear=True): result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=True, remote="prgs") self.assertTrue(result["success"]) self.assertTrue(result["cleanup_performed"]) self.assertEqual(result["released_comment_id"], 9999) self.assertEqual(len(calls["post"]), 1) self.assertIn("phase: released", calls["post"][0]["payload"]["body"]) self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"]) @patch("mcp_server.verify_preflight_purity", return_value=None) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.api_request") def test_apply_refuses_to_clean_open_pr(self, mock_api, _auth, _purity): side, calls = _api_side_effect( pr_state="open", pr_merged=False, comments=[_lease_comment()]) mock_api.side_effect = side with patch.dict(os.environ, MERGER_ENV, clear=True): result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=True, remote="prgs") self.assertFalse(result["cleanup_performed"]) self.assertFalse(result["pr_merged_or_closed"]) self.assertEqual(calls["post"], [], "never force-clean an open PR lease") self.assertTrue( any("still open" in r for r in result.get("cleanup_skipped_reason", []))) if __name__ == "__main__": unittest.main()