Compare commits

..
3 Commits
2 changed files with 85 additions and 5 deletions
+19 -3
View File
@@ -90,6 +90,17 @@ def list_local_worktrees(project_root: str) -> list[dict[str, Any]]:
return entries return entries
def discover_local_worktrees(project_root: str) -> dict[str, str]:
"""Return a mapping from branch name to absolute worktree path (#528)."""
mapping: dict[str, str] = {}
for entry in list_local_worktrees(project_root):
branch = entry.get("branch")
path = entry.get("path")
if branch and path:
mapping[str(branch)] = str(path)
return mapping
def _is_under_branches(project_root: str, path: str) -> bool: def _is_under_branches(project_root: str, path: str) -> bool:
branches_root = os.path.realpath(os.path.join(project_root, "branches")) branches_root = os.path.realpath(os.path.join(project_root, "branches"))
real_path = os.path.realpath(path) real_path = os.path.realpath(path)
@@ -148,12 +159,15 @@ def resolve_cleanup_worktree_state(
issue_match = wanted_issue is not None and entry_issue == wanted_issue issue_match = wanted_issue is not None and entry_issue == wanted_issue
if not (branch_match or issue_match): if not (branch_match or issue_match):
continue continue
if not _head_is_safe_for_cleanup( safe_head = _head_is_safe_for_cleanup(
project_root=project_root, project_root=project_root,
candidate_head=entry.get("head_sha"), candidate_head=entry.get("head_sha"),
pr_head_sha=pr_head_sha, pr_head_sha=pr_head_sha,
target_ref=target_ref, target_ref=target_ref,
): )
if not safe_head and branch_match and not pr_head_sha and not target_ref:
safe_head = True
if not safe_head:
unsafe_matches.append(path) unsafe_matches.append(path)
continue continue
candidates.append(entry) candidates.append(entry)
@@ -165,7 +179,9 @@ def resolve_cleanup_worktree_state(
alias_state["worktree_path"] = path alias_state["worktree_path"] = path
alias_state["expected_worktree_path"] = expected_path alias_state["expected_worktree_path"] = expected_path
alias_state["discovered_worktree_path"] = path alias_state["discovered_worktree_path"] = path
alias_state["match_type"] = "alias" alias_state["match_type"] = (
"branch" if candidates[0].get("branch") == head_branch else "alias"
)
alias_state["candidate_paths"] = [path] alias_state["candidate_paths"] = [path]
alias_state["alias_block_reasons"] = [] alias_state["alias_block_reasons"] = []
alias_state["alias_verified"] = True alias_state["alias_verified"] = True
+66 -2
View File
@@ -154,6 +154,70 @@ class TestMergedCleanupReport(unittest.TestCase):
finally: finally:
os.remove(lock_path) os.remove(lock_path)
@patch("subprocess.run")
def test_discover_local_worktrees(self, mock_run):
mock_output = (
"worktree /repo/Gitea-Tools\n"
"bare\n\n"
"worktree /repo/Gitea-Tools/branches/custom-name\n"
"branch refs/heads/feat/issue-11-x\n"
"HEAD cafebabe\n\n"
)
mock_run.return_value = Mock(returncode=0, stdout=mock_output)
res = mcr.discover_local_worktrees("/repo/Gitea-Tools")
self.assertEqual(res, {"feat/issue-11-x": "/repo/Gitea-Tools/branches/custom-name"})
@patch("subprocess.run")
@patch("merged_cleanup_reconcile.read_local_worktree_state")
def test_build_report_discovers_non_canonical_worktree_by_branch(self, mock_state, mock_run):
mock_output = (
"worktree /repo/Gitea-Tools\n"
"bare\n\n"
"worktree /repo/Gitea-Tools/branches/custom-name\n"
"branch refs/heads/feat/issue-11-x\n"
"HEAD cafebabe\n\n"
)
mock_run.return_value = Mock(returncode=0, stdout=mock_output)
mock_state.side_effect = [
{
"exists": False,
"clean": None,
"current_branch": None,
"head_sha": None,
"dirty_files": [],
},
{
"exists": True,
"clean": True,
"current_branch": "feat/issue-11-x",
"head_sha": "cafebabe",
"dirty_files": [],
},
]
report = mcr.build_reconciliation_report(
project_root="/repo/Gitea-Tools",
closed_prs=[
{
"number": 11,
"title": "merged",
"body": "Closes #11",
"head": {"ref": "feat/issue-11-x"},
"merged_at": "2026-07-06T12:00:00Z",
"merge_commit_sha": "abc123",
}
],
open_prs=[],
remote_branch_exists={"feat/issue-11-x": True},
head_on_master={11: True},
delete_capability_allowed=True,
)
self.assertEqual(report["merged_pr_count"], 1)
entry = report["entries"][0]
local = entry["local_worktree"]
self.assertEqual(local["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
self.assertEqual(local["match_type"], "branch")
@patch("merged_cleanup_reconcile.read_local_worktree_state") @patch("merged_cleanup_reconcile.read_local_worktree_state")
@patch("subprocess.run") @patch("subprocess.run")
def test_build_report_discovers_issue_alias_worktree(self, mock_run, mock_state): def test_build_report_discovers_issue_alias_worktree(self, mock_run, mock_state):
@@ -209,7 +273,7 @@ class TestMergedCleanupReport(unittest.TestCase):
local = report["entries"][0]["local_worktree"] local = report["entries"][0]["local_worktree"]
self.assertTrue(local["safe_to_remove_worktree"]) self.assertTrue(local["safe_to_remove_worktree"])
self.assertEqual(local["match_type"], "alias") self.assertEqual(local["match_type"], "branch")
self.assertEqual( self.assertEqual(
local["expected_worktree_path"], local["expected_worktree_path"],
"/repo/Gitea-Tools/branches/feat-issue-510-workspace-binding-isolation", "/repo/Gitea-Tools/branches/feat-issue-510-workspace-binding-isolation",
@@ -308,7 +372,7 @@ class TestMergedCleanupReport(unittest.TestCase):
) )
self.assertTrue(state["exists"]) self.assertTrue(state["exists"])
self.assertEqual(state["match_type"], "alias") self.assertEqual(state["match_type"], "branch")
self.assertEqual( self.assertEqual(
state["discovered_worktree_path"], state["discovered_worktree_path"],
"/repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation", "/repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation",