Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 79 Pull Requests 7 Actions Packages Projects Releases Wiki Activity

fix(cleanup): harden readback scope and ownership fail-closed (#687)

Browse Source 
Require branch-scoped post-delete not-found (not generic/repo/host 404),
emit consistent top-level cleanup fields, fail closed on ownership inventory
errors, never auto-reclaim expired control-plane leases, include active
comment-backed reviewer leases, apply the same gates to reconcile_merged_cleanups,
and match ownership with normalized host identity.
This commit is contained in:
sysadmin
2026-07-16 01:21:18 -04:00
parent 137426f7ad
commit 1c37e62014
3 changed files with 1008 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_branch_cleanup_guard.py
+441 -22
View File
@@ -95,7 +95,7 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
# Default: no active ownership records (tests that need ownership patch this).
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[],
return_value={"records": [], "inventory_error": False},
).start()
def tearDown(self):
@@ -201,6 +201,9 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
if len(get_branch_calls) <= 1:
return {"name": branch}
raise RuntimeError("HTTP 404: not found")
if method == "GET" and url.rstrip("/").endswith("/Example-Repo"):
# Repo reachability probe after branch 404 (R1 branch-scoped).
return {"full_name": "Example-Org/Example-Repo"}
if method == "DELETE":
return {}
raise AssertionError(f"unexpected {method} {url}")
@@ -215,6 +218,8 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
)
self.assertTrue(res["success"])
self.assertTrue(res["performed"])
self.assertTrue(res["delete_acknowledged"])
self.assertTrue(res["verified_absent"])
self.assertTrue((res.get("readback") or {}).get("verified_absent"))
delete_calls = [
call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
@@ -417,12 +422,37 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase):
class TestPostDeleteReadback(unittest.TestCase):
def test_not_found_is_verified_success(self):
readback = guard.classify_branch_readback_http_status(404)
def test_branch_scoped_not_found_is_verified_success(self):
readback = guard.classify_branch_readback_http_status(
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
)
result = guard.assess_post_delete_readback(readback)
self.assertTrue(result["ok"])
self.assertTrue(result["verified_absent"])
self.assertTrue(result["readback"]["verified_absent"])
def test_generic_404_not_verified_absent(self):
# R1: bare 404 must not verify absence
for scope in (None, guard.NOT_FOUND_SCOPE_UNKNOWN,
guard.NOT_FOUND_SCOPE_REPOSITORY,
guard.NOT_FOUND_SCOPE_HOST):
with self.subTest(scope=scope):
readback = guard.classify_branch_readback_http_status(
404, not_found_scope=scope
)
self.assertFalse(readback["verified_absent"])
result = guard.assess_post_delete_readback(readback)
self.assertFalse(result["ok"])
self.assertFalse(result["verified_absent"])
def test_exception_substring_404_not_verified(self):
# R1: substring/generic 404 without scope stays unverified
result = guard.classify_branch_readback_exception(
RuntimeError("HTTP 404: something not found")
)
self.assertFalse(result["verified_absent"])
self.assertNotEqual(result.get("not_found_scope"), guard.NOT_FOUND_SCOPE_BRANCH)
def test_exists_is_structured_failure(self):
readback = guard.classify_branch_readback_http_status(200)
result = guard.assess_post_delete_readback(readback)
@@ -465,6 +495,7 @@ class TestActiveBranchOwnership(unittest.TestCase):
"category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
"status": "active",
"remote": "prgs",
"host": "gitea.prgs.cc",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"branch": "feat/target",
@@ -479,6 +510,7 @@ class TestActiveBranchOwnership(unittest.TestCase):
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=[self._base()],
)
self.assertTrue(result["block"])
@@ -490,6 +522,7 @@ class TestActiveBranchOwnership(unittest.TestCase):
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=[
self._base(
category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE,
@@ -513,6 +546,7 @@ class TestActiveBranchOwnership(unittest.TestCase):
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=[self._base(category=cat, status="active")],
)
self.assertTrue(result["block"])
@@ -532,6 +566,7 @@ class TestActiveBranchOwnership(unittest.TestCase):
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=records,
)
self.assertFalse(result["block"])
@@ -542,26 +577,47 @@ class TestActiveBranchOwnership(unittest.TestCase):
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=[self._base(status="stale", reclaim_allowed=False)],
)
self.assertTrue(result["block"])
self.assertIn("sticky", " ".join(result["reasons"]))
self.assertTrue(
any(
token in " ".join(result["reasons"])
for token in ("sticky", "reclaim not proven", "fail closed")
)
)
def test_other_repo_or_branch_no_false_block(self):
records = [
self._base(repo="Other-Repo", status="active"),
self._base(branch="feat/other", status="active"),
self._base(remote="dadeschools", status="active"),
self._base(host="gitea.other.host", status="active"),
]
result = guard.assess_active_branch_ownership(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=records,
)
self.assertFalse(result["block"])
self.assertEqual(len(result["ignored_out_of_scope"]), 3)
self.assertEqual(len(result["ignored_out_of_scope"]), 4)
def test_normalized_host_matching(self):
# Host identity is normalized (scheme/path stripped)
records = [self._base(host="https://gitea.prgs.cc/")]
result = guard.assess_active_branch_ownership(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=records,
)
self.assertTrue(result["block"])
def test_denial_has_no_secrets(self):
result = guard.assess_active_branch_ownership(
@@ -569,6 +625,7 @@ class TestActiveBranchOwnership(unittest.TestCase):
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch="feat/target",
host="gitea.prgs.cc",
records=[
self._base(
status="active",
@@ -626,7 +683,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
branch = "feat/still-there"
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[],
return_value={"records": [], "inventory_error": False},
).start()
def _api(method, url, *a, **k):
@@ -649,13 +706,14 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
self.assertTrue(res.get("performed"))
self.assertFalse(res.get("success"))
self.assertTrue(res.get("delete_acknowledged"))
self.assertFalse(res.get("verified_absent"))
self.assertIn("still present", " ".join(res.get("reasons") or []))
def test_readback_authentication_failure(self):
branch = "feat/auth-fail-readback"
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[],
return_value={"records": [], "inventory_error": False},
).start()
state = {"branch_gets": 0}
@@ -687,7 +745,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
branch = "feat/transport-fail-readback"
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[],
return_value={"records": [], "inventory_error": False},
).start()
state = {"branch_gets": 0}
@@ -718,17 +776,21 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
branch = "feat/owned"
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[
{
"category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
"status": "active",
"remote": "prgs",
"org": "Example-Org",
"repo": "Example-Repo",
"branch": branch,
"reclaim_allowed": False,
}
],
return_value={
"records": [
{
"category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
"status": "active",
"remote": "prgs",
"host": "gitea.example.com",
"org": "Example-Org",
"repo": "Example-Repo",
"branch": branch,
"reclaim_allowed": False,
}
],
"inventory_error": False,
},
).start()
def _api(method, url, *a, **k):
@@ -762,7 +824,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
branch = "feat/open-pr-head"
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[],
return_value={"records": [], "inventory_error": False},
).start()
patch(
"mcp_server.api_get_all",
@@ -791,7 +853,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
branch = "feat/not-ancestor"
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[],
return_value={"records": [], "inventory_error": False},
).start()
patch(
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
@@ -820,7 +882,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
branch = "master"
patch(
"mcp_server._collect_branch_ownership_records",
return_value=[],
return_value={"records": [], "inventory_error": False},
).start()
def _api(method, url, *a, **k):
@@ -843,5 +905,362 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
class TestSecondRemediationR1R2(unittest.TestCase):
"""R1/R2 second-remediation: branch-scoped 404 and top-level fields."""
def test_cleanup_envelope_always_has_top_level_fields(self):
env = guard.cleanup_result_envelope(
success=False,
performed=False,
delete_acknowledged=False,
verified_absent=False,
reasons=["x"],
)
for key in ("success", "performed", "delete_acknowledged", "verified_absent"):
self.assertIn(key, env)
self.assertIsInstance(env[key], bool)
def test_repo_scoped_404_never_verified(self):
rb = guard.classify_branch_readback_http_status(
404, not_found_scope=guard.NOT_FOUND_SCOPE_REPOSITORY
)
self.assertFalse(rb["verified_absent"])
assessed = guard.assess_post_delete_readback(rb)
self.assertFalse(assessed["ok"])
self.assertFalse(assessed["verified_absent"])
def test_wrong_host_404_never_verified(self):
rb = guard.classify_branch_readback_http_status(
404, not_found_scope=guard.NOT_FOUND_SCOPE_HOST
)
self.assertFalse(rb["verified_absent"])
class TestSecondRemediationOwnership(unittest.TestCase):
"""O1/O2/O3 ownership second-remediation."""
def test_inventory_error_category_blocks(self):
result = guard.assess_active_branch_ownership(
remote="prgs",
org="Org",
repo="Repo",
branch="feat/x",
host="gitea.example.com",
records=[
{
"category": guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR,
"status": "unknown",
"remote": "prgs",
"host": "gitea.example.com",
"org": "Org",
"repo": "Repo",
"branch": "feat/x",
"reclaim_allowed": False,
}
],
)
self.assertTrue(result["block"])
self.assertIn(
guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR,
result["blocking_categories"],
)
def test_expired_without_explicit_reclaim_blocks(self):
# O2: expired must not auto-allow
result = guard.assess_active_branch_ownership(
remote="prgs",
org="Org",
repo="Repo",
branch="feat/x",
host="gitea.example.com",
records=[
{
"category": guard.OWNERSHIP_CATEGORY_MERGER_LEASE,
"status": "expired",
"remote": "prgs",
"host": "gitea.example.com",
"org": "Org",
"repo": "Repo",
"branch": "feat/x",
# reclaim_allowed omitted / False
"reclaim_allowed": False,
}
],
)
self.assertTrue(result["block"])
def test_expired_with_explicit_reclaim_allowed_does_not_block(self):
result = guard.assess_active_branch_ownership(
remote="prgs",
org="Org",
repo="Repo",
branch="feat/x",
host="gitea.example.com",
records=[
{
"category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
"status": "expired",
"remote": "prgs",
"host": "gitea.example.com",
"org": "Org",
"repo": "Repo",
"branch": "feat/x",
"reclaim_allowed": True,
}
],
)
self.assertFalse(result["block"])
def test_active_reviewer_comment_lease_blocks(self):
result = guard.assess_active_branch_ownership(
remote="prgs",
org="Org",
repo="Repo",
branch="feat/x",
host="gitea.example.com",
records=[
{
"category": guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE,
"status": "active",
"remote": "prgs",
"host": "gitea.example.com",
"org": "Org",
"repo": "Repo",
"branch": "feat/x",
"reclaim_allowed": False,
"role": "reviewer",
}
],
)
self.assertTrue(result["block"])
self.assertIn(
guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE,
result["blocking_categories"],
)
class TestSecondRemediationIntegration(unittest.TestCase):
def setUp(self):
self._remotes = patch.dict(
mcp_server.REMOTES,
{
"prgs": {
"host": "gitea.example.com",
"org": "Example-Org",
"repo": "Example-Repo",
}
},
)
self._remotes.start()
patch("gitea_audit.audit_enabled", return_value=False).start()
self.mock_api = patch("mcp_server.api_request").start()
patch("mcp_server.api_get_all", return_value=[]).start()
patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
patch(
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
return_value=True,
).start()
patch(
"mcp_server.get_profile",
return_value=dict(RECONCILER_WITH_DELETE),
).start()
def tearDown(self):
patch.stopall()
def _pr(self, branch, number=487):
return {
"number": number,
"merged": True,
"merged_at": "2026-07-08T01:00:00Z",
"head": {"ref": branch, "sha": "a" * 40},
"base": {"ref": "master"},
}
def test_r1_repo_404_after_delete_not_verified(self):
"""After DELETE, branch 404 + repo 404 must not verify absence."""
branch = "feat/r1-repo-404"
patch(
"mcp_server._collect_branch_ownership_records",
return_value={"records": [], "inventory_error": False},
).start()
state = {"branch_gets": 0}
def _api(method, url, *a, **k):
if method == "GET" and "/pulls/" in url:
return self._pr(branch)
if method == "GET" and "/branches/" in url:
state["branch_gets"] += 1
if state["branch_gets"] == 1:
return {"name": branch}
raise RuntimeError("HTTP 404: not found")
if method == "GET" and url.rstrip("/").endswith("/Example-Repo"):
raise RuntimeError("HTTP 404: repository not found")
if method == "DELETE":
return {}
raise AssertionError(method + " " + url)
self.mock_api.side_effect = _api
res = gitea_cleanup_merged_pr_branch(
pr_number=487,
confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
branch=branch,
remote="prgs",
worktree_path="/tmp/repo/branches/cleanup",
)
self.assertTrue(res["performed"])
self.assertTrue(res["delete_acknowledged"])
self.assertFalse(res["success"])
self.assertFalse(res["verified_absent"])
def test_o1_inventory_error_blocks_before_delete(self):
branch = "feat/o1-inv"
patch(
"mcp_server._collect_branch_ownership_records",
return_value={"records": [], "inventory_error": True},
).start()
def _api(method, url, *a, **k):
if method == "GET" and "/pulls/" in url:
return self._pr(branch)
if method == "GET" and "/branches/" in url:
return {"name": branch}
raise AssertionError(f"no mutation expected {method}")
self.mock_api.side_effect = _api
res = gitea_cleanup_merged_pr_branch(
pr_number=487,
confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
branch=branch,
remote="prgs",
worktree_path="/tmp/repo/branches/cleanup",
)
self.assertFalse(res["performed"])
self.assertFalse(res["delete_acknowledged"])
self.assertFalse(res["verified_absent"])
self.assertEqual(res.get("blocker_kind"), "active_branch_ownership")
def test_o3_comment_reviewer_lease_in_collector(self):
"""Collector includes active comment-backed reviewer leases."""
active_lease = {
"pr_number": 10,
"phase": "claimed",
"session_id": "s1",
"expires_at": "2099-01-02T00:00:00Z",
}
with patch(
"mcp_server.api_get_all", return_value=[{"id": 1, "body": "x"}]
), patch(
"mcp_server.reviewer_pr_lease.find_active_reviewer_lease",
return_value=active_lease,
), patch(
"mcp_server.issue_lock_store.iter_lock_files", return_value=[]
), patch(
"mcp_server.worktree_cleanup_audit.list_worktrees", return_value=[]
), patch.object(
mcp_server.control_plane_db,
"ControlPlaneDB",
side_effect=RuntimeError("no cp"),
):
bundle = mcp_server._collect_branch_ownership_records(
remote="prgs",
host="gitea.example.com",
org="Example-Org",
repo="Example-Repo",
branch="feat/x",
pr_number=10,
project_root="/tmp/repo",
auth=FAKE_AUTH,
base_api=(
"https://gitea.example.com/api/v1/repos/"
"Example-Org/Example-Repo"
),
)
cats = {r.get("category") for r in bundle.get("records") or []}
self.assertIn(guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, cats)
def test_o4_reconcile_merged_cleanups_runs_ownership_and_readback(self):
from mcp_server import gitea_reconcile_merged_cleanups
branch = "feat/reconcile-o4"
ownership_calls = []
def fake_collect(**kwargs):
ownership_calls.append(kwargs)
return {"records": [], "inventory_error": False}
probe_calls = []
def fake_probe(h, o, r, auth, br):
probe_calls.append(br)
return guard.classify_branch_readback_http_status(
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
)
report = {
"entries": [
{
"pr_number": 1,
"head_branch": branch,
"remote_branch": {"safe_to_delete_remote": True},
"local_worktree": {"safe_to_remove_worktree": False},
}
],
"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()
self.mock_api.side_effect = lambda *a, **k: {}
res = gitea_reconcile_merged_cleanups(
dry_run=False,
execute_confirmed=True,
remote="prgs",
)
self.assertTrue(res.get("performed") or res.get("executed"))
self.assertTrue(ownership_calls, "ownership must run before delete")
self.assertTrue(probe_calls, "post-delete readback must run")
actions = res.get("actions") or []
delete_actions = [
a for a in actions if a.get("action") == "delete_remote_branch"
]
self.assertEqual(len(delete_actions), 1)
self.assertIn("verified_absent", delete_actions[0])
self.assertIn("delete_acknowledged", delete_actions[0])
self.assertTrue(delete_actions[0].get("verified_absent"))
if __name__ == "__main__":
unittest.main()