feat: add conflict-fix leases and stale-head protection (Closes #399)
Introduce structured PR work leases so author conflict-fix pushes cannot race reviewer validation, approval, or merge on a moving head SHA. - pr_work_lease.py: parse/acquire leases, push and reviewer mutation gates - gitea_acquire_conflict_fix_lease, gitea_assess_conflict_fix_push MCP tools - Enforce reviewed head SHA on mark_final_review_decision, submit_pr_review, merge_pr - Final-report rules and workflow sections 20A / 26B - tests/test_pr_work_lease.py (14 cases)
This commit is contained in:
@@ -0,0 +1,207 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Regression tests for conflict-fix and reviewer PR work leases (#399)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from pr_work_lease import ( # noqa: E402
|
||||
CONFLICT_FIX_LEASE_MARKER,
|
||||
REVIEWER_LEASE_MARKER,
|
||||
assess_conflict_fix_final_report,
|
||||
assess_conflict_fix_push,
|
||||
assess_head_sha_equality,
|
||||
assess_reviewer_mutation_blocked,
|
||||
assess_reviewer_stale_head_final_report,
|
||||
format_conflict_fix_lease_body,
|
||||
parse_conflict_fix_lease_comment,
|
||||
parse_reviewer_lease_comment,
|
||||
)
|
||||
|
||||
HEAD_A = "a" * 40
|
||||
HEAD_B = "b" * 40
|
||||
NOW = datetime(2026, 7, 7, 15, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _reviewer_lease_body(*, phase: str = "validating", expires_minutes: int = 60) -> str:
|
||||
expires = (NOW + timedelta(minutes=expires_minutes)).isoformat().replace("+00:00", "Z")
|
||||
return "\n".join([
|
||||
REVIEWER_LEASE_MARKER,
|
||||
"pr: #376",
|
||||
"phase: " + phase,
|
||||
f"candidate_head: {HEAD_A}",
|
||||
f"expires_at: {expires}",
|
||||
"profile: prgs-reviewer",
|
||||
])
|
||||
|
||||
|
||||
def _conflict_fix_body(*, phase: str = "claimed", worktree: str = "branches/fix-376") -> str:
|
||||
expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
|
||||
return "\n".join([
|
||||
CONFLICT_FIX_LEASE_MARKER,
|
||||
"pr: #376",
|
||||
f"phase: {phase}",
|
||||
f"worktree: {worktree}",
|
||||
f"head_before: {HEAD_A}",
|
||||
f"expires_at: {expires}",
|
||||
"profile: prgs-author",
|
||||
])
|
||||
|
||||
|
||||
class TestLeaseParsing(unittest.TestCase):
|
||||
def test_parse_reviewer_lease(self):
|
||||
parsed = parse_reviewer_lease_comment(_reviewer_lease_body())
|
||||
self.assertEqual(parsed["pr_number"], 376)
|
||||
self.assertEqual(parsed["phase"], "validating")
|
||||
self.assertEqual(parsed["candidate_head"], HEAD_A)
|
||||
|
||||
def test_parse_conflict_fix_lease(self):
|
||||
parsed = parse_conflict_fix_lease_comment(_conflict_fix_body())
|
||||
self.assertEqual(parsed["pr_number"], 376)
|
||||
self.assertEqual(parsed["phase"], "claimed")
|
||||
|
||||
|
||||
class TestConflictFixPushGate(unittest.TestCase):
|
||||
def test_blocks_push_during_active_reviewer_lease(self):
|
||||
comments = [{"body": _reviewer_lease_body()}]
|
||||
result = assess_conflict_fix_push(
|
||||
pr_number=376,
|
||||
comments=comments,
|
||||
branch_head_before=HEAD_A,
|
||||
branch_head_after=HEAD_B,
|
||||
worktree_path="branches/fix-376",
|
||||
push_cwd="/proj/branches/fix-376",
|
||||
is_fast_forward=True,
|
||||
now=NOW,
|
||||
)
|
||||
self.assertFalse(result["push_allowed"])
|
||||
self.assertTrue(any("reviewer lease" in r for r in result["reasons"]))
|
||||
|
||||
def test_rejects_non_fast_forward(self):
|
||||
result = assess_conflict_fix_push(
|
||||
pr_number=376,
|
||||
comments=[],
|
||||
branch_head_before=HEAD_A,
|
||||
branch_head_after=HEAD_B,
|
||||
worktree_path="branches/fix-376",
|
||||
push_cwd="/proj/branches/fix-376",
|
||||
is_fast_forward=False,
|
||||
now=NOW,
|
||||
)
|
||||
self.assertFalse(result["push_allowed"])
|
||||
self.assertTrue(any("non-fast-forward" in r for r in result["reasons"]))
|
||||
|
||||
def test_wrong_cwd_push_attempt(self):
|
||||
result = assess_conflict_fix_push(
|
||||
pr_number=376,
|
||||
comments=[],
|
||||
branch_head_before=HEAD_A,
|
||||
branch_head_after=HEAD_B,
|
||||
worktree_path="branches/fix-376",
|
||||
push_cwd="/proj/master",
|
||||
is_fast_forward=True,
|
||||
now=NOW,
|
||||
)
|
||||
self.assertFalse(result["push_allowed"])
|
||||
self.assertTrue(any("cwd" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_sibling_conflict_fix_collision(self):
|
||||
comments = [{"body": _conflict_fix_body(phase="pushing", worktree="branches/other")}]
|
||||
result = assess_conflict_fix_push(
|
||||
pr_number=376,
|
||||
comments=comments,
|
||||
branch_head_before=HEAD_A,
|
||||
branch_head_after=HEAD_B,
|
||||
worktree_path="branches/fix-376",
|
||||
push_cwd="/proj/branches/fix-376",
|
||||
is_fast_forward=True,
|
||||
now=NOW,
|
||||
)
|
||||
self.assertFalse(result["push_allowed"])
|
||||
self.assertTrue(any("sibling conflict-fix" in r for r in result["reasons"]))
|
||||
|
||||
|
||||
class TestReviewerMutationGate(unittest.TestCase):
|
||||
def test_blocks_review_during_conflict_fix(self):
|
||||
comments = [{"body": _conflict_fix_body(phase="pushing")}]
|
||||
result = assess_reviewer_mutation_blocked(
|
||||
pr_number=376,
|
||||
comments=comments,
|
||||
reviewed_head_sha=HEAD_A,
|
||||
live_head_sha=HEAD_A,
|
||||
mutation="approve",
|
||||
now=NOW,
|
||||
)
|
||||
self.assertFalse(result["mutation_allowed"])
|
||||
self.assertTrue(any("conflict-fix lease" in r for r in result["reasons"]))
|
||||
|
||||
def test_stale_head_blocks_approval(self):
|
||||
result = assess_reviewer_mutation_blocked(
|
||||
pr_number=376,
|
||||
comments=[],
|
||||
reviewed_head_sha=HEAD_A,
|
||||
live_head_sha=HEAD_B,
|
||||
mutation="merge",
|
||||
now=NOW,
|
||||
)
|
||||
self.assertFalse(result["mutation_allowed"])
|
||||
self.assertTrue(result["head_check"]["head_changed"])
|
||||
|
||||
def test_head_equality_required_fields(self):
|
||||
result = assess_head_sha_equality(HEAD_A, HEAD_B)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["head_changed"])
|
||||
|
||||
|
||||
class TestFinalReportProof(unittest.TestCase):
|
||||
def test_reviewer_stale_head_report_requires_fields(self):
|
||||
result = assess_reviewer_stale_head_final_report("no head proof here")
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_reviewer_stale_head_report_passes(self):
|
||||
report = "\n".join([
|
||||
f"Reviewed head SHA: {HEAD_A}",
|
||||
f"Final live head SHA before approval: {HEAD_A}",
|
||||
f"Final live head SHA before merge: {HEAD_A}",
|
||||
"Push occurred during validation: no",
|
||||
])
|
||||
result = assess_reviewer_stale_head_final_report(report)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_conflict_fix_report_requires_fields(self):
|
||||
result = assess_conflict_fix_final_report("incomplete")
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_conflict_fix_report_passes(self):
|
||||
report = "\n".join([
|
||||
f"Branch head before push: {HEAD_A}",
|
||||
f"Branch head after push: {HEAD_B}",
|
||||
"Active reviewer lease status: none",
|
||||
"Whether push was fast-forward: yes",
|
||||
"Whether any reviewer was active: no",
|
||||
])
|
||||
result = assess_conflict_fix_final_report(report)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
|
||||
class TestFormatLease(unittest.TestCase):
|
||||
def test_format_conflict_fix_lease_includes_marker(self):
|
||||
body = format_conflict_fix_lease_body(
|
||||
pr_number=376,
|
||||
branch="feat/x",
|
||||
worktree="branches/fix-376",
|
||||
profile="prgs-author",
|
||||
head_before=HEAD_A,
|
||||
)
|
||||
self.assertIn(CONFLICT_FIX_LEASE_MARKER, body)
|
||||
parsed = parse_conflict_fix_lease_comment(body)
|
||||
self.assertEqual(parsed["pr_number"], 376)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user