360 lines
14 KiB
Python
360 lines
14 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,
|
|
find_active_conflict_fix_lease,
|
|
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)
|
|
|
|
|
|
class TestConflictFixLeaseLifecycle(unittest.TestCase):
|
|
def test_claim_followed_by_matching_release(self):
|
|
claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
|
|
expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
|
|
release_body = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"branch: feat/fix-376",
|
|
"worktree: branches/fix-376",
|
|
"profile: prgs-author",
|
|
"phase: released",
|
|
f"head_before: {HEAD_A}",
|
|
f"head_after: {HEAD_B}",
|
|
f"expires_at: {expires}",
|
|
])
|
|
comments = [{"body": claim_body}, {"body": release_body}]
|
|
lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
|
|
self.assertIsNone(lease)
|
|
|
|
def test_expired_claim_without_release(self):
|
|
past_expires = (NOW - timedelta(minutes=10)).isoformat().replace("+00:00", "Z")
|
|
claim_body = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"phase: claimed",
|
|
f"head_before: {HEAD_A}",
|
|
f"expires_at: {past_expires}",
|
|
"profile: prgs-author",
|
|
])
|
|
comments = [{"body": claim_body}]
|
|
lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
|
|
self.assertIsNone(lease)
|
|
|
|
def test_mismatched_release_different_head(self):
|
|
claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
|
|
expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
|
|
release_body = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"profile: prgs-author",
|
|
"phase: released",
|
|
f"head_before: {HEAD_B}",
|
|
f"expires_at: {expires}",
|
|
])
|
|
comments = [{"body": claim_body}, {"body": release_body}]
|
|
lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
|
|
self.assertIsNotNone(lease)
|
|
self.assertEqual(lease["phase"], "claimed")
|
|
|
|
def test_mismatched_release_different_branch(self):
|
|
claim_body = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"branch: feat/branch-A",
|
|
"phase: claimed",
|
|
f"head_before: {HEAD_A}",
|
|
f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
|
|
"profile: prgs-author",
|
|
])
|
|
release_body = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"branch: feat/branch-B",
|
|
"phase: released",
|
|
f"head_before: {HEAD_A}",
|
|
f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
|
|
"profile: prgs-author",
|
|
])
|
|
comments = [{"body": claim_body}, {"body": release_body}]
|
|
lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
|
|
self.assertIsNotNone(lease)
|
|
self.assertEqual(lease["phase"], "claimed")
|
|
|
|
def test_release_followed_by_newer_claim(self):
|
|
claim_1 = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
|
|
expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
|
|
release_1 = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"profile: prgs-author",
|
|
"phase: released",
|
|
f"head_before: {HEAD_A}",
|
|
f"head_after: {HEAD_B}",
|
|
f"expires_at: {expires}",
|
|
])
|
|
claim_2 = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"profile: prgs-author",
|
|
"phase: claimed",
|
|
f"head_before: {HEAD_B}",
|
|
f"expires_at: {expires}",
|
|
])
|
|
comments = [{"body": claim_1}, {"body": release_1}, {"body": claim_2}]
|
|
lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
|
|
self.assertIsNotNone(lease)
|
|
self.assertEqual(lease["head_before"], HEAD_B)
|
|
|
|
def test_malformed_or_ambiguous_markers(self):
|
|
malformed_release = "\n".join([
|
|
CONFLICT_FIX_LEASE_MARKER,
|
|
"pr: #376",
|
|
"phase: released",
|
|
# missing head_before and profile
|
|
])
|
|
claim_body = _conflict_fix_body(phase="claimed")
|
|
comments = [{"body": claim_body}, {"body": malformed_release}]
|
|
lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
|
|
self.assertIsNotNone(lease)
|
|
|
|
def test_pr818_historical_sequence(self):
|
|
comment_14696 = "\n".join([
|
|
"<!-- mcp-conflict-fix-lease:v1 -->",
|
|
"pr: #818",
|
|
"branch: feat/issue-638-webui-app-shell-phase1",
|
|
"worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
|
|
"profile: prgs-author",
|
|
"session_id: unknown",
|
|
"phase: claimed",
|
|
"head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
|
|
"expires_at: 2026-07-23T07:12:13Z",
|
|
"reviewer_active: no",
|
|
])
|
|
comment_14730 = "\n".join([
|
|
"<!-- mcp-conflict-fix-lease:v1 -->",
|
|
"pr: #818",
|
|
"branch: feat/issue-638-webui-app-shell-phase1",
|
|
"worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
|
|
"profile: prgs-author",
|
|
"session_id: prgs-author-61241-e5129c60",
|
|
"phase: released",
|
|
"head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
|
|
"head_after: 64b6eb5d5402663098de5ded3b0617cc3b3df98f",
|
|
"expires_at: 2026-07-23T06:05:00Z",
|
|
"reviewer_active: no",
|
|
])
|
|
comments = [{"body": comment_14696}, {"body": comment_14730}]
|
|
check_now = datetime(2026, 7, 23, 6, 30, tzinfo=timezone.utc)
|
|
lease = find_active_conflict_fix_lease(comments, pr_number=818, now=check_now)
|
|
self.assertIsNone(lease)
|
|
|
|
reviewer_gate = assess_reviewer_mutation_blocked(
|
|
pr_number=818,
|
|
comments=comments,
|
|
reviewed_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
|
|
live_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
|
|
mutation="approve",
|
|
now=check_now,
|
|
)
|
|
self.assertTrue(reviewer_gate["mutation_allowed"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |