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