Merge branch 'master' into feat/issue-642-sanctioned-restart-controls
This commit is contained in:
@@ -1266,6 +1266,378 @@ class TestSecondRemediationIntegration(unittest.TestCase):
|
||||
self.assertIn("delete_acknowledged", delete_actions[0])
|
||||
self.assertTrue(delete_actions[0].get("verified_absent"))
|
||||
|
||||
def test_issue_851_worktree_removed_when_remote_blocked_only_by_worktree_binding(self):
|
||||
"""#851: remote blocked by worktree_binding must not skip safe worktree removal.
|
||||
|
||||
Lifecycle: remove clean owned worktree → reassess ownership → delete
|
||||
remote only if independently safe. Unrelated entries stay untouched.
|
||||
"""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
target_branch = "fix/issue-844-exclude-epic-containers"
|
||||
foreign_branch = "fix/issue-999-unrelated-active"
|
||||
worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
|
||||
ownership_calls = []
|
||||
remove_calls = []
|
||||
delete_api_calls = []
|
||||
|
||||
def fake_collect(**kwargs):
|
||||
ownership_calls.append(dict(kwargs))
|
||||
# Ownership is reassessed *after* independent worktree removal (#851).
|
||||
# Target worktree is already gone → no worktree_binding remains.
|
||||
# Foreign branch keeps an active author lease → remote delete blocked.
|
||||
if kwargs.get("branch") == foreign_branch:
|
||||
# Match session-bound org/repo + host used by the tool resolve path.
|
||||
return {
|
||||
"records": [
|
||||
{
|
||||
"category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
|
||||
"status": "active",
|
||||
"remote": kwargs.get("remote") or "prgs",
|
||||
"host": kwargs.get("host") or "gitea.example.com",
|
||||
"org": kwargs.get("org") or "Scaled-Tech-Consulting",
|
||||
"repo": kwargs.get("repo") or "Gitea-Tools",
|
||||
"branch": foreign_branch,
|
||||
"reclaim_allowed": False,
|
||||
}
|
||||
],
|
||||
"inventory_error": False,
|
||||
}
|
||||
return {"records": [], "inventory_error": False}
|
||||
|
||||
def fake_remove(project_root, branch, worktree_path=None):
|
||||
remove_calls.append(
|
||||
{"branch": branch, "worktree_path": worktree_path}
|
||||
)
|
||||
return {
|
||||
"success": True,
|
||||
"performed": True,
|
||||
"message": f"removed worktree {worktree_path}",
|
||||
"worktree_path": worktree_path,
|
||||
}
|
||||
|
||||
def fake_probe(h, o, r, auth, br):
|
||||
return guard.classify_branch_readback_http_status(
|
||||
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
|
||||
)
|
||||
|
||||
def fake_api(method, url, auth, **kwargs):
|
||||
if method == "DELETE":
|
||||
delete_api_calls.append(url)
|
||||
return {}
|
||||
|
||||
report = {
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": 848,
|
||||
"head_branch": target_branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": True,
|
||||
"worktree_path": worktree_path,
|
||||
},
|
||||
},
|
||||
{
|
||||
"pr_number": 999,
|
||||
"head_branch": foreign_branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": False,
|
||||
"worktree_path": None,
|
||||
},
|
||||
},
|
||||
],
|
||||
"reviewer_scratch_entries": [],
|
||||
}
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.branch.delete",
|
||||
"gitea.pr.close",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
patch("mcp_server.api_get_all", return_value=[]).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value=report,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
|
||||
return_value=(True, []),
|
||||
).start()
|
||||
patch("mcp_server.verify_preflight_purity", return_value=None).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
side_effect=fake_collect,
|
||||
).start()
|
||||
patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
|
||||
side_effect=fake_remove,
|
||||
).start()
|
||||
self.mock_api.side_effect = fake_api
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res.get("performed") or res.get("executed"))
|
||||
actions = res.get("actions") or []
|
||||
|
||||
remove_actions = [
|
||||
a for a in actions if a.get("action") == "remove_local_worktree"
|
||||
]
|
||||
self.assertEqual(len(remove_actions), 1, actions)
|
||||
self.assertTrue(remove_actions[0].get("success"))
|
||||
self.assertEqual(remove_calls[0]["branch"], target_branch)
|
||||
self.assertEqual(remove_calls[0]["worktree_path"], worktree_path)
|
||||
|
||||
# Target remote delete succeeds after worktree removal + reassessment.
|
||||
target_deletes = [
|
||||
a
|
||||
for a in actions
|
||||
if a.get("action") == "delete_remote_branch"
|
||||
and a.get("branch") == target_branch
|
||||
]
|
||||
self.assertEqual(len(target_deletes), 1, actions)
|
||||
self.assertTrue(target_deletes[0].get("success"))
|
||||
self.assertTrue(target_deletes[0].get("after_worktree_removal"))
|
||||
self.assertTrue(target_deletes[0].get("ownership_reassessed"))
|
||||
self.assertTrue(target_deletes[0].get("verified_absent"))
|
||||
|
||||
# Foreign branch remains protected (author lease) and is not deleted.
|
||||
foreign_deletes = [
|
||||
a
|
||||
for a in actions
|
||||
if a.get("action") == "delete_remote_branch"
|
||||
and a.get("branch") == foreign_branch
|
||||
]
|
||||
self.assertEqual(len(foreign_deletes), 1, actions)
|
||||
self.assertFalse(foreign_deletes[0].get("success"))
|
||||
self.assertEqual(
|
||||
foreign_deletes[0].get("blocker_kind"), "active_branch_ownership"
|
||||
)
|
||||
self.assertIn(
|
||||
guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
|
||||
foreign_deletes[0].get("blocking_categories") or [],
|
||||
)
|
||||
# Only the target branch should hit the DELETE API.
|
||||
self.assertEqual(len(delete_api_calls), 1)
|
||||
|
||||
# Ownership collected for target (post-removal) and foreign; worktree
|
||||
# removal happened before target remote delete in the action log.
|
||||
target_idx = next(
|
||||
i
|
||||
for i, a in enumerate(actions)
|
||||
if a.get("action") == "remove_local_worktree"
|
||||
)
|
||||
delete_idx = next(
|
||||
i
|
||||
for i, a in enumerate(actions)
|
||||
if a.get("action") == "delete_remote_branch"
|
||||
and a.get("branch") == target_branch
|
||||
and a.get("success")
|
||||
)
|
||||
self.assertLess(target_idx, delete_idx)
|
||||
|
||||
def test_issue_851_dirty_worktree_not_removed_and_remote_stays_protected(self):
|
||||
"""#851: dirty/foreign worktrees remain protected; no unsafe cleanup."""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
branch = "fix/issue-851-dirty"
|
||||
remove_calls = []
|
||||
|
||||
def fake_collect(**kwargs):
|
||||
return {
|
||||
"records": [
|
||||
{
|
||||
"category": guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
|
||||
"status": "active",
|
||||
"remote": kwargs.get("remote") or "prgs",
|
||||
"host": kwargs.get("host") or "gitea.example.com",
|
||||
"org": kwargs.get("org") or "Scaled-Tech-Consulting",
|
||||
"repo": kwargs.get("repo") or "Gitea-Tools",
|
||||
"branch": branch,
|
||||
"reclaim_allowed": False,
|
||||
}
|
||||
],
|
||||
"inventory_error": False,
|
||||
}
|
||||
|
||||
report = {
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": 851,
|
||||
"head_branch": branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": False,
|
||||
"worktree_path": "/tmp/dirty-wt",
|
||||
},
|
||||
}
|
||||
],
|
||||
"reviewer_scratch_entries": [],
|
||||
}
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
patch("mcp_server.api_get_all", return_value=[]).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value=report,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
|
||||
return_value=(True, []),
|
||||
).start()
|
||||
patch("mcp_server.verify_preflight_purity", return_value=None).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
side_effect=fake_collect,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
|
||||
side_effect=lambda *a, **k: remove_calls.append(k) or {
|
||||
"success": True,
|
||||
"performed": True,
|
||||
},
|
||||
).start()
|
||||
self.mock_api.side_effect = lambda *a, **k: {}
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
remote="prgs",
|
||||
)
|
||||
actions = res.get("actions") or []
|
||||
self.assertEqual(remove_calls, [])
|
||||
self.assertFalse(
|
||||
any(a.get("action") == "remove_local_worktree" for a in actions)
|
||||
)
|
||||
deletes = [
|
||||
a for a in actions if a.get("action") == "delete_remote_branch"
|
||||
]
|
||||
self.assertEqual(len(deletes), 1)
|
||||
self.assertFalse(deletes[0].get("success"))
|
||||
self.assertEqual(deletes[0].get("blocker_kind"), "active_branch_ownership")
|
||||
self.assertIn(
|
||||
guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
|
||||
deletes[0].get("blocking_categories") or [],
|
||||
)
|
||||
|
||||
def test_issue_851_idempotent_resume_when_worktree_already_absent(self):
|
||||
"""#851: partial failures remain resumable and idempotent."""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
branch = "fix/issue-851-resume"
|
||||
ownership_calls = []
|
||||
|
||||
def fake_collect(**kwargs):
|
||||
ownership_calls.append(kwargs)
|
||||
return {"records": [], "inventory_error": False}
|
||||
|
||||
def fake_remove(project_root, branch, worktree_path=None):
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"message": f"worktree not found: {worktree_path}",
|
||||
}
|
||||
|
||||
def fake_probe(h, o, r, auth, br):
|
||||
return guard.classify_branch_readback_http_status(
|
||||
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
|
||||
)
|
||||
|
||||
report = {
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": 851,
|
||||
"head_branch": branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": True,
|
||||
"worktree_path": "/tmp/already-gone",
|
||||
},
|
||||
}
|
||||
],
|
||||
"reviewer_scratch_entries": [],
|
||||
}
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
patch("mcp_server.api_get_all", return_value=[]).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value=report,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
|
||||
return_value=(True, []),
|
||||
).start()
|
||||
patch("mcp_server.verify_preflight_purity", return_value=None).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
side_effect=fake_collect,
|
||||
).start()
|
||||
patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
|
||||
side_effect=fake_remove,
|
||||
).start()
|
||||
self.mock_api.side_effect = lambda *a, **k: {}
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
remote="prgs",
|
||||
)
|
||||
actions = res.get("actions") or []
|
||||
removes = [a for a in actions if a.get("action") == "remove_local_worktree"]
|
||||
deletes = [a for a in actions if a.get("action") == "delete_remote_branch"]
|
||||
self.assertEqual(len(removes), 1)
|
||||
self.assertFalse(removes[0].get("success"))
|
||||
self.assertEqual(len(deletes), 1)
|
||||
self.assertTrue(deletes[0].get("success"))
|
||||
self.assertTrue(deletes[0].get("after_worktree_removal"))
|
||||
self.assertTrue(ownership_calls)
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -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()
|
||||
@@ -12,6 +12,59 @@ import merged_cleanup_reconcile as mcr # noqa: E402
|
||||
|
||||
|
||||
class TestMergedCleanupAssessment(unittest.TestCase):
|
||||
def test_issue_851_plan_order_worktree_then_reassess_then_remote(self):
|
||||
"""#851 dry-run plan: remove worktree, reassess ownership, then remote."""
|
||||
plan = mcr.plan_cleanup_execution_order(
|
||||
remote_assessment={"safe_to_delete_remote": True},
|
||||
local_assessment={"safe_to_remove_worktree": True},
|
||||
)
|
||||
actions = [s["action"] for s in plan]
|
||||
self.assertEqual(
|
||||
actions,
|
||||
[
|
||||
"remove_local_worktree",
|
||||
"reassess_branch_ownership",
|
||||
"delete_remote_branch",
|
||||
],
|
||||
)
|
||||
self.assertEqual(plan[0]["phase"], 1)
|
||||
self.assertEqual(plan[-1]["phase"], 3)
|
||||
self.assertIn("independently_safe", plan[0]["reason"])
|
||||
self.assertIn("reassessment", plan[-1]["reason"])
|
||||
|
||||
def test_issue_851_plan_remote_only_when_worktree_not_safe(self):
|
||||
plan = mcr.plan_cleanup_execution_order(
|
||||
remote_assessment={"safe_to_delete_remote": True},
|
||||
local_assessment={"safe_to_remove_worktree": False},
|
||||
)
|
||||
self.assertEqual([s["action"] for s in plan], ["delete_remote_branch"])
|
||||
self.assertNotIn("reassess_branch_ownership", [s["action"] for s in plan])
|
||||
|
||||
def test_issue_851_plan_worktree_only_when_remote_not_safe(self):
|
||||
plan = mcr.plan_cleanup_execution_order(
|
||||
remote_assessment={"safe_to_delete_remote": False},
|
||||
local_assessment={"safe_to_remove_worktree": True},
|
||||
)
|
||||
self.assertEqual([s["action"] for s in plan], ["remove_local_worktree"])
|
||||
|
||||
def test_issue_851_entry_includes_planned_execution_order(self):
|
||||
entry = mcr.build_pr_cleanup_entry(
|
||||
pr={
|
||||
"number": 848,
|
||||
"title": "Closes #844",
|
||||
"body": "",
|
||||
"merged_at": "2026-07-23T00:00:00Z",
|
||||
"head": {"ref": "fix/issue-844-x", "sha": "a" * 40},
|
||||
},
|
||||
project_root="/tmp/not-a-real-root",
|
||||
open_pr_heads=set(),
|
||||
remote_branch_exists=True,
|
||||
head_on_master=True,
|
||||
delete_capability_allowed=True,
|
||||
)
|
||||
self.assertIn("planned_execution_order", entry)
|
||||
self.assertIsInstance(entry["planned_execution_order"], list)
|
||||
|
||||
def test_extract_linked_issue_from_closes(self):
|
||||
issue = mcr.extract_linked_issue(
|
||||
"feat: cleanup (Closes #269)",
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tests for the Phase 1 operator console application shell (#638)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.routing import Route
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui import layout
|
||||
from webui.app import create_app
|
||||
from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
|
||||
|
||||
|
||||
class TestShellNav(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_nav_group_labels_present(self):
|
||||
text = self.client.get("/").text
|
||||
for group in NAV_GROUPS:
|
||||
with self.subTest(group=group.label):
|
||||
self.assertIn(f">{group.label}<", text)
|
||||
|
||||
def test_phase1_group_labels_cover_expected_ia(self):
|
||||
labels = {group.label for group in NAV_GROUPS}
|
||||
for expected in (
|
||||
"Health",
|
||||
"Traffic",
|
||||
"Runtime/Sessions",
|
||||
"Projects",
|
||||
"Inventory",
|
||||
"Timeline",
|
||||
"Policy",
|
||||
"Insights",
|
||||
):
|
||||
with self.subTest(label=expected):
|
||||
self.assertIn(expected, labels)
|
||||
|
||||
def test_every_nav_href_resolves_to_a_get_route(self):
|
||||
app = create_app()
|
||||
get_paths = {
|
||||
route.path
|
||||
for route in app.routes
|
||||
if isinstance(route, Route) and "GET" in route.methods
|
||||
}
|
||||
for href in nav_hrefs():
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(href, get_paths, f"nav href {href} has no GET route")
|
||||
|
||||
def test_legacy_hrefs_still_navigable(self):
|
||||
text = self.client.get("/").text
|
||||
for href in ("/queue", "/projects", "/prompts", "/runtime",
|
||||
"/audit", "/worktrees", "/leases", "/actions"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
class TestShellBadges(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_mode_badge_present(self):
|
||||
self.assertIn("mode: read-only", self.client.get("/").text)
|
||||
|
||||
def test_environment_badge_present(self):
|
||||
self.assertIn("env:", self.client.get("/").text)
|
||||
|
||||
def test_default_environment_is_local(self):
|
||||
self.assertEqual(layout.environment_label(), "local")
|
||||
|
||||
def test_remote_bind_reports_remote_environment(self):
|
||||
import os
|
||||
|
||||
prior = os.environ.get("WEBUI_HOST")
|
||||
os.environ["WEBUI_HOST"] = "10.0.0.5"
|
||||
try:
|
||||
self.assertEqual(layout.environment_label(), "remote")
|
||||
finally:
|
||||
if prior is None:
|
||||
os.environ.pop("WEBUI_HOST", None)
|
||||
else:
|
||||
os.environ["WEBUI_HOST"] = prior
|
||||
|
||||
def test_docs_link_present(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn(layout.DOCS_URL, text)
|
||||
self.assertIn(">Docs<", text)
|
||||
|
||||
|
||||
class TestShellStubs(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_stub_routes_render_200(self):
|
||||
for path, (title, _desc) in STUB_PAGES.items():
|
||||
with self.subTest(path=path):
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 200, path)
|
||||
self.assertIn(title, response.text)
|
||||
self.assertIn("placeholder", response.text)
|
||||
|
||||
def test_stub_routes_are_read_only(self):
|
||||
for path in STUB_PAGES:
|
||||
with self.subTest(path=path):
|
||||
response = self.client.post(path)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
def test_stub_pages_carry_nav_and_badges(self):
|
||||
response = self.client.get("/inventory")
|
||||
self.assertIn("mode: read-only", response.text)
|
||||
self.assertIn('href="/queue"', response.text)
|
||||
|
||||
|
||||
class TestShellHome(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_home_summarizes_console(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("Operator console", text)
|
||||
self.assertIn("Phase 1", text)
|
||||
|
||||
def test_home_links_legacy_pages(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("MVP legacy pages", text)
|
||||
for href in ("/queue", "/audit", "/leases"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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