feat: require post-merge cleanup safety-gate proof in reviewer reports (Closes #402)
Adds post_merge_cleanup_proof validation so cleanup claims must carry the branch deletion and worktree removal checklists, or report CLEANUP_SKIPPED with an exact blocker. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
@@ -0,0 +1,169 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user