"""Tests for post-merge cleanup proof enforcement (#402).""" import sys import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from final_report_validator import assess_final_report_validator # noqa: E402 from post_merge_cleanup_proof import ( # noqa: E402 CLEANUP_SKIPPED, assess_post_merge_cleanup_proof, ) MERGE_SHA = "a" * 40 HEAD_BRANCH = "feat/issue-274-branches-only-worktrees" WORKTREE = "branches/review-pr374-conflicts" def _full_remote_cleanup_report(**overrides): fields = { "Task": "review PR #374", "Merge result": "merged", "Merge commit SHA": MERGE_SHA, "Cleanup status": "remote branch deleted", "Delete-branch capability resolved": "gitea.branch.delete allowed via delete_branch task", "Merged PR head branch": HEAD_BRANCH, "Deleted branch": HEAD_BRANCH, "Branch protection": "none", "Open PR inventory proof": "no other open PR references branch (inventory complete)", "Active heartbeat/claim/lease": "none", "Cleanup mutations": "gitea_delete_branch on remote head branch", } fields.update(overrides) lines = ["## Controller Handoff", ""] lines.extend(f"- {key}: {value}" for key, value in fields.items()) return "\n".join(lines) def _full_worktree_cleanup_report(**overrides): fields = { "Task": "review PR #374", "Merge result": "merged", "Cleanup status": "local worktree removed", "Removed worktree path": WORKTREE, "Pre-removal tracked state": "clean", "Pre-removal untracked state": "clean", "Git worktree list after removal": "only main checkout listed", "Cleanup mutations": "git worktree remove on session-owned review worktree", } fields.update(overrides) lines = ["## Controller Handoff", ""] lines.extend(f"- {key}: {value}" for key, value in fields.items()) return "\n".join(lines) class TestPostMergeCleanupProof(unittest.TestCase): def test_no_cleanup_claim_passes(self): report = "## Controller Handoff\n- Task: review PR #1\n- Merge result: merged" result = assess_post_merge_cleanup_proof(report) self.assertFalse(result["block"]) def test_missing_capability_proof_blocked(self): report = _full_remote_cleanup_report( **{"Delete-branch capability resolved": "delete succeeded"} ) result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) self.assertTrue(any("capability" in r.lower() for r in result["reasons"])) def test_unmerged_pr_blocked(self): report = _full_remote_cleanup_report(**{"Merge result": "not merged"}) result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) self.assertTrue(any("merge result" in r.lower() for r in result["reasons"])) def test_wrong_branch_blocked(self): report = _full_remote_cleanup_report( **{"Deleted branch": "feat/other-branch"} ) report += "\nDeleted branch does not match merged PR head branch" result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) def test_protected_branch_blocked(self): report = _full_remote_cleanup_report(**{"Branch protection": "enabled"}) result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) def test_open_pr_reference_blocked(self): report = _full_remote_cleanup_report( **{"Open PR inventory proof": "PR #999 still references branch"} ) result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) def test_active_claim_blocked(self): report = _full_remote_cleanup_report( **{"Active heartbeat/claim/lease": "status:in-progress on linked issue"} ) result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) def test_dirty_worktree_blocked(self): report = _full_worktree_cleanup_report( **{"Pre-removal tracked state": "dirty"} ) result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) self.assertTrue(any("tracked" in r.lower() for r in result["reasons"])) def test_foreign_worktree_blocked(self): report = _full_worktree_cleanup_report( **{"Removed worktree path": "/tmp/foreign-worktree"} ) result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) self.assertTrue(any("branches/" in r.lower() for r in result["reasons"])) def test_cleanup_skipped_with_blocker_passes(self): report = "\n".join([ "## Controller Handoff", "- Cleanup outcome: CLEANUP_SKIPPED", "- Cleanup blocker: gitea.branch.delete capability not resolved in active profile", ]) result = assess_post_merge_cleanup_proof(report) self.assertFalse(result["block"]) self.assertEqual(result["outcome"], CLEANUP_SKIPPED) def test_cleanup_skipped_without_blocker_blocked(self): report = "## Controller Handoff\n- Cleanup outcome: CLEANUP_SKIPPED" result = assess_post_merge_cleanup_proof(report) self.assertTrue(result["block"]) def test_full_remote_cleanup_passes(self): result = assess_post_merge_cleanup_proof(_full_remote_cleanup_report()) self.assertFalse(result["block"]) self.assertTrue(result["remote_delete_claimed"]) def test_full_worktree_cleanup_passes(self): result = assess_post_merge_cleanup_proof(_full_worktree_cleanup_report()) self.assertFalse(result["block"]) self.assertTrue(result["worktree_remove_claimed"]) def test_validator_integration_blocks_unproven_delete(self): report = ( "## Controller Handoff\n" "- Cleanup mutations: gitea_delete_branch deleted remote branch\n" ) result = assess_final_report_validator(report, task_kind="review_pr") self.assertTrue(result["blocked"]) rule_ids = [f["rule_id"] for f in result["findings"]] self.assertIn("reviewer.post_merge_cleanup_proof", rule_ids) def test_validator_integration_allows_skipped(self): report = "\n".join([ "## Controller Handoff", "- Cleanup outcome: CLEANUP_SKIPPED", "- Cleanup blocker: branch still referenced by open PR #414", ]) result = assess_final_report_validator(report, task_kind="review_pr") cleanup_findings = [ f for f in result["findings"] if f["rule_id"] == "reviewer.post_merge_cleanup_proof" ] self.assertEqual(cleanup_findings, []) if __name__ == "__main__": unittest.main()