gitea_audit_worktree_cleanup had no PR linkage. Issue worktrees therefore
reported pr_number=null and classified as active_issue_work with
removable=false permanently, even once their PR was merged and the head was
already contained in master. Observed on live master 9301739910: 42
issue_work worktrees, 0 removable, 0 with pr_number populated, while
gitea_reconcile_merged_cleanups reported the same worktree safe to remove.
Two independent gaps caused it:
* build_worktree_metadata was never given a pr_number, and only open PRs were
fetched, so no owning-PR evidence existed at all.
* clean_stale_removable was unreachable for issue_work: it required
ttl_expired, derived from a last_used_at that nothing populates, and
is_ttl_expired fail-safes to False when the timestamp is unknown.
This adds deterministic merged-PR linkage and gates removal on the complete
cleanup policy:
* build_pr_index / resolve_owning_pr link a worktree branch to exactly one
owning PR. Competing PRs on one branch, a still-open owner, a head-branch
mismatch, or missing PR state all fail closed while still reporting the
resolved pr_number.
* assess_merged_pr_worktree_cleanup requires all of: conclusive merged
ownership, branch agreement, containment of the head in authoritative
master, no open/competing PR, no active lease, no issue lock, no live
session, a clean tree, and a non-protected checkout. Unknown state blocks.
* Containment reuses merged_cleanup_reconcile.is_head_ancestor_of_ref so the
audit and the PR-scoped reconciler agree on what "already landed" means.
Lease evidence is now supplied. audit_branches_directory already accepted
leased_branches but the MCP tool never passed it, so has_active_lease was
false for every worktree in a live run. That was inert only while issue
worktrees could never become removable; it is wired to authoritative
control-plane leases here, scoped so a lease on issue N protects that issue's
work worktree and not a baseline or review tree merely named after it.
Issue work no longer becomes removable on TTL age alone, since age is not
proof that a branch landed and would otherwise reclaim a worktree holding
unmerged commits. conflict_fix keeps its existing TTL behaviour, and review,
baseline, merge-simulation, detached, dirty, open-PR, and protected
classifications are unchanged.
The assessor still performs no deletion and gains no cleanup mutation. This
is assessor-side only and does not implement the PR-scoped executor or the
expired-lease reclaim policy tracked separately by #855.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
555 lines
20 KiB
Python
555 lines
20 KiB
Python
"""Tests for session-owned worktree cleanup audit, TTL, and integrity (#401, #404)."""
|
|
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import worktree_cleanup_audit as wca # noqa: E402
|
|
|
|
|
|
NOW = datetime(2026, 7, 7, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _iso(dt):
|
|
return dt.isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
class TestWorkflowTypeInference(unittest.TestCase):
|
|
def test_review_pr_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/review-pr376", "review-pr376"),
|
|
wca.WORKFLOW_REVIEW,
|
|
)
|
|
|
|
def test_baseline_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/baseline-master-issue-401"),
|
|
wca.WORKFLOW_BASELINE,
|
|
)
|
|
|
|
def test_merge_simulation_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/merge-sim-pr380"),
|
|
wca.WORKFLOW_MERGE_SIMULATION,
|
|
)
|
|
|
|
def test_issue_work_branch(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type(
|
|
"branches/issue-401-worktree", "feat/issue-401-worktree"
|
|
),
|
|
wca.WORKFLOW_ISSUE_WORK,
|
|
)
|
|
|
|
def test_conflict_fix_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/conflict-fix-pr376"),
|
|
wca.WORKFLOW_CONFLICT_FIX,
|
|
)
|
|
|
|
def test_unknown_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/scratchpad"), wca.WORKFLOW_UNKNOWN
|
|
)
|
|
|
|
|
|
class TestMetadata(unittest.TestCase):
|
|
def test_metadata_fields_present(self):
|
|
meta = wca.build_worktree_metadata(
|
|
path="branches/issue-401-worktree",
|
|
branch="feat/issue-401-worktree",
|
|
head_sha="abc123",
|
|
creator="jcwalker3",
|
|
profile="prgs-author",
|
|
created_at=_iso(NOW),
|
|
last_used_at=_iso(NOW),
|
|
)
|
|
for field in (
|
|
"path",
|
|
"workflow_type",
|
|
"issue_number",
|
|
"pr_number",
|
|
"branch",
|
|
"head_sha",
|
|
"creator",
|
|
"profile",
|
|
"created_at",
|
|
"last_used_at",
|
|
"cleanup_eligibility",
|
|
):
|
|
self.assertIn(field, meta)
|
|
self.assertEqual(meta["issue_number"], 401)
|
|
self.assertEqual(meta["workflow_type"], wca.WORKFLOW_ISSUE_WORK)
|
|
|
|
def test_review_metadata_auto_removable_flag(self):
|
|
meta = wca.build_worktree_metadata(path="branches/review-pr42")
|
|
self.assertTrue(meta["auto_remove_on_success"])
|
|
|
|
|
|
class TestTTL(unittest.TestCase):
|
|
def test_expired(self):
|
|
old = _iso(NOW - timedelta(hours=48))
|
|
self.assertTrue(wca.is_ttl_expired(last_used_at=old, now=NOW, ttl_hours=24))
|
|
|
|
def test_not_expired(self):
|
|
recent = _iso(NOW - timedelta(hours=1))
|
|
self.assertFalse(
|
|
wca.is_ttl_expired(last_used_at=recent, now=NOW, ttl_hours=24)
|
|
)
|
|
|
|
def test_unknown_timestamp_fails_safe(self):
|
|
self.assertFalse(wca.is_ttl_expired(last_used_at=None, now=NOW))
|
|
self.assertFalse(
|
|
wca.is_ttl_expired(last_used_at="not-a-date", now=NOW)
|
|
)
|
|
|
|
|
|
class TestClassification(unittest.TestCase):
|
|
def test_successful_review_cleanup(self):
|
|
# Scenario 1: clean review worktree -> removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW, is_dirty=False
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
|
|
self.assertTrue(wca.is_removable(cls))
|
|
|
|
def test_dirty_worktree_preserved(self):
|
|
# Scenario 3: dirty worktree is never removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW, is_dirty=True
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_active_pr_worktree_preserved(self):
|
|
# Scenario 4: open PR wins over an otherwise-removable review worktree.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW,
|
|
is_dirty=False,
|
|
has_open_pr=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_OPEN_PR)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_stale_clean_issue_worktree_needs_merged_pr_proof(self):
|
|
# Scenario 5 (#858): age is not proof that the branch landed, so a
|
|
# TTL-expired issue worktree stays active work. Only authoritative
|
|
# merged-PR evidence makes it removable, which is what keeps a
|
|
# worktree holding unmerged commits from being reclaimed by age.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
ttl_expired=True,
|
|
merged_pr_cleanup={"proven": True},
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
|
|
self.assertTrue(wca.is_removable(cls))
|
|
|
|
def test_stale_clean_conflict_fix_worktree_removable(self):
|
|
# conflict_fix keeps the original TTL rule; #858 changed issue work only.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_CONFLICT_FIX,
|
|
is_dirty=False,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
|
|
self.assertTrue(wca.is_removable(cls))
|
|
|
|
def test_fresh_clean_issue_worktree_preserved(self):
|
|
# Clean issue worktree not yet TTL-expired stays active.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
ttl_expired=False,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_detached_review_worktree_classified(self):
|
|
# Scenario 6: detached review worktree -> detached_review_leftover.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW,
|
|
is_dirty=False,
|
|
is_detached=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_DETACHED_REVIEW_LEFTOVER)
|
|
self.assertTrue(wca.is_removable(cls))
|
|
|
|
def test_ttl_expired_but_dirty_preserved(self):
|
|
# Scenario 7: dirty wins over TTL expiry.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=True,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_lease_protected_worktree_preserved(self):
|
|
# Scenario 8: active lease is never removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW,
|
|
is_dirty=False,
|
|
has_active_lease=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_active_issue_lock_preserved(self):
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
has_active_issue_lock=True,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_protected_base_worktree_never_removable(self):
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
is_protected=True,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_unknown_workflow_type_unsafe(self):
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_UNKNOWN, is_dirty=False, ttl_expired=True
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
|
|
class TestRemovalDecision(unittest.TestCase):
|
|
def test_safe_removal_proof(self):
|
|
decision = wca.assess_worktree_removal(
|
|
path="branches/review-pr42",
|
|
branch="review-pr42",
|
|
head_sha="abc123",
|
|
is_dirty=False,
|
|
has_open_pr=False,
|
|
has_active_lease=False,
|
|
classification=wca.CLASS_CLEAN_STALE_REMOVABLE,
|
|
)
|
|
self.assertTrue(decision["safe_to_remove"])
|
|
self.assertEqual(decision["block_reasons"], [])
|
|
self.assertTrue(decision["clean"])
|
|
self.assertTrue(decision["no_active_pr"])
|
|
self.assertTrue(decision["no_active_lease"])
|
|
|
|
def test_dirty_blocks_removal(self):
|
|
decision = wca.assess_worktree_removal(
|
|
path="branches/review-pr42",
|
|
branch="review-pr42",
|
|
head_sha="abc123",
|
|
is_dirty=True,
|
|
has_open_pr=False,
|
|
has_active_lease=False,
|
|
classification=wca.CLASS_DIRTY_LOCAL,
|
|
)
|
|
self.assertFalse(decision["safe_to_remove"])
|
|
self.assertIn("worktree has uncommitted changes", decision["block_reasons"])
|
|
|
|
def test_open_pr_and_lease_block_removal(self):
|
|
decision = wca.assess_worktree_removal(
|
|
path="branches/review-pr42",
|
|
branch="review-pr42",
|
|
head_sha="abc123",
|
|
is_dirty=False,
|
|
has_open_pr=True,
|
|
has_active_lease=True,
|
|
classification=wca.CLASS_ACTIVE_OPEN_PR,
|
|
)
|
|
self.assertFalse(decision["safe_to_remove"])
|
|
self.assertIn("worktree branch has an open PR", decision["block_reasons"])
|
|
self.assertIn("worktree has an active lease", decision["block_reasons"])
|
|
|
|
|
|
class TestSuccessCleanupPlan(unittest.TestCase):
|
|
def test_review_worktree_removed_at_success(self):
|
|
# Scenario 1 end-to-end: clean review worktree removed at success.
|
|
meta = wca.build_worktree_metadata(path="branches/review-pr42")
|
|
plan = wca.plan_success_cleanup(
|
|
metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
|
|
)
|
|
self.assertTrue(plan["remove"])
|
|
|
|
def test_failed_review_leaves_worktree_reported(self):
|
|
# Scenario 2: dirty review worktree preserved and reported at failure.
|
|
meta = wca.build_worktree_metadata(path="branches/review-pr42")
|
|
plan = wca.plan_success_cleanup(
|
|
metadata=meta, is_dirty=True, has_open_pr=False, has_active_lease=False
|
|
)
|
|
self.assertFalse(plan["remove"])
|
|
self.assertIn("uncommitted", plan["reason"])
|
|
report = wca.cleanup_failure_report(meta["path"], plan["reason"])
|
|
self.assertFalse(report["removed"])
|
|
self.assertEqual(report["path"], "branches/review-pr42")
|
|
|
|
def test_issue_worktree_preserved_by_policy(self):
|
|
meta = wca.build_worktree_metadata(
|
|
path="branches/issue-401-worktree", branch="feat/issue-401-worktree"
|
|
)
|
|
plan = wca.plan_success_cleanup(
|
|
metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
|
|
)
|
|
self.assertFalse(plan["remove"])
|
|
self.assertIn("preserved by policy", plan["reason"])
|
|
|
|
|
|
class TestPorcelainParser(unittest.TestCase):
|
|
def test_parse_branch_and_detached(self):
|
|
text = (
|
|
"worktree /repo\n"
|
|
"HEAD 1111111111111111111111111111111111111111\n"
|
|
"branch refs/heads/master\n"
|
|
"\n"
|
|
"worktree /repo/branches/review-pr42\n"
|
|
"HEAD 2222222222222222222222222222222222222222\n"
|
|
"detached\n"
|
|
)
|
|
entries = wca.parse_worktree_porcelain(text)
|
|
self.assertEqual(len(entries), 2)
|
|
self.assertEqual(entries[0]["branch"], "master")
|
|
self.assertFalse(entries[0]["detached"])
|
|
self.assertIsNone(entries[1]["branch"])
|
|
self.assertTrue(entries[1]["detached"])
|
|
|
|
|
|
class TestAuditReportAccuracy(unittest.TestCase):
|
|
"""Scenario 9: cleanup report accuracy over a mixed branches/ directory."""
|
|
|
|
PORCELAIN = (
|
|
"worktree /repo\n"
|
|
"HEAD 1111111111111111111111111111111111111111\n"
|
|
"branch refs/heads/master\n"
|
|
"\n"
|
|
"worktree /repo/branches/review-pr42\n"
|
|
"HEAD 2222222222222222222222222222222222222222\n"
|
|
"branch refs/heads/review-pr42\n"
|
|
"\n"
|
|
"worktree /repo/branches/issue-400-open-pr\n"
|
|
"HEAD 3333333333333333333333333333333333333333\n"
|
|
"branch refs/heads/feat/issue-400-open-pr\n"
|
|
"\n"
|
|
"worktree /repo/branches/issue-401-dirty\n"
|
|
"HEAD 4444444444444444444444444444444444444444\n"
|
|
"branch refs/heads/feat/issue-401-dirty\n"
|
|
)
|
|
|
|
def _fake_dirty(self, path):
|
|
if path.endswith("issue-401-dirty"):
|
|
return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]}
|
|
return {"exists": True, "dirty": False, "dirty_files": []}
|
|
|
|
def test_mixed_directory_classified(self):
|
|
with patch.object(
|
|
wca,
|
|
"list_worktrees",
|
|
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
|
|
), patch.object(
|
|
wca, "read_worktree_dirty", side_effect=self._fake_dirty
|
|
), patch.object(
|
|
wca, "git_worktree_list", return_value="(mocked)"
|
|
):
|
|
report = wca.audit_branches_directory(
|
|
"/repo",
|
|
open_pr_branches={"feat/issue-400-open-pr"},
|
|
)
|
|
|
|
by_path = {wt["path"]: wt for wt in report["worktrees"]}
|
|
# main checkout on master -> protected -> unsafe/unknown, not removable
|
|
self.assertEqual(
|
|
by_path["/repo"]["classification"], wca.CLASS_UNSAFE_UNKNOWN
|
|
)
|
|
# clean review worktree -> removable
|
|
self.assertEqual(
|
|
by_path["/repo/branches/review-pr42"]["classification"],
|
|
wca.CLASS_CLEAN_STALE_REMOVABLE,
|
|
)
|
|
# open PR branch -> preserved
|
|
self.assertEqual(
|
|
by_path["/repo/branches/issue-400-open-pr"]["classification"],
|
|
wca.CLASS_ACTIVE_OPEN_PR,
|
|
)
|
|
# dirty worktree -> preserved
|
|
self.assertEqual(
|
|
by_path["/repo/branches/issue-401-dirty"]["classification"],
|
|
wca.CLASS_DIRTY_LOCAL,
|
|
)
|
|
# exactly one removable candidate (the clean review worktree)
|
|
self.assertEqual(report["removable_count"], 1)
|
|
self.assertEqual(
|
|
report["removable_candidates"][0]["path"],
|
|
"/repo/branches/review-pr42",
|
|
)
|
|
self.assertEqual(report["total"], 4)
|
|
self.assertEqual(report["git_worktree_list"], "(mocked)")
|
|
|
|
|
|
def _integrity_entry(
|
|
path: str,
|
|
*,
|
|
classification: str,
|
|
registered: bool = True,
|
|
preserve: bool | None = None,
|
|
) -> dict:
|
|
preserve_flag = preserve if preserve is not None else classification in {
|
|
"active_open_pr",
|
|
"active_issue_work",
|
|
"dirty_local_worktree",
|
|
"unsafe_unknown",
|
|
}
|
|
return {
|
|
"path": path,
|
|
"classification": classification,
|
|
"preserve": preserve_flag,
|
|
"registered_worktree": registered,
|
|
"worktree_state": {"exists": True, "clean": classification == "clean_stale_removable"},
|
|
"worktree_record": {"branch": "feat/x"} if registered else None,
|
|
}
|
|
|
|
|
|
def _integrity_snapshot(entries: list[dict]) -> dict:
|
|
return {"entries": entries}
|
|
|
|
|
|
class TestParseWorktreePorcelain(unittest.TestCase):
|
|
def test_parses_multiple_worktrees(self):
|
|
text = "\n".join([
|
|
"worktree /proj/branches/foo",
|
|
"HEAD abcdef0123456789abcdef0123456789abcdef0",
|
|
"branch refs/heads/feat/foo",
|
|
"",
|
|
"worktree /proj",
|
|
"HEAD 1111111111111111111111111111111111111111",
|
|
"branch refs/heads/master",
|
|
])
|
|
parsed = wca.parse_worktree_list_porcelain(text)
|
|
self.assertEqual(len(parsed), 2)
|
|
self.assertEqual(parsed[0]["branch"], "feat/foo")
|
|
|
|
|
|
class TestClassifyEntry(unittest.TestCase):
|
|
def test_active_pr_classification(self):
|
|
result = wca.classify_branches_entry(
|
|
rel_path="branches/feat-issue-1-x",
|
|
worktree_record={"branch": "feat/issue-1-x"},
|
|
worktree_state={"exists": True, "clean": True, "dirty_files": []},
|
|
open_pr_branches={"feat/issue-1-x"},
|
|
)
|
|
self.assertEqual(result, "active_open_pr")
|
|
|
|
def test_dirty_classification(self):
|
|
result = wca.classify_branches_entry(
|
|
rel_path="branches/dirty-one",
|
|
worktree_record={"branch": "feat/dirty-one"},
|
|
worktree_state={"exists": True, "dirty_files": ["a.py"]},
|
|
open_pr_branches=set(),
|
|
)
|
|
self.assertEqual(result, "dirty_local_worktree")
|
|
|
|
|
|
class TestCleanupIntegrity(unittest.TestCase):
|
|
def test_preserved_worktree_remains(self):
|
|
path = "branches/keep-me"
|
|
before = _integrity_snapshot([
|
|
_integrity_entry(path, classification="clean_stale_removable", preserve=False),
|
|
])
|
|
after = _integrity_snapshot([
|
|
_integrity_entry(path, classification="clean_stale_removable", preserve=False),
|
|
])
|
|
result = wca.assess_worktree_cleanup_integrity(before=before, after=after)
|
|
self.assertTrue(result["integrity_passed"])
|
|
|
|
def test_intentional_removal_passes(self):
|
|
path = "branches/remove-me"
|
|
before = _integrity_snapshot([
|
|
_integrity_entry(path, classification="clean_stale_removable", preserve=False),
|
|
])
|
|
after = _integrity_snapshot([])
|
|
result = wca.assess_worktree_cleanup_integrity(
|
|
before=before,
|
|
after=after,
|
|
removals=[{
|
|
"path": path,
|
|
"method": "git worktree remove",
|
|
"pre_removal_proof": "clean status",
|
|
}],
|
|
)
|
|
self.assertTrue(result["integrity_passed"])
|
|
|
|
def test_dirty_worktree_disappears_fails(self):
|
|
path = "branches/dirty-wt"
|
|
before = _integrity_snapshot([_integrity_entry(path, classification="dirty_local_worktree")])
|
|
after = _integrity_snapshot([])
|
|
recon = wca.reconcile_cleanup_audit(before, after)
|
|
result = wca.assess_cleanup_audit_integrity(recon)
|
|
self.assertFalse(result["proven"])
|
|
|
|
def test_active_pr_worktree_disappears_fails(self):
|
|
path = "branches/review-pr99"
|
|
before = _integrity_snapshot([_integrity_entry(path, classification="active_open_pr")])
|
|
after = _integrity_snapshot([])
|
|
result = wca.assess_worktree_cleanup_integrity(before=before, after=after)
|
|
self.assertFalse(result["integrity_passed"])
|
|
|
|
def test_explained_missing_allowed(self):
|
|
path = "branches/review-pr382"
|
|
before = _integrity_snapshot([
|
|
_integrity_entry(path, classification="detached_review_leftover", preserve=False),
|
|
])
|
|
after = _integrity_snapshot([])
|
|
result = wca.assess_worktree_cleanup_integrity(
|
|
before=before,
|
|
after=after,
|
|
explained_missing={path: "removed concurrently by sibling session"},
|
|
)
|
|
self.assertTrue(result["integrity_passed"])
|
|
|
|
def test_removal_log_omits_clean_stale_fails(self):
|
|
path = "branches/a"
|
|
before = _integrity_snapshot([
|
|
_integrity_entry(path, classification="clean_stale_removable", preserve=False),
|
|
])
|
|
after = _integrity_snapshot([])
|
|
recon = wca.reconcile_cleanup_audit(before, after, removal_log=[])
|
|
self.assertFalse(recon["removal_log_complete"])
|
|
|
|
|
|
class TestCleanupReportProof(unittest.TestCase):
|
|
def test_complete_report_passes(self):
|
|
report = "\n".join([
|
|
"Cleanup audit reconciliation table:",
|
|
"Initial count: 10",
|
|
"Removed count: 3",
|
|
"Preserved count: 7",
|
|
"Missing-unexplained count: 0",
|
|
"Final count: 7",
|
|
"Final verification: git worktree list proof attached",
|
|
])
|
|
result = wca.assess_cleanup_audit_final_report(report)
|
|
self.assertTrue(result["proven"])
|
|
|
|
def test_incomplete_report_fails(self):
|
|
result = wca.assess_cleanup_audit_final_report("removed some worktrees")
|
|
self.assertFalse(result["proven"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|