sysadmin
9b96085d8d
find_active_reviewer_lease walked past terminal markers and re-armed older claimed leases after gitea_release_reviewer_pr_lease. Treat the newest marker as authoritative (append-only ledger). Closes #577
241 lines
8.1 KiB
Python
241 lines
8.1 KiB
Python
"""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))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |