sysadmin
87beb44394
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)
207 lines
7.2 KiB
Python
207 lines
7.2 KiB
Python
#!/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() |