sysadmin
e27b6ddefb
Reconcile before/after branches/ snapshots so preserved worktrees cannot disappear without removal logs or explained state transitions. - worktree_cleanup_audit.py: snapshot capture, disposition reconciliation - gitea_capture_branches_worktree_snapshot, gitea_assess_worktree_cleanup_integrity - Final-report rule author.worktree_cleanup_audit_proof - worktree-cleanup.md bulk audit section - tests/test_worktree_cleanup_audit.py (11 cases)
533 lines
19 KiB
Python
533 lines
19 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_removable(self):
|
|
# Scenario 5: clean issue worktree, TTL expired, no lock -> removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
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()
|