sysadmin
98fbfb3de7
Prevent approved PRs from stalling when master advances. Add gitea_assess_pr_sync_status and gitea_update_pr_branch_by_merge with head/base pinning, author-only updates, conflict handoff, and approval invalidation after head changes. Wire task capability map, sequential controller routing in review-merge workflow, and hermetic AC tests.
339 lines
12 KiB
Python
339 lines
12 KiB
Python
"""Hermetic tests for PR synchronization / conflict-remediation lifecycle."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from pr_sync_status import (
|
|
ACTION_AUTHOR_CONFLICT_REMEDIATION,
|
|
ACTION_BLOCKED,
|
|
ACTION_FRESH_REVIEW_REQUIRED,
|
|
ACTION_MERGE_NOW,
|
|
ACTION_UPDATE_BRANCH_BY_MERGE,
|
|
assess_post_update_head_transition,
|
|
assess_pr_sync_status,
|
|
assess_sequential_queue_step,
|
|
assess_update_pr_branch_preflight,
|
|
lease_usable_at_head,
|
|
merge_approval_usable_at_head,
|
|
)
|
|
|
|
|
|
def _sha(prefix: str) -> str:
|
|
return (prefix + "0" * 40)[:40]
|
|
|
|
|
|
PR_HEAD = _sha("aaaaaaaa")
|
|
BASE_HEAD = _sha("bbbbbbbb")
|
|
NEW_HEAD = _sha("cccccccc")
|
|
OLD_HEAD = _sha("dddddddd")
|
|
|
|
|
|
def _base_kwargs(**overrides):
|
|
data = {
|
|
"host": "gitea.prgs.cc",
|
|
"org": "Scaled-Tech-Consulting",
|
|
"repo": "Gitea-Tools",
|
|
"pr_number": 100,
|
|
"pr_state": "open",
|
|
"source_branch": "fix/issue-100",
|
|
"pr_head_sha": PR_HEAD,
|
|
"base_head_sha": BASE_HEAD,
|
|
"commits_behind": 0,
|
|
"mergeable": True,
|
|
"has_conflicts": False,
|
|
"branch_protection_requires_current_base": False,
|
|
"approval_at_current_head": True,
|
|
"checks_status": "success",
|
|
"active_author_lock": True,
|
|
"active_reviewer_lease": False,
|
|
"active_merger_lease": False,
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
class TestAssessPrSyncStatus(unittest.TestCase):
|
|
def test_approved_current_mergeable_merge_now(self):
|
|
result = assess_pr_sync_status(**_base_kwargs())
|
|
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
|
self.assertTrue(result["approval_valid_for_merge"])
|
|
self.assertFalse(result["stale_approval"])
|
|
self.assertEqual(result["pr_head_sha"], PR_HEAD)
|
|
self.assertEqual(result["base_head_sha"], BASE_HEAD)
|
|
|
|
def test_approved_outdated_update_not_required_merge_now(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
commits_behind=3,
|
|
branch_protection_requires_current_base=False,
|
|
)
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
|
self.assertTrue(result["approval_valid_for_merge"])
|
|
self.assertTrue(any("does not require" in r for r in result["reasons"]))
|
|
|
|
def test_outdated_update_required_no_conflict(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
commits_behind=2,
|
|
branch_protection_requires_current_base=True,
|
|
mergeable=True,
|
|
has_conflicts=False,
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
|
|
)
|
|
self.assertFalse(result["approval_valid_for_merge"])
|
|
self.assertTrue(any("update_branch_by_merge" in r for r in result["reasons"]))
|
|
|
|
def test_outdated_has_conflicts_author_remediation(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
commits_behind=5,
|
|
branch_protection_requires_current_base=True,
|
|
mergeable=False,
|
|
has_conflicts=True,
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
result["recommended_next_action"], ACTION_AUTHOR_CONFLICT_REMEDIATION
|
|
)
|
|
self.assertFalse(result["approval_valid_for_merge"])
|
|
|
|
def test_stale_approval_fresh_review_required(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
approval_at_current_head=False,
|
|
commits_behind=0,
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
|
|
)
|
|
self.assertTrue(result["stale_approval"])
|
|
self.assertFalse(result["approval_valid_for_merge"])
|
|
|
|
def test_stale_prepared_verdict_cannot_cross_heads(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
approval_at_current_head=True,
|
|
prepared_verdict_head_sha=OLD_HEAD,
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
|
|
)
|
|
self.assertTrue(result["stale_prepared_verdict"])
|
|
self.assertFalse(result["approval_valid_for_merge"])
|
|
|
|
def test_missing_heads_fail_closed(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(pr_head_sha="short", base_head_sha=None)
|
|
)
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
self.assertTrue(any("PR head" in r for r in result["reasons"]))
|
|
|
|
def test_closed_pr_blocked(self):
|
|
result = assess_pr_sync_status(**_base_kwargs(pr_state="closed"))
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
|
|
def test_failed_checks_block_merge_now(self):
|
|
result = assess_pr_sync_status(**_base_kwargs(checks_status="failure"))
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
|
|
def test_stale_approval_with_update_required_routes_update(self):
|
|
result = assess_pr_sync_status(
|
|
**_base_kwargs(
|
|
approval_at_current_head=False,
|
|
commits_behind=1,
|
|
branch_protection_requires_current_base=True,
|
|
mergeable=True,
|
|
has_conflicts=False,
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
|
|
)
|
|
|
|
|
|
class TestUpdatePrBranchPreflight(unittest.TestCase):
|
|
def _ok_kwargs(self, **overrides):
|
|
data = {
|
|
"role_kind": "author",
|
|
"expected_pr_head_sha": PR_HEAD,
|
|
"live_pr_head_sha": PR_HEAD,
|
|
"expected_base_head_sha": BASE_HEAD,
|
|
"live_base_head_sha": BASE_HEAD,
|
|
"has_conflicts": False,
|
|
"mergeable": True,
|
|
"style": "merge",
|
|
"has_author_lock": True,
|
|
"worktree_path": "/Users/x/Development/Gitea-Tools/branches/issue-100",
|
|
"worktree_owns_source_branch": True,
|
|
"control_checkout_clean": True,
|
|
"master_parity_ok": True,
|
|
"force_push": False,
|
|
"rebase": False,
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
def test_author_allowed_when_preflight_clean(self):
|
|
result = assess_update_pr_branch_preflight(**self._ok_kwargs())
|
|
self.assertTrue(result["mutation_allowed"])
|
|
self.assertFalse(result["head_race"])
|
|
self.assertFalse(result["base_race"])
|
|
|
|
def test_head_race_fail_closed(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(live_pr_head_sha=NEW_HEAD)
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(result["head_race"])
|
|
self.assertTrue(any("PR head race" in r for r in result["reasons"]))
|
|
|
|
def test_base_race_fail_closed(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(live_base_head_sha=NEW_HEAD)
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(result["base_race"])
|
|
self.assertTrue(any("base head race" in r for r in result["reasons"]))
|
|
|
|
def test_conflicts_structured_handoff_no_mutation(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(has_conflicts=True, mergeable=False)
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(result["author_remediation_handoff"])
|
|
self.assertTrue(any("conflicts" in r.lower() for r in result["reasons"]))
|
|
|
|
def test_reviewer_denied(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(role_kind="reviewer")
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(any("author-only" in r for r in result["reasons"]))
|
|
|
|
def test_merger_denied(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(role_kind="merger")
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(any("author-only" in r for r in result["reasons"]))
|
|
|
|
def test_no_force_push(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(force_push=True)
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(any("force-push" in r for r in result["reasons"]))
|
|
|
|
def test_no_rebase(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(style="rebase", rebase=True)
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(any("rebase" in r for r in result["reasons"]))
|
|
|
|
def test_missing_lock_fail_closed(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(has_author_lock=False)
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
|
|
def test_worktree_not_under_branches(self):
|
|
result = assess_update_pr_branch_preflight(
|
|
**self._ok_kwargs(worktree_path="/tmp/not-a-branches-path")
|
|
)
|
|
self.assertFalse(result["mutation_allowed"])
|
|
self.assertTrue(any("branches/" in r for r in result["reasons"]))
|
|
|
|
|
|
class TestPostUpdateHeadTransition(unittest.TestCase):
|
|
def test_update_invalidates_approval_and_requires_fresh_review(self):
|
|
result = assess_post_update_head_transition(
|
|
former_pr_head_sha=PR_HEAD,
|
|
new_pr_head_sha=NEW_HEAD,
|
|
former_approval_head_sha=PR_HEAD,
|
|
former_reviewer_lease_head_sha=PR_HEAD,
|
|
former_merger_lease_head_sha=PR_HEAD,
|
|
prepared_verdict_head_sha=PR_HEAD,
|
|
)
|
|
self.assertTrue(result["head_changed"])
|
|
self.assertTrue(result["approval_invalidated"])
|
|
self.assertTrue(result["reviewer_lease_superseded"])
|
|
self.assertTrue(result["merger_lease_superseded"])
|
|
self.assertTrue(result["prepared_verdict_invalidated"])
|
|
self.assertEqual(
|
|
result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
|
|
)
|
|
|
|
def test_same_head_unexpected(self):
|
|
result = assess_post_update_head_transition(
|
|
former_pr_head_sha=PR_HEAD,
|
|
new_pr_head_sha=PR_HEAD,
|
|
)
|
|
self.assertFalse(result["head_changed"])
|
|
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
|
|
|
|
|
class TestStaleArtifacts(unittest.TestCase):
|
|
def test_stale_approval_cannot_authorize_merge(self):
|
|
result = merge_approval_usable_at_head(
|
|
current_head_sha=NEW_HEAD,
|
|
approved_head_sha=PR_HEAD,
|
|
)
|
|
self.assertFalse(result["approval_usable"])
|
|
|
|
def test_fresh_approval_usable(self):
|
|
result = merge_approval_usable_at_head(
|
|
current_head_sha=NEW_HEAD,
|
|
approved_head_sha=NEW_HEAD,
|
|
)
|
|
self.assertTrue(result["approval_usable"])
|
|
|
|
def test_stale_reviewer_lease_cannot_cross_heads(self):
|
|
result = lease_usable_at_head(
|
|
lease_kind="reviewer",
|
|
lease_head_sha=PR_HEAD,
|
|
current_head_sha=NEW_HEAD,
|
|
)
|
|
self.assertFalse(result["lease_usable"])
|
|
|
|
def test_stale_merger_lease_cannot_cross_heads(self):
|
|
result = lease_usable_at_head(
|
|
lease_kind="merger",
|
|
lease_head_sha=PR_HEAD,
|
|
current_head_sha=NEW_HEAD,
|
|
)
|
|
self.assertFalse(result["lease_usable"])
|
|
|
|
|
|
class TestSequentialQueue(unittest.TestCase):
|
|
def test_reassess_after_merge_and_master_refresh(self):
|
|
result = assess_sequential_queue_step(
|
|
just_merged=True,
|
|
master_refreshed=True,
|
|
next_pr_number=101,
|
|
)
|
|
self.assertTrue(result["reassess_allowed"])
|
|
self.assertTrue(result["process_next"])
|
|
self.assertTrue(result["post_merge_cleanup_handoff"])
|
|
self.assertEqual(result["next_pr_number"], 101)
|
|
|
|
def test_block_reassess_without_master_refresh(self):
|
|
result = assess_sequential_queue_step(
|
|
just_merged=True,
|
|
master_refreshed=False,
|
|
next_pr_number=101,
|
|
)
|
|
self.assertFalse(result["reassess_allowed"])
|
|
self.assertTrue(any("master must be refreshed" in r for r in result["reasons"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|