"""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()