"""Tests for session-owned worktree cleanup audit and TTL enforcement (#401).""" 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)") if __name__ == "__main__": unittest.main()