"""Tests for per-PR reviewer leases (#407).""" import sys import unittest from datetime import datetime, timedelta, timezone from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import merger_lease_adoption as mla import reviewer_pr_lease as leases def _lease_comment( pr_number: int, session_id: str, *, phase: str = "claimed", minutes_ago: int = 0, candidate_head: str = "a" * 40, ) -> dict: now = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago) body = leases.format_lease_body( repo="Scaled-Tech-Consulting/Gitea-Tools", pr_number=pr_number, issue_number=295, reviewer_identity="rev1", profile="prgs-reviewer", session_id=session_id, worktree="branches/review-pr382", phase=phase, candidate_head=candidate_head, target_branch="master", target_branch_sha="b" * 40, last_activity=now, ) return {"id": 1, "body": body, "user": {"login": "rev1"}} class TestReviewerLeaseAcquire(unittest.TestCase): def setUp(self): leases.clear_session_lease() def test_two_reviewers_cannot_lease_same_pr(self): comments = [_lease_comment(382, "session-a")] result = leases.assess_acquire_lease( comments, pr_number=382, reviewer_identity="rev2", profile="prgs-reviewer", session_id="session-b", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=295, worktree="branches/review-pr382-b", candidate_head="c" * 40, target_branch="master", target_branch_sha="d" * 40, ) self.assertFalse(result["acquire_allowed"]) self.assertTrue(any("already has active" in r for r in result["reasons"])) def test_two_reviewers_can_lease_different_prs(self): comments = [_lease_comment(382, "session-a")] result = leases.assess_acquire_lease( comments, pr_number=383, reviewer_identity="rev2", profile="prgs-reviewer", session_id="session-b", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=296, worktree="branches/review-pr383", candidate_head="c" * 40, target_branch="master", target_branch_sha="d" * 40, ) self.assertTrue(result["acquire_allowed"]) self.assertIsNotNone(result["lease_body"]) class TestReviewerLeaseFreshness(unittest.TestCase): def test_stale_warning_after_30_minutes(self): lease = leases.parse_lease_comment( _lease_comment(382, "session-a", minutes_ago=35)["body"] ) self.assertEqual( leases.classify_lease_freshness(lease), "stale_warning", ) def test_reclaimable_after_60_minutes(self): lease = leases.parse_lease_comment( _lease_comment(382, "session-a", minutes_ago=65)["body"] ) self.assertEqual( leases.classify_lease_freshness(lease), "reclaimable", ) class TestReviewerLeaseNewestWins(unittest.TestCase): """#577: terminal release must supersede older claimed markers.""" def test_released_after_claimed_has_no_active_lease(self): claim = _lease_comment(516, "session-a", phase="claimed") claim["id"] = 1 release = _lease_comment(516, "session-a", phase="released") release["id"] = 2 active = leases.find_active_reviewer_lease( [claim, release], pr_number=516 ) self.assertIsNone(active) def test_new_claim_after_release_is_active(self): claim = _lease_comment(516, "session-a", phase="claimed") claim["id"] = 1 release = _lease_comment(516, "session-a", phase="released") release["id"] = 2 reclaim = _lease_comment(516, "session-b", phase="claimed") reclaim["id"] = 3 active = leases.find_active_reviewer_lease( [claim, release, reclaim], pr_number=516 ) self.assertIsNotNone(active) self.assertEqual(active.get("session_id"), "session-b") self.assertEqual(active.get("phase"), "claimed") self.assertEqual(active.get("comment_id"), 3) def test_done_after_claimed_has_no_active_lease(self): claim = _lease_comment(516, "session-a", phase="claimed") claim["id"] = 10 done = _lease_comment(516, "session-a", phase="done") done["id"] = 11 active = leases.find_active_reviewer_lease( [claim, done], pr_number=516 ) self.assertIsNone(active) class TestReviewerLeaseMutationGate(unittest.TestCase): def setUp(self): leases.clear_session_lease() def test_reviewer_without_lease_cannot_mutate(self): head = "f" * 40 comments = [_lease_comment(382, "other-session", candidate_head=head)] result = leases.assess_mutation_lease_gate( pr_number=382, comments=comments, reviewer_identity="rev1", session_id="my-session", mutation="approve", live_head_sha=head, pinned_head_sha=head, ) self.assertTrue(result["block"]) def test_owned_lease_allows_mutation(self): head = "f" * 40 comments = [_lease_comment(382, "my-session", candidate_head=head)] leases.record_session_lease({ "pr_number": 382, "session_id": "my-session", "candidate_head": head, "target_branch": "master", "comment_id": 101, }, lease_provenance=mla.build_lease_provenance( source=mla.SOURCE_ACQUIRE, comment_id=101, )) result = leases.assess_mutation_lease_gate( pr_number=382, comments=comments, reviewer_identity="rev1", session_id="my-session", mutation="approve", live_head_sha=head, pinned_head_sha=head, ) self.assertFalse(result["block"]) def test_head_change_invalidates_lease(self): reviewed = "f" * 40 live = "e" * 40 comments = [_lease_comment(382, "my-session", candidate_head=reviewed)] leases.record_session_lease({ "pr_number": 382, "session_id": "my-session", "candidate_head": reviewed, "comment_id": 102, }, lease_provenance=mla.build_lease_provenance( source=mla.SOURCE_ACQUIRE, comment_id=102, )) result = leases.assess_mutation_lease_gate( pr_number=382, comments=comments, reviewer_identity="rev1", session_id="my-session", mutation="merge", live_head_sha=live, pinned_head_sha=reviewed, ) self.assertTrue(result["block"]) self.assertTrue(any("head" in r.lower() for r in result["reasons"])) class TestReviewerLeaseMcpGate(unittest.TestCase): def setUp(self): leases.clear_session_lease() patch("mcp_server.verify_preflight_purity").start() patch("gitea_audit.audit_enabled", return_value=False).start() mcp_server = __import__("mcp_server") mcp_server._IDENTITY_CACHE.clear() mcp_server.init_review_decision_lock("prgs", "review_pr") mcp_server.record_preflight_check("whoami") mcp_server.record_preflight_check("capability", "reviewer") def tearDown(self): patch.stopall() leases.clear_session_lease() def test_reviewer_pr_lease_gate_helper_blocks_without_session(self): import mcp_server head = "a" * 40 with patch("mcp_server._fetch_pr_comments", return_value=[]): reasons = mcp_server._reviewer_pr_lease_gate( pr_number=382, remote="prgs", host=None, org=None, repo=None, mutation="approve", live_head_sha=head, pinned_head_sha=head, ) self.assertTrue(any("lease" in r.lower() for r in reasons)) class TestReviewerLeaseHandoffDiagnose(unittest.TestCase): """#599: canonical next-action diagnosis for open-PR lease handoff.""" def setUp(self): leases.clear_session_lease() def test_no_lease_next_action_acquire(self): result = leases.diagnose_reviewer_pr_lease_handoff( [], pr_number=592, current_session_id="me-1", current_reviewer_identity="sysadmin", proposed_worktree="branches/review-pr-592", ) self.assertEqual(result["classification"], "no_lease") self.assertEqual(result["next_action"], leases.NEXT_ACTION_ACQUIRE) self.assertIsNone(result["active_lease"]) def test_foreign_active_wait_pr592_style(self): # Instructed lease gone; newer foreign claim is active. old = _lease_comment(592, "23139-da018dab43ba", phase="released") old["id"] = 8640 active = _lease_comment(592, "51515-2bfb70f0685f", phase="claimed") active["id"] = 8647 result = leases.diagnose_reviewer_pr_lease_handoff( [old, active], pr_number=592, current_session_id="fresh-session", current_reviewer_identity="sysadmin", proposed_worktree="branches/review-pr-592-issue-590", instructed_session_id="23139-da018dab43ba", instructed_comment_id=8640, ) self.assertEqual( result["classification"], "instructed_lease_missing_with_replacement", ) self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT) self.assertTrue(result["instructed_lease_missing_with_replacement"]) self.assertEqual(result["active_lease"]["session_id"], "51515-2bfb70f0685f") self.assertEqual(result["active_lease"]["comment_id"], 8647) self.assertFalse(result["mutation_allowed"]) def test_foreign_reclaimable_release_expired(self): reclaim = _lease_comment( 592, "foreign-old", phase="claimed", minutes_ago=65 ) reclaim["id"] = 99 result = leases.diagnose_reviewer_pr_lease_handoff( [reclaim], pr_number=592, current_session_id="me-1", current_reviewer_identity="sysadmin", proposed_worktree="branches/review-pr-592", ) self.assertEqual(result["classification"], "foreign_reclaimable") self.assertEqual( result["next_action"], leases.NEXT_ACTION_RELEASE_EXPIRED_LEASE ) def test_own_active_resume(self): head = "a" * 40 claim = _lease_comment(592, "me-1", phase="claimed", candidate_head=head) claim["id"] = 10 leases.record_session_lease({ "pr_number": 592, "session_id": "me-1", "candidate_head": head, "comment_id": 10, }, lease_provenance=mla.build_lease_provenance( source=mla.SOURCE_ACQUIRE, comment_id=10, )) result = leases.diagnose_reviewer_pr_lease_handoff( [claim], pr_number=592, current_session_id="me-1", current_reviewer_identity="rev1", proposed_worktree="branches/review-pr382", ) self.assertEqual(result["classification"], "own_active") self.assertEqual( result["next_action"], leases.NEXT_ACTION_RESUME_EXACT_OWNER_SESSION ) self.assertTrue(result["mutation_allowed"]) def test_worktree_binding_mismatch(self): claim = _lease_comment(592, "me-1", phase="claimed") claim["id"] = 11 # _lease_comment uses branches/review-pr382 result = leases.diagnose_reviewer_pr_lease_handoff( [claim], pr_number=592, current_session_id="me-1", current_reviewer_identity="rev1", proposed_worktree="branches/review-pr-592-issue-590", env_bound_worktree="branches/review-pr-538", ) self.assertEqual(result["classification"], "worktree_binding_mismatch") self.assertEqual( result["next_action"], leases.NEXT_ACTION_REPAIR_WORKTREE_BINDING ) self.assertFalse(result["worktree_binding"]["match"]) self.assertFalse(result["mutation_allowed"]) if __name__ == "__main__": unittest.main()