Merge branch 'master' into fix/issue-850-native-mcp-bootstrap
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,551 @@
|
||||
"""Merged-PR awareness for the worktree cleanup audit (#858).
|
||||
|
||||
Before #858 an ``issue_work`` worktree could never leave ``active_issue_work``:
|
||||
the audit had no PR linkage at all (``pr_number`` was structurally ``None``)
|
||||
and its only route to ``clean_stale_removable`` was a TTL derived from a
|
||||
``last_used_at`` that nothing ever populated. A merged, clean, unprotected
|
||||
worktree was therefore reported as active work forever, disagreeing with the
|
||||
PR-scoped reconciler.
|
||||
|
||||
These tests use fabricated temporary repositories and synthetic PR records
|
||||
only. Nothing here removes a worktree or deletes a branch.
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import merged_cleanup_reconcile as mcr # noqa: E402
|
||||
import worktree_cleanup_audit as wca # noqa: E402
|
||||
|
||||
|
||||
MERGED_BRANCH = "feat/issue-777-timeline"
|
||||
MERGED_PATH = "/repo/branches/issue-777-timeline"
|
||||
HEAD_SHA = "a" * 40
|
||||
|
||||
|
||||
def _pr(number, branch, *, merged=True, sha=HEAD_SHA, state=None):
|
||||
"""Synthetic Gitea PR payload."""
|
||||
return {
|
||||
"number": number,
|
||||
"head": {"ref": branch, "sha": sha},
|
||||
"merged_at": "2026-07-24T01:00:00Z" if merged else None,
|
||||
"state": state or ("closed" if merged else "open"),
|
||||
}
|
||||
|
||||
|
||||
def _porcelain(*entries):
|
||||
out = []
|
||||
for path, branch, sha in entries:
|
||||
out.append(f"worktree {path}")
|
||||
out.append(f"HEAD {sha}")
|
||||
if branch is None:
|
||||
out.append("detached")
|
||||
else:
|
||||
out.append(f"branch refs/heads/{branch}")
|
||||
out.append("")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
class _AuditHarness(unittest.TestCase):
|
||||
"""Runs audit_branches_directory over a fabricated worktree listing."""
|
||||
|
||||
PORCELAIN = _porcelain(
|
||||
("/repo", "master", "f" * 40),
|
||||
(MERGED_PATH, MERGED_BRANCH, HEAD_SHA),
|
||||
)
|
||||
|
||||
def run_audit(self, *, dirty_paths=(), contained=True, **kwargs):
|
||||
def fake_dirty(path):
|
||||
if path in dirty_paths:
|
||||
return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]}
|
||||
return {"exists": True, "dirty": False, "dirty_files": []}
|
||||
|
||||
with patch.object(
|
||||
wca, "list_worktrees",
|
||||
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
|
||||
), patch.object(
|
||||
wca, "read_worktree_dirty", side_effect=fake_dirty
|
||||
), patch.object(
|
||||
wca, "git_worktree_list", return_value="(mocked)"
|
||||
), patch.object(
|
||||
wca, "is_head_ancestor_of_ref", return_value=contained
|
||||
):
|
||||
report = wca.audit_branches_directory("/repo", **kwargs)
|
||||
return {wt["path"]: wt for wt in report["worktrees"]}, report
|
||||
|
||||
def merged_audit(self, **kwargs):
|
||||
kwargs.setdefault("pr_index", wca.build_pr_index([_pr(849, MERGED_BRANCH)]))
|
||||
kwargs.setdefault("master_ref", "prgs/master")
|
||||
return self.run_audit(**kwargs)
|
||||
|
||||
|
||||
class TestMergedWorktreeBecomesRemovable(_AuditHarness):
|
||||
def test_clean_merged_issue_worktree_is_linked_and_removable(self):
|
||||
by_path, report = self.merged_audit()
|
||||
entry = by_path[MERGED_PATH]
|
||||
|
||||
self.assertEqual(entry["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE)
|
||||
self.assertTrue(entry["removable"])
|
||||
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_MERGED)
|
||||
self.assertEqual(entry["merged_pr_cleanup"]["block_reasons"], [])
|
||||
self.assertIn(MERGED_PATH, [c["path"] for c in report["removable_candidates"]])
|
||||
|
||||
def test_pr_number_populated_from_authoritative_linkage(self):
|
||||
by_path, _ = self.merged_audit()
|
||||
self.assertEqual(by_path[MERGED_PATH]["pr_number"], 849)
|
||||
|
||||
def test_regression_without_pr_evidence_stays_active_issue_work(self):
|
||||
"""The pre-#858 behaviour, still correct when no PR state is supplied."""
|
||||
by_path, _ = self.run_audit()
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
|
||||
self.assertFalse(entry["removable"])
|
||||
self.assertIsNone(entry["pr_number"])
|
||||
|
||||
|
||||
class TestProtectiveSignalsSurvive(_AuditHarness):
|
||||
def test_open_pr_worktree_is_not_removable(self):
|
||||
index = wca.build_pr_index([_pr(900, MERGED_BRANCH, merged=False)])
|
||||
by_path, _ = self.run_audit(
|
||||
pr_index=index,
|
||||
master_ref="prgs/master",
|
||||
open_pr_branches={MERGED_BRANCH},
|
||||
)
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_OPEN_PR)
|
||||
self.assertFalse(entry["removable"])
|
||||
# linkage still reports the owning PR, it just is not merge proof
|
||||
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_OPEN)
|
||||
self.assertEqual(entry["pr_number"], 900)
|
||||
|
||||
def test_dirty_tracked_worktree_is_not_removable(self):
|
||||
by_path, _ = self.merged_audit(dirty_paths=(MERGED_PATH,))
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL)
|
||||
self.assertFalse(entry["removable"])
|
||||
self.assertIn(
|
||||
"worktree has uncommitted changes",
|
||||
entry["merged_pr_cleanup"]["block_reasons"],
|
||||
)
|
||||
|
||||
def test_untracked_only_worktree_is_not_removable(self):
|
||||
"""``git status --porcelain`` reports untracked files as dirty too."""
|
||||
def untracked(path):
|
||||
if path == MERGED_PATH:
|
||||
return {"exists": True, "dirty": True, "dirty_files": ["?? scratch.txt"]}
|
||||
return {"exists": True, "dirty": False, "dirty_files": []}
|
||||
|
||||
with patch.object(
|
||||
wca, "list_worktrees",
|
||||
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
|
||||
), patch.object(
|
||||
wca, "read_worktree_dirty", side_effect=untracked
|
||||
), patch.object(
|
||||
wca, "git_worktree_list", return_value="(mocked)"
|
||||
), patch.object(
|
||||
wca, "is_head_ancestor_of_ref", return_value=True
|
||||
):
|
||||
report = wca.audit_branches_directory(
|
||||
"/repo",
|
||||
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
|
||||
master_ref="prgs/master",
|
||||
)
|
||||
entry = {wt["path"]: wt for wt in report["worktrees"]}[MERGED_PATH]
|
||||
self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_active_lease_by_issue_number_is_protective(self):
|
||||
by_path, _ = self.merged_audit(leased_issue_numbers={777})
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertTrue(entry["has_active_lease"])
|
||||
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_active_lease_by_branch_is_protective(self):
|
||||
by_path, _ = self.merged_audit(leased_branches={MERGED_BRANCH})
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertTrue(entry["has_active_lease"])
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_active_issue_lock_is_protective(self):
|
||||
by_path, _ = self.merged_audit(active_issue_branches={MERGED_BRANCH})
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertTrue(entry["has_active_issue_lock"])
|
||||
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_live_session_worktree_is_protective(self):
|
||||
by_path, _ = self.merged_audit(live_session_paths={MERGED_PATH})
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertTrue(entry["has_live_session"])
|
||||
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_head_not_contained_in_master_is_not_removable(self):
|
||||
by_path, _ = self.merged_audit(contained=False)
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
|
||||
self.assertFalse(entry["removable"])
|
||||
self.assertIn(
|
||||
"worktree head is not contained in authoritative master "
|
||||
"(unmerged commits remain)",
|
||||
entry["merged_pr_cleanup"]["block_reasons"],
|
||||
)
|
||||
|
||||
def test_unknown_containment_fails_closed(self):
|
||||
by_path, _ = self.merged_audit(contained=None)
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertFalse(entry["removable"])
|
||||
self.assertIn(
|
||||
"containment of the worktree head in master is unknown",
|
||||
entry["merged_pr_cleanup"]["block_reasons"],
|
||||
)
|
||||
|
||||
def test_missing_master_ref_fails_closed(self):
|
||||
by_path, _ = self.run_audit(
|
||||
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)])
|
||||
)
|
||||
self.assertFalse(by_path[MERGED_PATH]["removable"])
|
||||
|
||||
def test_unmerged_owning_pr_is_not_removable(self):
|
||||
index = wca.build_pr_index([_pr(901, MERGED_BRANCH, merged=False)])
|
||||
by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master")
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertFalse(entry["removable"])
|
||||
self.assertIn(
|
||||
"owning PR #901 is not merged",
|
||||
entry["merged_pr_cleanup"]["block_reasons"],
|
||||
)
|
||||
|
||||
def test_control_checkout_is_never_removable(self):
|
||||
by_path, _ = self.merged_audit()
|
||||
control = by_path["/repo"]
|
||||
self.assertTrue(control["is_protected"])
|
||||
self.assertEqual(control["classification"], wca.CLASS_UNSAFE_UNKNOWN)
|
||||
self.assertFalse(control["removable"])
|
||||
|
||||
def test_control_checkout_not_removable_even_if_linked_and_merged(self):
|
||||
"""A merged PR on the control checkout must not unlock removal."""
|
||||
porcelain = _porcelain(("/repo", MERGED_BRANCH, HEAD_SHA))
|
||||
with patch.object(
|
||||
wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain)
|
||||
), patch.object(
|
||||
wca, "read_worktree_dirty",
|
||||
return_value={"exists": True, "dirty": False, "dirty_files": []},
|
||||
), patch.object(
|
||||
wca, "git_worktree_list", return_value="(mocked)"
|
||||
), patch.object(
|
||||
wca, "is_head_ancestor_of_ref", return_value=True
|
||||
):
|
||||
report = wca.audit_branches_directory(
|
||||
"/repo",
|
||||
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
|
||||
master_ref="prgs/master",
|
||||
)
|
||||
entry = report["worktrees"][0]
|
||||
self.assertEqual(entry["classification"], wca.CLASS_UNSAFE_UNKNOWN)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
|
||||
class TestAmbiguousLinkageFailsClosed(_AuditHarness):
|
||||
def test_competing_prs_on_one_branch_fail_closed(self):
|
||||
index = wca.build_pr_index(
|
||||
[_pr(849, MERGED_BRANCH), _pr(860, MERGED_BRANCH)]
|
||||
)
|
||||
by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master")
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS)
|
||||
self.assertIsNone(entry["pr_number"])
|
||||
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_merged_plus_open_pr_on_one_branch_fails_closed(self):
|
||||
index = wca.build_pr_index(
|
||||
[_pr(849, MERGED_BRANCH), _pr(861, MERGED_BRANCH, merged=False)]
|
||||
)
|
||||
by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master")
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_no_owning_pr_fails_closed(self):
|
||||
by_path, _ = self.run_audit(
|
||||
pr_index=wca.build_pr_index([_pr(849, "feat/other-branch")]),
|
||||
master_ref="prgs/master",
|
||||
)
|
||||
entry = by_path[MERGED_PATH]
|
||||
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_NONE)
|
||||
self.assertFalse(entry["removable"])
|
||||
|
||||
def test_malformed_pr_records_are_dropped_not_guessed(self):
|
||||
index = wca.build_pr_index(
|
||||
[
|
||||
{"number": None, "head": {"ref": MERGED_BRANCH}},
|
||||
{"number": 5, "head": {}},
|
||||
{"number": "not-an-int", "head": {"ref": MERGED_BRANCH}},
|
||||
]
|
||||
)
|
||||
self.assertEqual(index, {})
|
||||
self.assertEqual(
|
||||
wca.resolve_owning_pr(branch=MERGED_BRANCH, pr_index=index)["status"],
|
||||
wca.LINKAGE_NONE,
|
||||
)
|
||||
|
||||
def test_detached_worktree_has_no_branch_linkage(self):
|
||||
self.assertEqual(
|
||||
wca.resolve_owning_pr(branch=None, pr_index={})["status"],
|
||||
wca.LINKAGE_UNKNOWN,
|
||||
)
|
||||
|
||||
|
||||
class TestUnrelatedClassificationsUnchanged(unittest.TestCase):
|
||||
"""Non-issue_work worktrees keep their pre-#858 classifications."""
|
||||
|
||||
PORCELAIN = _porcelain(
|
||||
("/repo", "master", "f" * 40),
|
||||
("/repo/branches/review-pr42", "review-pr42", "2" * 40),
|
||||
("/repo/branches/baseline-master-x", "baseline-master-x", "3" * 40),
|
||||
("/repo/branches/conflict-fix-pr50", "conflict-fix-pr50", "4" * 40),
|
||||
("/repo/branches/review-pr99", None, "5" * 40),
|
||||
)
|
||||
|
||||
def _audit(self, **kwargs):
|
||||
with patch.object(
|
||||
wca, "list_worktrees",
|
||||
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
|
||||
), patch.object(
|
||||
wca, "read_worktree_dirty",
|
||||
return_value={"exists": True, "dirty": False, "dirty_files": []},
|
||||
), patch.object(
|
||||
wca, "git_worktree_list", return_value="(mocked)"
|
||||
), patch.object(
|
||||
wca, "is_head_ancestor_of_ref", return_value=True
|
||||
):
|
||||
report = wca.audit_branches_directory("/repo", **kwargs)
|
||||
return {wt["path"]: wt for wt in report["worktrees"]}
|
||||
|
||||
def test_classifications_identical_with_and_without_pr_evidence(self):
|
||||
without = self._audit()
|
||||
with_evidence = self._audit(
|
||||
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
|
||||
master_ref="prgs/master",
|
||||
)
|
||||
self.assertEqual(
|
||||
{p: e["classification"] for p, e in without.items()},
|
||||
{p: e["classification"] for p, e in with_evidence.items()},
|
||||
)
|
||||
|
||||
def test_lease_on_issue_does_not_capture_similarly_named_scratch_trees(self):
|
||||
"""A lease on issue 777 protects issue work, not baseline/review trees."""
|
||||
porcelain = _porcelain(
|
||||
("/repo/branches/baseline-master-issue-777", "baseline-issue-777", "7" * 40),
|
||||
("/repo/branches/issue-777-timeline", MERGED_BRANCH, HEAD_SHA),
|
||||
)
|
||||
with patch.object(
|
||||
wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain)
|
||||
), patch.object(
|
||||
wca, "read_worktree_dirty",
|
||||
return_value={"exists": True, "dirty": False, "dirty_files": []},
|
||||
), patch.object(
|
||||
wca, "git_worktree_list", return_value="(mocked)"
|
||||
), patch.object(
|
||||
wca, "is_head_ancestor_of_ref", return_value=True
|
||||
):
|
||||
report = wca.audit_branches_directory(
|
||||
"/repo",
|
||||
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
|
||||
master_ref="prgs/master",
|
||||
leased_issue_numbers={777},
|
||||
)
|
||||
by_path = {wt["path"]: wt for wt in report["worktrees"]}
|
||||
|
||||
baseline = by_path["/repo/branches/baseline-master-issue-777"]
|
||||
self.assertFalse(baseline["has_active_lease"])
|
||||
self.assertEqual(baseline["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE)
|
||||
|
||||
issue_work = by_path["/repo/branches/issue-777-timeline"]
|
||||
self.assertTrue(issue_work["has_active_lease"])
|
||||
self.assertFalse(issue_work["removable"])
|
||||
|
||||
def test_review_and_baseline_still_removable(self):
|
||||
by_path = self._audit(
|
||||
pr_index=wca.build_pr_index([]), master_ref="prgs/master"
|
||||
)
|
||||
self.assertEqual(
|
||||
by_path["/repo/branches/review-pr42"]["classification"],
|
||||
wca.CLASS_CLEAN_STALE_REMOVABLE,
|
||||
)
|
||||
self.assertEqual(
|
||||
by_path["/repo/branches/baseline-master-x"]["classification"],
|
||||
wca.CLASS_CLEAN_STALE_REMOVABLE,
|
||||
)
|
||||
self.assertEqual(
|
||||
by_path["/repo/branches/review-pr99"]["classification"],
|
||||
wca.CLASS_DETACHED_REVIEW_LEFTOVER,
|
||||
)
|
||||
|
||||
def test_conflict_fix_ttl_behaviour_unchanged(self):
|
||||
"""conflict_fix still needs only TTL expiry; #858 did not touch it."""
|
||||
self.assertEqual(
|
||||
wca.classify_worktree(
|
||||
workflow_type=wca.WORKFLOW_CONFLICT_FIX,
|
||||
is_dirty=False,
|
||||
ttl_expired=True,
|
||||
),
|
||||
wca.CLASS_CLEAN_STALE_REMOVABLE,
|
||||
)
|
||||
self.assertEqual(
|
||||
wca.classify_worktree(
|
||||
workflow_type=wca.WORKFLOW_CONFLICT_FIX,
|
||||
is_dirty=False,
|
||||
ttl_expired=False,
|
||||
),
|
||||
wca.CLASS_ACTIVE_ISSUE_WORK,
|
||||
)
|
||||
|
||||
def test_issue_work_ttl_alone_no_longer_grants_removal(self):
|
||||
"""Age is not landing proof: TTL alone must not reclaim issue work."""
|
||||
self.assertEqual(
|
||||
wca.classify_worktree(
|
||||
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
||||
is_dirty=False,
|
||||
ttl_expired=True,
|
||||
),
|
||||
wca.CLASS_ACTIVE_ISSUE_WORK,
|
||||
)
|
||||
|
||||
|
||||
class TestAssessorPerformsNoDeletion(_AuditHarness):
|
||||
def test_audit_never_removes_a_worktree(self):
|
||||
with patch.object(wca, "remove_worktree") as removal:
|
||||
self.merged_audit()
|
||||
removal.assert_not_called()
|
||||
|
||||
def test_audit_shells_out_to_no_destructive_git_command(self):
|
||||
seen = []
|
||||
real_run = subprocess.run
|
||||
|
||||
def recording_run(cmd, *args, **kwargs):
|
||||
seen.append(cmd)
|
||||
return real_run(["true"], *args, **kwargs)
|
||||
|
||||
with patch.object(subprocess, "run", side_effect=recording_run):
|
||||
wca.audit_branches_directory("/nonexistent-repo-for-audit")
|
||||
|
||||
joined = [" ".join(c) if isinstance(c, list) else str(c) for c in seen]
|
||||
for cmd in joined:
|
||||
self.assertNotIn("worktree remove", cmd)
|
||||
self.assertNotIn("branch -D", cmd)
|
||||
self.assertNotIn("push", cmd)
|
||||
|
||||
|
||||
class TestAgreementWithPrScopedReconciler(unittest.TestCase):
|
||||
"""The audit and merged_cleanup_reconcile must agree on identical input.
|
||||
|
||||
Uses a real throwaway git repository so containment is computed by git
|
||||
rather than asserted. Nothing outside the temporary directory is touched.
|
||||
"""
|
||||
|
||||
def _git(self, *args):
|
||||
subprocess.run(
|
||||
["git", "-C", self.root, *args],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.root = os.path.realpath(self._tmp.name)
|
||||
self._git("init", "-b", "master", ".")
|
||||
self._git("config", "user.email", "[email protected]")
|
||||
self._git("config", "user.name", "Test")
|
||||
with open(os.path.join(self.root, "seed.txt"), "w") as fh:
|
||||
fh.write("seed\n")
|
||||
self._git("add", "seed.txt")
|
||||
self._git("commit", "-m", "seed")
|
||||
|
||||
self.branch = "feat/issue-777-timeline"
|
||||
self._git("checkout", "-b", self.branch)
|
||||
with open(os.path.join(self.root, "feature.txt"), "w") as fh:
|
||||
fh.write("feature\n")
|
||||
self._git("add", "feature.txt")
|
||||
self._git("commit", "-m", "feature")
|
||||
self.head_sha = subprocess.run(
|
||||
["git", "-C", self.root, "rev-parse", "HEAD"],
|
||||
capture_output=True, text=True, check=True,
|
||||
).stdout.strip()
|
||||
self._git("checkout", "master")
|
||||
self._git("merge", "--no-ff", "-m", "merge feature", self.branch)
|
||||
|
||||
self.worktree = os.path.join(self.root, "branches", "issue-777-timeline")
|
||||
self._git("worktree", "add", self.worktree, self.branch)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def _pr_index(self):
|
||||
return wca.build_pr_index(
|
||||
[
|
||||
{
|
||||
"number": 849,
|
||||
"head": {"ref": self.branch, "sha": self.head_sha},
|
||||
"merged_at": "2026-07-24T01:00:00Z",
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
def _audit_entry(self):
|
||||
report = wca.audit_branches_directory(
|
||||
self.root, pr_index=self._pr_index(), master_ref="master"
|
||||
)
|
||||
return next(wt for wt in report["worktrees"] if wt["path"] == self.worktree)
|
||||
|
||||
def _reconciler_entry(self):
|
||||
return mcr.assess_local_worktree_cleanup(
|
||||
pr_number=849,
|
||||
head_branch=self.branch,
|
||||
merged=True,
|
||||
worktree_state=mcr.resolve_cleanup_worktree_state(
|
||||
project_root=self.root,
|
||||
head_branch=self.branch,
|
||||
issue_number=777,
|
||||
pr_head_sha=self.head_sha,
|
||||
target_ref="master",
|
||||
),
|
||||
active_lock=False,
|
||||
)
|
||||
|
||||
def test_both_assessors_agree_the_worktree_is_safe(self):
|
||||
audit_entry = self._audit_entry()
|
||||
reconciler = self._reconciler_entry()
|
||||
|
||||
self.assertTrue(reconciler["safe_to_remove_worktree"], reconciler)
|
||||
self.assertTrue(audit_entry["removable"], audit_entry)
|
||||
self.assertEqual(audit_entry["pr_number"], reconciler["pr_number"])
|
||||
self.assertEqual(audit_entry["merged_pr_cleanup"]["block_reasons"], [])
|
||||
self.assertEqual(reconciler["block_reasons"], [])
|
||||
|
||||
def test_both_assessors_agree_a_dirty_worktree_is_unsafe(self):
|
||||
with open(os.path.join(self.worktree, "feature.txt"), "a") as fh:
|
||||
fh.write("local edit\n")
|
||||
|
||||
audit_entry = self._audit_entry()
|
||||
reconciler = self._reconciler_entry()
|
||||
|
||||
self.assertFalse(audit_entry["removable"])
|
||||
self.assertFalse(reconciler["safe_to_remove_worktree"])
|
||||
|
||||
def test_worktree_still_present_after_audit(self):
|
||||
self._audit_entry()
|
||||
self.assertTrue(os.path.isdir(self.worktree))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -134,13 +134,35 @@ class TestClassification(unittest.TestCase):
|
||||
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.
|
||||
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))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user