Merge branch 'master' into feat/issue-658-mcp-restart-coordinator
This commit is contained in:
@@ -1639,6 +1639,358 @@ class TestSecondRemediationIntegration(unittest.TestCase):
|
||||
self.assertTrue(ownership_calls)
|
||||
|
||||
|
||||
class TestIssue855ExactPrSelector(unittest.TestCase):
|
||||
"""#855: exact pr_number pin for reconcile_merged_cleanups (#851 lifecycle)."""
|
||||
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(
|
||||
mcp_server.REMOTES,
|
||||
{
|
||||
"prgs": {
|
||||
"host": "gitea.example.com",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Gitea-Tools",
|
||||
}
|
||||
},
|
||||
)
|
||||
self._remotes.start()
|
||||
patch("gitea_audit.audit_enabled", return_value=False).start()
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
self.mock_all = 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()
|
||||
patch(
|
||||
"mcp_server._profile_operation_gate",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
return_value={"records": [], "inventory_error": False},
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch("mcp_server.verify_preflight_purity", return_value=None).start()
|
||||
patch(
|
||||
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
|
||||
return_value=(True, []),
|
||||
).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
|
||||
def _merged_pr(self, number, branch, sha="c" * 40):
|
||||
return {
|
||||
"number": number,
|
||||
"title": f"PR {number}",
|
||||
"body": f"Closes #{number - 4}",
|
||||
"merged": True,
|
||||
"merged_at": "2026-07-23T12:00:00Z",
|
||||
"merge_commit_sha": "f" * 40,
|
||||
"state": "closed",
|
||||
"head": {"ref": branch, "sha": sha},
|
||||
"base": {"ref": "master"},
|
||||
}
|
||||
|
||||
def test_exact_pr_848_ignores_newer_852_in_batch_queue(self):
|
||||
"""pr_number=848 selects only #848 even when #852 is newer/first."""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
pr_848 = self._merged_pr(
|
||||
848, "fix/issue-844-exclude-epic-containers", sha="c3f282ba" + "0" * 32
|
||||
)
|
||||
# Closed list would rank #852 first in batch mode; exact pin must ignore it.
|
||||
closed_batch = [
|
||||
self._merged_pr(852, "fix/issue-851-cleanup-worktree-before-remote-delete"),
|
||||
pr_848,
|
||||
self._merged_pr(849, "fix/issue-849-other"),
|
||||
self._merged_pr(846, "fix/issue-846-other"),
|
||||
self._merged_pr(845, "fix/issue-845-other"),
|
||||
]
|
||||
batch_fetch_calls = []
|
||||
|
||||
def fake_api(method, url, *args, **kwargs):
|
||||
if method == "GET" and url.rstrip("/").endswith("/pulls/848"):
|
||||
return dict(pr_848)
|
||||
if method == "GET" and "/pulls/" in url:
|
||||
raise AssertionError(f"unexpected PR fetch: {url}")
|
||||
if method == "GET" and "/branches/" in url:
|
||||
return {"name": "present"}
|
||||
return {}
|
||||
|
||||
def fake_all(url, auth, limit=None):
|
||||
batch_fetch_calls.append((url, limit))
|
||||
if "state=open" in url:
|
||||
return []
|
||||
if "state=closed" in url:
|
||||
# Exact mode must not use the closed batch list.
|
||||
raise AssertionError(
|
||||
"exact pr_number mode must not page closed PRs: " + url
|
||||
)
|
||||
return []
|
||||
|
||||
self.mock_api.side_effect = fake_api
|
||||
self.mock_all.side_effect = fake_all
|
||||
patch(
|
||||
"mcp_server._remote_branch_exists",
|
||||
return_value=True,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
side_effect=lambda **kwargs: {
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": int(pr["number"]),
|
||||
"head_branch": (pr.get("head") or {}).get("ref"),
|
||||
"issue_number": 844,
|
||||
"remote_branch": {
|
||||
"safe_to_delete_remote": True,
|
||||
"head_branch": (pr.get("head") or {}).get("ref"),
|
||||
},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": True,
|
||||
"worktree_path": (
|
||||
"/tmp/branches/fix-issue-844-exclude-epic-containers"
|
||||
),
|
||||
},
|
||||
"planned_execution_order": (
|
||||
mcp_server.merged_cleanup_reconcile.plan_cleanup_execution_order(
|
||||
remote_assessment={"safe_to_delete_remote": True},
|
||||
local_assessment={"safe_to_remove_worktree": True},
|
||||
)
|
||||
),
|
||||
}
|
||||
for pr in kwargs.get("closed_prs") or []
|
||||
if pr.get("merged_at") or pr.get("merged")
|
||||
],
|
||||
"reviewer_scratch_entries": [],
|
||||
"merged_pr_count": len(kwargs.get("closed_prs") or []),
|
||||
},
|
||||
).start()
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=True,
|
||||
pr_number=848,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertTrue(res.get("success"))
|
||||
self.assertFalse(res.get("performed"))
|
||||
self.assertEqual(res.get("selection_mode"), "exact_pr")
|
||||
self.assertEqual(res.get("selected_pr_number"), 848)
|
||||
entries = res.get("entries") or []
|
||||
self.assertEqual(len(entries), 1, entries)
|
||||
self.assertEqual(entries[0].get("pr_number"), 848)
|
||||
self.assertEqual(
|
||||
entries[0].get("head_branch"),
|
||||
"fix/issue-844-exclude-epic-containers",
|
||||
)
|
||||
# No other PR appears in plan.
|
||||
self.assertEqual(list((res.get("planned_execution_orders") or {}).keys()), ["848"])
|
||||
plan = (res.get("planned_execution_orders") or {}).get("848") or []
|
||||
actions = [s.get("action") for s in plan]
|
||||
self.assertEqual(
|
||||
actions,
|
||||
[
|
||||
"remove_local_worktree",
|
||||
"reassess_branch_ownership",
|
||||
"delete_remote_branch",
|
||||
],
|
||||
)
|
||||
# Prove we never scanned the multi-PR closed batch.
|
||||
self.assertFalse(any("state=closed" in (u or "") for u, _ in batch_fetch_calls))
|
||||
# closed_batch fixture must remain unused (sanity).
|
||||
self.assertEqual(closed_batch[0]["number"], 852)
|
||||
|
||||
def test_exact_pr_execute_only_mutates_selected_pr(self):
|
||||
"""Execute with pr_number must never touch #845/#846/#849/#852."""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
pr_848 = self._merged_pr(848, "fix/issue-844-exclude-epic-containers")
|
||||
worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
|
||||
remove_calls = []
|
||||
delete_api_calls = []
|
||||
ownership_branches = []
|
||||
|
||||
def fake_api(method, url, *args, **kwargs):
|
||||
if method == "GET" and url.rstrip("/").endswith("/pulls/848"):
|
||||
return dict(pr_848)
|
||||
if method == "DELETE":
|
||||
delete_api_calls.append(url)
|
||||
# Forbid foreign PR branch deletion by URL content.
|
||||
for forbidden in ("845", "846", "849", "852"):
|
||||
self.assertNotIn(forbidden, url)
|
||||
return {}
|
||||
|
||||
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_path}",
|
||||
"worktree_path": worktree_path,
|
||||
}
|
||||
|
||||
def fake_collect(**kwargs):
|
||||
ownership_branches.append(kwargs.get("branch"))
|
||||
return {"records": [], "inventory_error": False}
|
||||
|
||||
def fake_probe(h, o, r, auth, br):
|
||||
return guard.classify_branch_readback_http_status(
|
||||
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
|
||||
)
|
||||
|
||||
self.mock_api.side_effect = fake_api
|
||||
self.mock_all.side_effect = lambda url, auth, limit=None: []
|
||||
patch("mcp_server._remote_branch_exists", return_value=True).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value={
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": 848,
|
||||
"head_branch": "fix/issue-844-exclude-epic-containers",
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": True,
|
||||
"worktree_path": worktree_path,
|
||||
},
|
||||
"planned_execution_order": [
|
||||
{"action": "remove_local_worktree", "phase": 1},
|
||||
{"action": "reassess_branch_ownership", "phase": 2},
|
||||
{"action": "delete_remote_branch", "phase": 3},
|
||||
],
|
||||
}
|
||||
],
|
||||
"reviewer_scratch_entries": [
|
||||
# Foreign scratch must be filtered before report execute loop;
|
||||
# if present here it would still be a test failure if acted on.
|
||||
],
|
||||
"merged_pr_count": 1,
|
||||
},
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
|
||||
side_effect=fake_remove,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
side_effect=fake_collect,
|
||||
).start()
|
||||
patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
pr_number=848,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertTrue(res.get("performed") or res.get("executed"))
|
||||
self.assertEqual(res.get("selection_mode"), "exact_pr")
|
||||
self.assertEqual(res.get("selected_pr_number"), 848)
|
||||
actions = res.get("actions") or []
|
||||
pr_numbers_touched = {
|
||||
a.get("pr_number") for a in actions if a.get("pr_number") is not None
|
||||
}
|
||||
self.assertTrue(pr_numbers_touched.issubset({None, 848}) or not pr_numbers_touched)
|
||||
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.assertEqual(remove_calls[0]["branch"], "fix/issue-844-exclude-epic-containers")
|
||||
self.assertEqual(len(deletes), 1)
|
||||
self.assertTrue(deletes[0].get("success"))
|
||||
self.assertTrue(deletes[0].get("after_worktree_removal"))
|
||||
self.assertEqual(len(delete_api_calls), 1)
|
||||
self.assertEqual(
|
||||
ownership_branches, ["fix/issue-844-exclude-epic-containers"]
|
||||
)
|
||||
|
||||
def test_exact_pr_unknown_fails_closed_without_mutation(self):
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
def fake_api(method, url, *args, **kwargs):
|
||||
if method == "GET" and "/pulls/99999" in url:
|
||||
raise RuntimeError("HTTP 404 Not Found")
|
||||
raise AssertionError(f"unexpected API call {method} {url}")
|
||||
|
||||
self.mock_api.side_effect = fake_api
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=True,
|
||||
pr_number=99999,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertFalse(res.get("performed"))
|
||||
self.assertEqual(res.get("blocker_kind"), "pr_unresolvable")
|
||||
self.assertIn("99999", " ".join(res.get("reasons") or []))
|
||||
|
||||
def test_exact_pr_not_merged_fails_closed(self):
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
def fake_api(method, url, *args, **kwargs):
|
||||
if method == "GET" and url.rstrip("/").endswith("/pulls/900"):
|
||||
return {
|
||||
"number": 900,
|
||||
"merged": False,
|
||||
"merged_at": None,
|
||||
"state": "open",
|
||||
"head": {"ref": "feat/x", "sha": "a" * 40},
|
||||
}
|
||||
raise AssertionError(f"unexpected {method} {url}")
|
||||
|
||||
self.mock_api.side_effect = fake_api
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
pr_number=900,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertFalse(res.get("performed"))
|
||||
self.assertEqual(res.get("blocker_kind"), "pr_not_merged")
|
||||
|
||||
def test_exact_pr_invalid_number_fails_closed(self):
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=True,
|
||||
pr_number=0,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("blocker_kind"), "invalid_pr_number")
|
||||
self.mock_api.assert_not_called()
|
||||
|
||||
def test_batch_mode_still_works_without_pr_number(self):
|
||||
"""Unfiltered batch path remains backward compatible."""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
self.mock_all.side_effect = lambda url, auth, limit=None: []
|
||||
self.mock_api.side_effect = lambda *a, **k: {}
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value={
|
||||
"entries": [],
|
||||
"reviewer_scratch_entries": [],
|
||||
"merged_pr_count": 0,
|
||||
},
|
||||
).start()
|
||||
res = gitea_reconcile_merged_cleanups(dry_run=True, remote="prgs", limit=10)
|
||||
self.assertTrue(res.get("success"))
|
||||
self.assertEqual(res.get("selection_mode"), "batch")
|
||||
self.assertIsNone(res.get("selected_pr_number"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,244 @@
|
||||
"""#855 AC4: an expired reviewer lease must not indefinitely protect an
|
||||
already-merged branch when no live claimant exists.
|
||||
|
||||
Two layers are covered:
|
||||
|
||||
* ``branch_cleanup_guard.assess_expired_reviewer_lease_reclaim`` — the pure,
|
||||
fail-closed reclaim decision. Every condition must be provably satisfied or
|
||||
the lease keeps protecting the branch.
|
||||
* ``gitea_mcp_server._collect_branch_ownership_records`` — the wiring that
|
||||
supplies authoritative evidence (PR merged state, owner-process liveness,
|
||||
competing ownership) to that decision, and flips an expired reviewer lease
|
||||
to reclaimable only under the full policy.
|
||||
|
||||
All inputs are fabricated; no real repository, lease, or credential is used.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import branch_cleanup_guard
|
||||
|
||||
mcp_server = importlib.import_module("gitea_mcp_server")
|
||||
|
||||
FAKE_AUTH = "token fake"
|
||||
REMOTE = "prgs"
|
||||
ORG = "Scaled-Tech-Consulting"
|
||||
REPO = "Gitea-Tools"
|
||||
HOST = "gitea.prgs.cc"
|
||||
BRANCH = "feat/issue-638-webui-app-shell-phase1"
|
||||
PR_NUMBER = 818
|
||||
|
||||
|
||||
class TestAssessExpiredReviewerLeaseReclaim(unittest.TestCase):
|
||||
"""Pure fail-closed reclaim decision (#855 AC4)."""
|
||||
|
||||
def _call(self, **overrides):
|
||||
base = dict(
|
||||
role="reviewer",
|
||||
status="expired",
|
||||
pr_merged=True,
|
||||
owner_pid_alive=False,
|
||||
competing_active_claimant=False,
|
||||
)
|
||||
base.update(overrides)
|
||||
return branch_cleanup_guard.assess_expired_reviewer_lease_reclaim(**base)
|
||||
|
||||
def test_full_policy_satisfied_allows_reclaim(self):
|
||||
out = self._call()
|
||||
self.assertTrue(out["reclaim_allowed"])
|
||||
self.assertEqual(out["reasons"], [])
|
||||
self.assertEqual(out["decision"], "reclaim_expired_reviewer_lease")
|
||||
|
||||
def test_stale_dead_process_reviewer_also_reclaimable(self):
|
||||
out = self._call(status="stale_dead_process")
|
||||
self.assertTrue(out["reclaim_allowed"])
|
||||
|
||||
def test_non_reviewer_role_never_reclaims(self):
|
||||
for role in ("author", "merger", "controller", "reconciler", "unknown"):
|
||||
with self.subTest(role=role):
|
||||
out = self._call(role=role)
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
self.assertTrue(out["reasons"])
|
||||
self.assertEqual(out["decision"], "keep_protecting")
|
||||
|
||||
def test_active_status_never_reclaims(self):
|
||||
out = self._call(status="active")
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
|
||||
def test_pr_not_merged_blocks_reclaim(self):
|
||||
out = self._call(pr_merged=False)
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
|
||||
def test_pr_merged_unknown_fails_closed(self):
|
||||
out = self._call(pr_merged=None)
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
|
||||
def test_owner_process_alive_blocks_reclaim(self):
|
||||
out = self._call(owner_pid_alive=True)
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
|
||||
def test_owner_liveness_unknown_fails_closed(self):
|
||||
out = self._call(owner_pid_alive=None)
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
|
||||
def test_competing_active_claimant_blocks_reclaim(self):
|
||||
out = self._call(competing_active_claimant=True)
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
|
||||
def test_competing_claimant_unknown_fails_closed(self):
|
||||
out = self._call(competing_active_claimant=None)
|
||||
self.assertFalse(out["reclaim_allowed"])
|
||||
|
||||
def test_reasons_never_leak_secrets(self):
|
||||
out = self._call(role="author")
|
||||
blob = " ".join(out["reasons"]).lower()
|
||||
self.assertNotIn("token", blob)
|
||||
self.assertNotIn("password", blob)
|
||||
|
||||
|
||||
class _FakeLease(dict):
|
||||
pass
|
||||
|
||||
|
||||
class TestCollectorExpiredReviewerReclaimWiring(unittest.TestCase):
|
||||
"""`_collect_branch_ownership_records` supplies authoritative evidence and
|
||||
flips an expired reviewer lease to reclaimable only under the full policy."""
|
||||
|
||||
def _run(
|
||||
self,
|
||||
*,
|
||||
lease_role="reviewer",
|
||||
lease_freshness="stale_dead_process",
|
||||
owner_pid_alive=False,
|
||||
pr_merged=True,
|
||||
extra_leases=None,
|
||||
worktree_on_branch=False,
|
||||
):
|
||||
lease = _FakeLease(
|
||||
role=lease_role,
|
||||
work_kind="pr",
|
||||
work_number=PR_NUMBER,
|
||||
branch=BRANCH,
|
||||
status="active",
|
||||
owner_pid=999999,
|
||||
remote=REMOTE,
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
host=HOST,
|
||||
freshness={
|
||||
"freshness": lease_freshness,
|
||||
"owner_pid": 999999,
|
||||
"owner_pid_alive": owner_pid_alive,
|
||||
"expired_by_time": lease_freshness == "expired",
|
||||
},
|
||||
)
|
||||
leases = [lease] + list(extra_leases or [])
|
||||
|
||||
pr_payload = {
|
||||
"number": PR_NUMBER,
|
||||
"merged": pr_merged,
|
||||
"merged_at": "2026-07-23T00:00:00Z" if pr_merged else None,
|
||||
"head": {"ref": BRANCH},
|
||||
}
|
||||
|
||||
def fake_api_request(method, url, *a, **k):
|
||||
if method == "GET" and f"/pulls/{PR_NUMBER}" in url:
|
||||
return pr_payload
|
||||
raise AssertionError(f"unexpected api_request {method} {url}")
|
||||
|
||||
wt_entries = []
|
||||
if worktree_on_branch:
|
||||
wt_entries = [{"branch": BRANCH, "path": f"/x/branches/{BRANCH}"}]
|
||||
|
||||
with patch.object(
|
||||
mcp_server.lease_lifecycle,
|
||||
"list_active_leases",
|
||||
return_value={"leases": leases},
|
||||
), patch.object(
|
||||
mcp_server.control_plane_db, "get_db", return_value=object(), create=True
|
||||
), patch.object(
|
||||
mcp_server.issue_lock_store, "iter_lock_files", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server.worktree_cleanup_audit,
|
||||
"list_worktrees",
|
||||
return_value=wt_entries,
|
||||
), patch.object(
|
||||
mcp_server, "api_get_all", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server, "api_request", side_effect=fake_api_request
|
||||
):
|
||||
return mcp_server._collect_branch_ownership_records(
|
||||
remote=REMOTE,
|
||||
host=HOST,
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
branch=BRANCH,
|
||||
pr_number=PR_NUMBER,
|
||||
project_root="/x",
|
||||
auth=FAKE_AUTH,
|
||||
base_api="https://gitea.prgs.cc/api/v1/repos/x/y",
|
||||
)
|
||||
|
||||
def _reviewer_records(self, bundle):
|
||||
return [
|
||||
rec
|
||||
for rec in bundle["records"]
|
||||
if rec.get("category")
|
||||
== branch_cleanup_guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE
|
||||
]
|
||||
|
||||
def test_merged_dead_uncontested_reviewer_lease_is_reclaimable(self):
|
||||
bundle = self._run()
|
||||
self.assertFalse(bundle["inventory_error"])
|
||||
recs = self._reviewer_records(bundle)
|
||||
self.assertEqual(len(recs), 1)
|
||||
self.assertTrue(recs[0]["reclaim_allowed"])
|
||||
# And the guard consequently does not block deletion on it.
|
||||
ownership = branch_cleanup_guard.assess_active_branch_ownership(
|
||||
remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST,
|
||||
records=bundle["records"],
|
||||
)
|
||||
self.assertFalse(ownership["block"])
|
||||
|
||||
def test_unmerged_pr_keeps_reviewer_lease_protective(self):
|
||||
bundle = self._run(pr_merged=False)
|
||||
recs = self._reviewer_records(bundle)
|
||||
self.assertEqual(len(recs), 1)
|
||||
self.assertFalse(recs[0]["reclaim_allowed"])
|
||||
ownership = branch_cleanup_guard.assess_active_branch_ownership(
|
||||
remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST,
|
||||
records=bundle["records"],
|
||||
)
|
||||
self.assertTrue(ownership["block"])
|
||||
|
||||
def test_owner_process_alive_keeps_reviewer_lease_protective(self):
|
||||
bundle = self._run(owner_pid_alive=True, lease_freshness="expired")
|
||||
recs = self._reviewer_records(bundle)
|
||||
self.assertFalse(recs[0]["reclaim_allowed"])
|
||||
|
||||
def test_competing_worktree_binding_keeps_reviewer_lease_protective(self):
|
||||
bundle = self._run(worktree_on_branch=True)
|
||||
recs = self._reviewer_records(bundle)
|
||||
self.assertFalse(recs[0]["reclaim_allowed"])
|
||||
ownership = branch_cleanup_guard.assess_active_branch_ownership(
|
||||
remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST,
|
||||
records=bundle["records"],
|
||||
)
|
||||
self.assertTrue(ownership["block"])
|
||||
|
||||
def test_expired_author_lease_never_reclaimed_by_reviewer_policy(self):
|
||||
bundle = self._run(lease_role="author")
|
||||
author_recs = [
|
||||
rec
|
||||
for rec in bundle["records"]
|
||||
if rec.get("category")
|
||||
== branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE
|
||||
]
|
||||
self.assertEqual(len(author_recs), 1)
|
||||
self.assertFalse(author_recs[0]["reclaim_allowed"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,630 @@
|
||||
"""Durable linked-issue lock head refresh + merge-sync dead-session recovery (#871).
|
||||
|
||||
``gitea_update_pr_branch_by_merge`` advances a PR's *remote* head but historically
|
||||
never advanced the linked durable issue lock's recorded head. After the owning
|
||||
session died the drifted lock became unrecoverable and no further synchronization
|
||||
was possible (PR #866 / issue #855).
|
||||
|
||||
Two halves are covered:
|
||||
|
||||
* the write-side refresh (``issue_lock_store.assess/apply_durable_lock_head_refresh``)
|
||||
that records the new synced head under compare-and-swap with read-after-write; and
|
||||
* the read-side recovery relation (``issue_lock_recovery`` +
|
||||
``issue_lock_worktree.read_merge_sync_provenance``) that lets a dead-session lock
|
||||
whose recorded head is a merge-sync *ancestor* of the live PR head be recovered —
|
||||
and nothing else.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import issue_lock_recovery # noqa: E402
|
||||
import issue_lock_store # noqa: E402
|
||||
import issue_lock_worktree # noqa: E402
|
||||
|
||||
ISSUE = 8710
|
||||
PR_NUMBER = 8711
|
||||
BRANCH = f"fix/issue-{ISSUE}-durable-lock-head-refresh"
|
||||
IDENTITY = "example-user"
|
||||
PROFILE = "example-author"
|
||||
OLD = "a" * 40
|
||||
NEW1 = "b" * 40
|
||||
NEW2 = "c" * 40
|
||||
BASE = "d" * 40
|
||||
REMOTE = "prgs"
|
||||
ORG = "ExampleOrg"
|
||||
REPO = "ExampleRepo"
|
||||
|
||||
|
||||
def dead_pid() -> int:
|
||||
proc = subprocess.Popen([sys.executable, "-c", "pass"])
|
||||
proc.wait()
|
||||
return proc.pid
|
||||
|
||||
|
||||
def future_ts(hours: int = 4) -> str:
|
||||
return (
|
||||
(datetime.now(timezone.utc) + timedelta(hours=hours))
|
||||
.isoformat()
|
||||
.replace("+00:00", "Z")
|
||||
)
|
||||
|
||||
|
||||
def _git(cwd, *args):
|
||||
return subprocess.run(
|
||||
["git", "-C", cwd, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
|
||||
|
||||
def _rev(cwd, ref="HEAD") -> str:
|
||||
return _git(cwd, "rev-parse", ref).stdout.strip()
|
||||
|
||||
|
||||
def build_merge_sync_repo(tmp: str) -> dict:
|
||||
"""Build a repo where a feature branch was synced by merging master in.
|
||||
|
||||
Returns a dict with the prior (branch) head, the synced merge-commit head,
|
||||
the master tip, plus a rebase-style linear descendant and an unrelated head.
|
||||
"""
|
||||
_git(tmp, "init", "-q", "-b", "master")
|
||||
_git(tmp, "config", "user.email", "[email protected]")
|
||||
_git(tmp, "config", "user.name", "T")
|
||||
Path(tmp, "base.txt").write_text("base\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "root")
|
||||
|
||||
# Feature branch cut from root, one commit — this is the PRIOR/recorded head.
|
||||
_git(tmp, "checkout", "-q", "-b", BRANCH)
|
||||
Path(tmp, "feature.txt").write_text("feature\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "feature work")
|
||||
prior = _rev(tmp)
|
||||
|
||||
# Master advances (the base the sync will merge in).
|
||||
_git(tmp, "checkout", "-q", "master")
|
||||
Path(tmp, "base.txt").write_text("base\nmore\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "master advance")
|
||||
master_tip = _rev(tmp)
|
||||
|
||||
# Sync: merge master INTO the feature branch → merge commit, first parent = prior.
|
||||
_git(tmp, "checkout", "-q", BRANCH)
|
||||
_git(tmp, "merge", "-q", "--no-ff", "-m", "Merge master into feature", "master")
|
||||
synced = _rev(tmp)
|
||||
|
||||
# A plain linear descendant of prior (NOT a merge) — a rebase/extra-commit shape.
|
||||
_git(tmp, "checkout", "-q", "-b", "linear-branch", prior)
|
||||
Path(tmp, "extra.txt").write_text("extra\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "extra linear commit")
|
||||
linear = _rev(tmp)
|
||||
|
||||
# An unrelated root (force-push / rewritten history shape).
|
||||
unrelated_dir = tempfile.mkdtemp()
|
||||
_git(unrelated_dir, "init", "-q", "-b", "x")
|
||||
_git(unrelated_dir, "config", "user.email", "[email protected]")
|
||||
_git(unrelated_dir, "config", "user.name", "T")
|
||||
Path(unrelated_dir, "z.txt").write_text("z\n")
|
||||
_git(unrelated_dir, "add", "-A")
|
||||
_git(unrelated_dir, "commit", "-q", "-m", "unrelated")
|
||||
unrelated = _rev(unrelated_dir)
|
||||
|
||||
# Leave the worktree checked out on the feature branch at the PRIOR head, as
|
||||
# a dead author session that never advanced would have left it.
|
||||
_git(tmp, "checkout", "-q", BRANCH)
|
||||
_git(tmp, "reset", "-q", "--hard", prior)
|
||||
|
||||
return {
|
||||
"prior": prior,
|
||||
"master_tip": master_tip,
|
||||
"synced": synced,
|
||||
"linear": linear,
|
||||
"unrelated": unrelated,
|
||||
}
|
||||
|
||||
|
||||
# ─────────────────────────── write-side refresh ───────────────────────────
|
||||
|
||||
|
||||
class TestDurableLockHeadRefresh(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.lock_dir = tempfile.mkdtemp()
|
||||
self.wt = tempfile.mkdtemp()
|
||||
lock_data = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": self.wt,
|
||||
"remote": REMOTE,
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"work_lease": {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": BRANCH,
|
||||
"worktree_path": self.wt,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"expires_at": future_ts(),
|
||||
},
|
||||
}
|
||||
issue_lock_store.bind_session_lock(lock_data, lock_dir=self.lock_dir)
|
||||
|
||||
def _apply(self, **over):
|
||||
kw = dict(
|
||||
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER,
|
||||
identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(),
|
||||
expected_old_head=OLD, new_head=NEW1, synced_at=future_ts(0),
|
||||
base_head=BASE, lock_dir=self.lock_dir,
|
||||
)
|
||||
kw.update(over)
|
||||
return issue_lock_store.apply_durable_lock_head_refresh(**kw)
|
||||
|
||||
def _load(self):
|
||||
return issue_lock_store.load_issue_lock(
|
||||
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
|
||||
def test_first_sync_updates_recorded_head(self):
|
||||
"""AC1: first base sync writes the resulting head to the durable lock."""
|
||||
res = self._apply()
|
||||
self.assertTrue(res["refreshed"], res["reasons"])
|
||||
self.assertTrue(res["read_after_write_ok"])
|
||||
self.assertEqual(self._load().get("synced_pr_head"), NEW1)
|
||||
|
||||
def test_second_sync_after_master_advance(self):
|
||||
"""AC2: a later master advance permits a second sanctioned sync."""
|
||||
self.assertTrue(self._apply()["refreshed"])
|
||||
res2 = self._apply(expected_old_head=NEW1, new_head=NEW2)
|
||||
self.assertTrue(res2["refreshed"], res2["reasons"])
|
||||
self.assertEqual(self._load().get("synced_pr_head"), NEW2)
|
||||
history = self._load().get("branch_sync_history")
|
||||
self.assertEqual(len(history), 2)
|
||||
self.assertEqual(history[0]["last_synced_pr_head"], NEW1)
|
||||
self.assertEqual(history[1]["prior_pr_head"], NEW1)
|
||||
|
||||
def test_cas_detects_concurrent_head_change(self):
|
||||
"""AC6: CAS refuses when the recorded synced head is not the old head."""
|
||||
self.assertTrue(self._apply()["refreshed"]) # recorded head now NEW1
|
||||
# A second sync claiming the old head is still OLD must fail closed.
|
||||
res = self._apply(expected_old_head=OLD, new_head=NEW2)
|
||||
self.assertFalse(res["refreshed"])
|
||||
self.assertTrue(any("CAS" in r or "concurrent" in r for r in res["reasons"]))
|
||||
self.assertEqual(self._load().get("synced_pr_head"), NEW1)
|
||||
|
||||
def test_wrong_issue_fails_closed(self):
|
||||
res = self._apply(issue_number=999999)
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_branch_fails_closed(self):
|
||||
res = self._apply(branch_name="fix/issue-8710-wrong")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_repo_fails_closed(self):
|
||||
res = self._apply(repo="OtherRepo")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_identity_fails_closed(self):
|
||||
res = self._apply(identity="intruder")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_profile_fails_closed(self):
|
||||
res = self._apply(profile="prgs-reviewer")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_foreign_session_fails_closed(self):
|
||||
"""A refresh is not a recovery: the current process must own the lock."""
|
||||
path = issue_lock_store.lock_file_path(
|
||||
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
rec = issue_lock_store.read_lock_file(path)
|
||||
rec["session_pid"] = dead_pid()
|
||||
rec["pid"] = rec["session_pid"]
|
||||
issue_lock_store.save_lock_file(path, rec)
|
||||
res = self._apply()
|
||||
self.assertFalse(res["refreshed"])
|
||||
self.assertTrue(any("current session" in r or "live owner" in r for r in res["reasons"]))
|
||||
|
||||
def test_new_equals_old_fails_closed(self):
|
||||
res = self._apply(expected_old_head=OLD, new_head=OLD)
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_non_full_sha_fails_closed(self):
|
||||
self.assertFalse(self._apply(new_head="deadbeef")["refreshed"])
|
||||
self.assertFalse(self._apply(expected_old_head="xyz")["refreshed"])
|
||||
|
||||
def test_no_lock_fails_closed(self):
|
||||
assessment = issue_lock_store.assess_durable_lock_head_refresh(
|
||||
None, remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER,
|
||||
identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(),
|
||||
expected_old_head=OLD, new_head=NEW1,
|
||||
)
|
||||
self.assertFalse(assessment["allowed"])
|
||||
|
||||
|
||||
# ─────────────────────── merge-sync provenance (real git) ───────────────────
|
||||
|
||||
|
||||
class TestMergeSyncProvenanceObservation(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
self.shas = build_merge_sync_repo(self.tmp)
|
||||
|
||||
def test_merge_sync_is_recognized(self):
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=self.shas["prior"],
|
||||
synced_head_sha=self.shas["synced"],
|
||||
)
|
||||
self.assertTrue(obs["is_merge_sync"], obs["reasons"])
|
||||
self.assertTrue(obs["prior_is_ancestor"])
|
||||
self.assertTrue(obs["synced_is_merge"])
|
||||
self.assertTrue(obs["first_parent_reaches_prior"])
|
||||
|
||||
def test_linear_descendant_is_not_a_merge_sync(self):
|
||||
"""A plain non-merge descendant (rebase/extra commit) is not a sync."""
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=self.shas["prior"],
|
||||
synced_head_sha=self.shas["linear"],
|
||||
)
|
||||
self.assertTrue(obs["probe_ok"])
|
||||
self.assertFalse(obs["is_merge_sync"])
|
||||
self.assertFalse(obs["synced_is_merge"])
|
||||
|
||||
def test_unrelated_history_fails_closed(self):
|
||||
"""A rewritten/force-pushed head where prior is unreachable fails closed."""
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=self.shas["prior"],
|
||||
synced_head_sha=self.shas["unrelated"],
|
||||
)
|
||||
self.assertFalse(obs["is_merge_sync"])
|
||||
|
||||
def test_missing_args_fail_closed(self):
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=None, synced_head_sha=self.shas["synced"],
|
||||
)
|
||||
self.assertFalse(obs["is_merge_sync"])
|
||||
|
||||
|
||||
# ──────────────────── merge-sync dead-session recovery ──────────────────────
|
||||
|
||||
|
||||
def make_dead_lock(worktree, **over):
|
||||
pid = dead_pid()
|
||||
lock = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": worktree,
|
||||
"remote": REMOTE,
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"session_pid": pid,
|
||||
"pid": pid,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"work_lease": {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": BRANCH,
|
||||
"worktree_path": worktree,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"expires_at": future_ts(),
|
||||
},
|
||||
}
|
||||
lock.update(over)
|
||||
return lock
|
||||
|
||||
|
||||
def sync_prov(prior, synced, **over):
|
||||
d = {
|
||||
"prior_head_sha": prior,
|
||||
"synced_head_sha": synced,
|
||||
"probe_ok": True,
|
||||
"prior_present": True,
|
||||
"synced_present": True,
|
||||
"prior_is_ancestor": True,
|
||||
"synced_is_merge": True,
|
||||
"first_parent_reaches_prior": True,
|
||||
"is_merge_sync": True,
|
||||
"first_parent_sha": prior,
|
||||
"parent_count": 2,
|
||||
"proof": f"{synced} merged base into branch above {prior}",
|
||||
"reasons": [],
|
||||
}
|
||||
d.update(over)
|
||||
return d
|
||||
|
||||
|
||||
class TestMergeSyncRecovery(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
self.shas = build_merge_sync_repo(self.tmp)
|
||||
self.prior = self.shas["prior"]
|
||||
self.synced = self.shas["synced"]
|
||||
|
||||
def _assess(self, **over):
|
||||
lock = over.pop("_lock", None) or make_dead_lock(self.tmp)
|
||||
kw = dict(
|
||||
issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp,
|
||||
remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE,
|
||||
current_branch=BRANCH, porcelain_status="",
|
||||
head_sha=self.prior, remote_head_sha=self.synced,
|
||||
pr_head_sha=self.synced, pr_number=PR_NUMBER,
|
||||
competing_live_locks=[], candidate_branches=[BRANCH],
|
||||
current_pid=os.getpid(),
|
||||
remote_branch_exists=True,
|
||||
sync_provenance=sync_prov(self.prior, self.synced),
|
||||
)
|
||||
kw.update(over)
|
||||
return issue_lock_recovery.assess_dead_session_lock_recovery(lock, **kw)
|
||||
|
||||
def test_merge_sync_drift_is_recoverable(self):
|
||||
"""AC3/AC4: dead session, recorded head is a merge-sync ancestor of PR head."""
|
||||
res = self._assess()
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
|
||||
self.assertEqual(
|
||||
res["evidence"]["head_relation"],
|
||||
issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
|
||||
)
|
||||
self.assertEqual(res["evidence"]["accepted_head"], self.synced)
|
||||
|
||||
def test_missing_provenance_fails_closed(self):
|
||||
"""No server-derived provenance → cannot accept a remote ahead of local."""
|
||||
res = self._assess(sync_provenance=None)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_non_ancestor_recorded_head_fails_closed(self):
|
||||
"""AC7: provenance that does not prove ancestry is rejected."""
|
||||
res = self._assess(
|
||||
sync_provenance=sync_prov(
|
||||
self.prior, self.synced, prior_is_ancestor=False, is_merge_sync=False,
|
||||
reasons=["prior head is not an ancestor"],
|
||||
)
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_force_pushed_history_fails_closed(self):
|
||||
"""AC8: a rewritten head (not a merge sync) stays protected."""
|
||||
res = self._assess(
|
||||
sync_provenance=sync_prov(
|
||||
self.prior, self.synced, is_merge_sync=False, synced_is_merge=False,
|
||||
reasons=["not a merge-based sync"],
|
||||
)
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_provenance_for_other_commits_fails_closed(self):
|
||||
"""Provenance whose endpoints differ from the heads under assessment is rejected."""
|
||||
res = self._assess(
|
||||
sync_provenance=sync_prov("f" * 40, self.synced),
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_dirty_worktree_fails_closed(self):
|
||||
"""AC11: dirty worktrees remain protected."""
|
||||
res = self._assess(porcelain_status=" M feature.txt\n")
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_live_owner_fails_closed(self):
|
||||
"""AC10: a live recorded owner is not a dead-session recovery."""
|
||||
lock = make_dead_lock(self.tmp, session_pid=os.getpid(), pid=os.getpid())
|
||||
res = self._assess(_lock=lock)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_competing_claimant_fails_closed(self):
|
||||
"""AC13: a competing live lock blocks recovery."""
|
||||
res = self._assess(
|
||||
competing_live_locks=[{
|
||||
"issue_number": ISSUE, "branch_name": BRANCH,
|
||||
"worktree_path": "/some/other/wt", "pid": os.getpid(),
|
||||
}]
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_wrong_branch_fails_closed(self):
|
||||
"""AC9: worktree on a different branch fails closed."""
|
||||
res = self._assess(current_branch="fix/issue-8710-other")
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_wrong_identity_fails_closed(self):
|
||||
res = self._assess(identity="intruder")
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_pr_head_mismatch_fails_closed(self):
|
||||
"""The open PR must sit at the synced remote head."""
|
||||
res = self._assess(pr_head_sha="e" * 40)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_owning_pr_evidence_for_merge_sync(self):
|
||||
res = self._assess()
|
||||
ev = issue_lock_recovery.owning_pr_recovery_evidence(res)
|
||||
self.assertIsNotNone(ev)
|
||||
self.assertEqual(ev["pr_number"], PR_NUMBER)
|
||||
self.assertEqual(ev["head_sha"], self.synced)
|
||||
self.assertEqual(
|
||||
ev["head_relation"],
|
||||
issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
|
||||
)
|
||||
|
||||
def test_recovered_owning_pr_from_persisted_record(self):
|
||||
res = self._assess()
|
||||
record = issue_lock_recovery.build_recovery_record(res, recovered_at=future_ts(0))
|
||||
lock = {"issue_number": ISSUE, "branch_name": BRANCH,
|
||||
"dead_session_recovery": record}
|
||||
rebuilt = issue_lock_recovery.recovered_owning_pr_from_lock(lock)
|
||||
self.assertIsNotNone(rebuilt)
|
||||
self.assertEqual(rebuilt["head_sha"], self.synced)
|
||||
self.assertEqual(
|
||||
rebuilt["head_relation"],
|
||||
issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
|
||||
)
|
||||
|
||||
|
||||
class TestExistingRelationsUnchanged(unittest.TestCase):
|
||||
"""AC14/AC15: equal-head recovery still works; merge-sync did not weaken it."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
self.shas = build_merge_sync_repo(self.tmp)
|
||||
|
||||
def test_equal_head_recovery_still_sanctioned(self):
|
||||
# Worktree at prior head; remote also at prior head → the #753 equal case.
|
||||
prior = self.shas["prior"]
|
||||
lock = make_dead_lock(self.tmp)
|
||||
res = issue_lock_recovery.assess_dead_session_lock_recovery(
|
||||
lock, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp,
|
||||
remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE,
|
||||
current_branch=BRANCH, porcelain_status="",
|
||||
head_sha=prior, remote_head_sha=prior,
|
||||
pr_head_sha=prior, pr_number=PR_NUMBER,
|
||||
competing_live_locks=[], candidate_branches=[BRANCH],
|
||||
current_pid=os.getpid(), remote_branch_exists=True,
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
|
||||
self.assertEqual(
|
||||
res["evidence"]["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL,
|
||||
)
|
||||
|
||||
|
||||
class TestUpdatePrWrapperPartialFailure(unittest.TestCase):
|
||||
"""AC5/AC16: the tool advances the remote head then refreshes the durable lock.
|
||||
|
||||
When the durable refresh fails after the remote advance, the tool must report a
|
||||
partial lifecycle failure and NOT a fully successful synchronization. Exact PR-
|
||||
head / base-head pinning is preserved (delegated to the real preflight, stubbed
|
||||
here only to isolate the post-update lifecycle branch).
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms # noqa: E402
|
||||
self.gms = gms
|
||||
self._orig = {}
|
||||
|
||||
def _patch(name, value):
|
||||
self._orig[name] = getattr(gms, name)
|
||||
setattr(gms, name, value)
|
||||
|
||||
_patch("get_profile", lambda *a, **k: {
|
||||
"allowed_operations": ["gitea.branch.push"],
|
||||
"forbidden_operations": [],
|
||||
"profile_name": "prgs-author",
|
||||
})
|
||||
_patch("_role_kind", lambda *a, **k: "author")
|
||||
_patch("_profile_operation_gate", lambda *a, **k: None)
|
||||
_patch("_permission_block_report", lambda *a, **k: {})
|
||||
_patch("_resolve", lambda *a, **k: ("gitea.prgs.cc", ORG, REPO))
|
||||
_patch("_verify_role_mutation_workspace", lambda *a, **k: None)
|
||||
_patch("_get_workspace_porcelain", lambda *a, **k: "")
|
||||
_patch("_canonical_local_git_root", lambda *a, **k: "/x")
|
||||
_patch("_master_parity_block", lambda *a, **k: None)
|
||||
_patch("_auth", lambda *a, **k: {"token": "x"})
|
||||
_patch("repo_api_url", lambda *a, **k: "http://api")
|
||||
_patch("_redact", lambda s: s)
|
||||
_patch("_work_lease_claimant", lambda *a, **k: {
|
||||
"username": IDENTITY, "profile": PROFILE,
|
||||
})
|
||||
_patch("_prove_author_ownership_for_pr", lambda *a, **k: {
|
||||
"has_author_lock": True, "matched_issue": ISSUE,
|
||||
"matched_via": "branch", "linked_issues": [ISSUE],
|
||||
"recovered_owning_pr": None, "reasons": [],
|
||||
})
|
||||
|
||||
# Real preflight is unit-tested elsewhere; stub it to isolate the
|
||||
# post-update durable-lock lifecycle branch under test.
|
||||
orig_pf = gms.pr_sync_status.assess_update_pr_branch_preflight
|
||||
self._orig_pf = orig_pf
|
||||
gms.pr_sync_status.assess_update_pr_branch_preflight = (
|
||||
lambda *a, **k: {"mutation_allowed": True, "reasons": [], "performed": False}
|
||||
)
|
||||
|
||||
# Sequence the two GET /pulls calls: OLD before update, NEW after.
|
||||
self._pull_calls = {"n": 0}
|
||||
|
||||
def fake_api_request(method, url, auth, *a, **k):
|
||||
m = method.upper()
|
||||
if m == "GET" and url.endswith(f"/pulls/{PR_NUMBER}"):
|
||||
self._pull_calls["n"] += 1
|
||||
head = OLD if self._pull_calls["n"] == 1 else NEW1
|
||||
return {
|
||||
"state": "open",
|
||||
"head": {"sha": head, "ref": BRANCH},
|
||||
"base": {"sha": BASE, "ref": "master"},
|
||||
"mergeable": True, "title": "t", "body": "b",
|
||||
}
|
||||
if m == "GET" and "/branches/" in url:
|
||||
return {"commit": {"id": BASE}}
|
||||
if m == "POST" and "/update" in url:
|
||||
return {}
|
||||
return {}
|
||||
|
||||
_patch("api_request", fake_api_request)
|
||||
|
||||
def tearDown(self):
|
||||
for name, value in self._orig.items():
|
||||
setattr(self.gms, name, value)
|
||||
self.gms.pr_sync_status.assess_update_pr_branch_preflight = self._orig_pf
|
||||
|
||||
def _run(self):
|
||||
return self.gms.gitea_update_pr_branch_by_merge(
|
||||
pr_number=PR_NUMBER,
|
||||
expected_pr_head_sha=OLD,
|
||||
expected_base_head_sha=BASE,
|
||||
remote=REMOTE,
|
||||
worktree_path="/tmp/branches/wt-871",
|
||||
)
|
||||
|
||||
def test_partial_failure_when_refresh_fails(self):
|
||||
self._orig["apply_durable_lock_head_refresh"] = (
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh
|
||||
)
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
lambda **k: {"refreshed": False, "reasons": ["forced refresh failure"]}
|
||||
)
|
||||
try:
|
||||
res = self._run()
|
||||
finally:
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
self._orig["apply_durable_lock_head_refresh"]
|
||||
)
|
||||
self.assertTrue(res["performed"])
|
||||
self.assertEqual(res["new_pr_head_sha"], NEW1)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertTrue(res["partial_lifecycle_failure"])
|
||||
self.assertFalse(res["durable_lock_refreshed"])
|
||||
|
||||
def test_full_success_when_refresh_succeeds(self):
|
||||
self._orig["apply_durable_lock_head_refresh"] = (
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh
|
||||
)
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
lambda **k: {"refreshed": True, "read_after_write_ok": True,
|
||||
"new_head": NEW1, "reasons": ["ok"]}
|
||||
)
|
||||
try:
|
||||
res = self._run()
|
||||
finally:
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
self._orig["apply_durable_lock_head_refresh"]
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertTrue(res["performed"])
|
||||
self.assertTrue(res["durable_lock_refreshed"])
|
||||
self.assertTrue(res["fully_synchronized"])
|
||||
self.assertEqual(res["new_pr_head_sha"], NEW1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Documentation acceptance for the MCP restart governance ADR (#656).
|
||||
|
||||
Enforces issue #656 acceptance criteria:
|
||||
|
||||
* AC1 — policy document exists with an authorization matrix and the recorded
|
||||
v1 decision (controller approval + automated safety gates).
|
||||
* AC2 — restart is stated as a last resort with enumerated narrower recoveries.
|
||||
* AC3 — a unilateral LLM full restart with affected sessions is forbidden.
|
||||
* AC4 — break-glass conditions are listed.
|
||||
* AC5 — the ADR is linked to #655, #652, #653, #630, #642, and is cross-linked
|
||||
from the safety model and the web-console deployment boundary docs.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
ADR = REPO_ROOT / "docs" / "architecture" / "mcp-restart-governance.md"
|
||||
ADR_BASENAME = "mcp-restart-governance.md"
|
||||
|
||||
CROSS_LINK_DOCS = (
|
||||
REPO_ROOT / "docs" / "safety-model.md",
|
||||
REPO_ROOT / "docs" / "webui-deployment.md",
|
||||
)
|
||||
|
||||
LINKED_ISSUES = ("#655", "#652", "#653", "#630", "#642")
|
||||
POLICY_IDS = ("RG-01", "RG-02", "RG-03", "RG-04", "RG-05", "RG-06", "RG-07", "RG-08")
|
||||
|
||||
|
||||
def _read(path: Path) -> str:
|
||||
assert path.is_file(), f"missing {path.relative_to(REPO_ROOT)}"
|
||||
return path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_ac1_adr_exists_with_matrix_and_v1_decision():
|
||||
text = _read(ADR)
|
||||
lower = text.lower()
|
||||
assert text.lstrip().startswith("#"), "ADR lacks a title"
|
||||
assert "#656" in text
|
||||
assert "authorization matrix" in lower
|
||||
# The matrix is a real table with the worker and privileged roles.
|
||||
for role in ("author", "reviewer", "merger", "reconciler", "controller",
|
||||
"operator", "admin"):
|
||||
assert role in lower, f"authorization matrix missing role {role!r}"
|
||||
# Recorded v1 decision.
|
||||
assert "restart-governance/v1" in text
|
||||
assert "controller approval" in lower and "automated safety gates" in lower
|
||||
|
||||
|
||||
def test_ac2_restart_is_last_resort_with_narrower_recoveries():
|
||||
text = _read(ADR)
|
||||
lower = text.lower()
|
||||
assert "last resort" in lower
|
||||
# Enumerated narrower recoveries precede full restart on the ladder.
|
||||
for rung in ("reconnect", "rebind", "scoped restart", "full restart",
|
||||
"host"):
|
||||
assert rung in lower, f"recovery ladder missing rung {rung!r}"
|
||||
|
||||
|
||||
def test_ac3_forbids_unilateral_llm_full_restart_with_affected_sessions():
|
||||
text = _read(ADR)
|
||||
lower = text.lower()
|
||||
assert "forbidden" in lower
|
||||
assert "llm" in lower and "restart" in lower
|
||||
assert "unilateral" in lower
|
||||
# A worker role must not perform or authorize full/host restart.
|
||||
assert "must not" in lower
|
||||
|
||||
|
||||
def test_ac4_break_glass_conditions_listed():
|
||||
text = _read(ADR)
|
||||
lower = text.lower()
|
||||
assert "break-glass" in lower
|
||||
assert "incident" in lower
|
||||
assert "audit" in lower
|
||||
|
||||
|
||||
def test_ac5_adr_links_issue_lineage():
|
||||
text = _read(ADR)
|
||||
for issue in LINKED_ISSUES:
|
||||
assert issue in text, f"ADR must link issue {issue}"
|
||||
|
||||
|
||||
def test_ac5_safety_model_and_deployment_cross_link_adr():
|
||||
for path in CROSS_LINK_DOCS:
|
||||
text = _read(path)
|
||||
assert ADR_BASENAME in text, (
|
||||
f"{path.relative_to(REPO_ROOT)} must cross-link {ADR_BASENAME} "
|
||||
f"(issue #656 acceptance criterion 5)"
|
||||
)
|
||||
|
||||
|
||||
def test_policy_ids_present_for_enforcement_code():
|
||||
text = _read(ADR)
|
||||
for pid in POLICY_IDS:
|
||||
assert pid in text, f"policy id {pid} missing from ADR"
|
||||
|
||||
|
||||
def test_failure_behavior_denies_on_ambiguity():
|
||||
text = _read(ADR)
|
||||
lower = text.lower()
|
||||
assert "ambiguous" in lower and "deny" in lower
|
||||
|
||||
|
||||
def test_cross_links_do_not_embed_secrets():
|
||||
for path in (ADR,) + CROSS_LINK_DOCS:
|
||||
text = _read(path)
|
||||
for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
|
||||
assert marker not in text, f"{path} contains {marker!r}"
|
||||
@@ -0,0 +1,146 @@
|
||||
"""Tests for the MCP restart-path inventory and guards (#657).
|
||||
|
||||
Covers:
|
||||
* the registry is well-formed and every path is classified;
|
||||
* unknown restart attempts fail closed (AC "fail closed on unknown restart");
|
||||
* the previously-unguarded full-restart primitives stay guarded/absent
|
||||
against the real source tree (AC "tests for at least one previously
|
||||
unguarded path");
|
||||
* pkill of the daemon is still classified as contamination (#630, AC3);
|
||||
* the inventory doc and module stay in lock-step.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import mcp_restart_paths as rp
|
||||
import runtime_recovery_guard
|
||||
|
||||
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
DOC_PATH = os.path.join(REPO_ROOT, "docs", "mcp-restart-path-inventory.md")
|
||||
|
||||
|
||||
class TestRegistryWellformed(unittest.TestCase):
|
||||
def test_registry_is_wellformed(self):
|
||||
# Must not raise.
|
||||
rp.assert_registry_wellformed()
|
||||
|
||||
def test_every_path_has_valid_classification(self):
|
||||
for path in rp.iter_restart_paths():
|
||||
self.assertIn(path.classification, rp.VALID_CLASSIFICATIONS)
|
||||
self.assertTrue(path.guard.strip(), path.path_id)
|
||||
self.assertTrue(path.references, path.path_id)
|
||||
self.assertTrue(path.locations, path.path_id)
|
||||
|
||||
def test_ids_are_unique(self):
|
||||
ids = [p.path_id for p in rp.iter_restart_paths()]
|
||||
self.assertEqual(len(ids), len(set(ids)))
|
||||
|
||||
def test_covers_every_classification(self):
|
||||
present = {p.classification for p in rp.iter_restart_paths()}
|
||||
self.assertEqual(present, set(rp.VALID_CLASSIFICATIONS))
|
||||
|
||||
|
||||
class TestUnknownAttemptFailsClosed(unittest.TestCase):
|
||||
def test_unknown_path_raises(self):
|
||||
with self.assertRaises(rp.UnknownRestartPathError):
|
||||
rp.assert_restart_attempt_registered("totally_novel_restart_hack")
|
||||
|
||||
def test_get_unknown_raises(self):
|
||||
with self.assertRaises(rp.UnknownRestartPathError):
|
||||
rp.get_restart_path("nope")
|
||||
|
||||
def test_registered_attempt_returns_path(self):
|
||||
path = rp.assert_restart_attempt_registered("manual_daemon_kill")
|
||||
self.assertEqual(path.classification, rp.CLASS_FORBIDDEN)
|
||||
|
||||
|
||||
class TestDaemonNeverSelfReplaces(unittest.TestCase):
|
||||
"""Previously-unguarded full-restart primitive: daemon self-replacement."""
|
||||
|
||||
def test_no_self_replacement_in_source(self):
|
||||
# The live daemon modules must contain no os.execv/os.kill/os._exit
|
||||
# self-restart call. Must not raise.
|
||||
rp.assert_no_daemon_self_replacement(REPO_ROOT)
|
||||
|
||||
def test_scanner_flags_injected_violation(self):
|
||||
# Guard the guard: prove the scanner catches a real self-replace call.
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
bad = Path(tmp) / "gitea_mcp_server.py"
|
||||
bad.write_text(
|
||||
"import os\n"
|
||||
"def restart():\n"
|
||||
" os.execv('/usr/bin/python', ['python'])\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
found = rp.scan_daemon_self_replacement(tmp)
|
||||
self.assertTrue(found)
|
||||
with self.assertRaises(AssertionError):
|
||||
rp.assert_no_daemon_self_replacement(tmp)
|
||||
|
||||
def test_scanner_ignores_comment_and_docstring_mentions(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
ok = Path(tmp) / "gitea_mcp_server.py"
|
||||
ok.write_text(
|
||||
"import os\n"
|
||||
"# NOT os.execv() to re-point the interpreter here.\n"
|
||||
'"""Never calls os._exit to restart."""\n'
|
||||
"value = 1\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
self.assertEqual(rp.scan_daemon_self_replacement(tmp), [])
|
||||
|
||||
|
||||
class TestLegacyAutoRestartHelperRemoved(unittest.TestCase):
|
||||
"""Previously-unguarded full-restart path: _trigger_mcp_auto_restart."""
|
||||
|
||||
def test_helper_absent_in_source(self):
|
||||
# Must not raise: helper was removed in #685.
|
||||
rp.assert_auto_restart_helper_absent(REPO_ROOT)
|
||||
|
||||
def test_scanner_flags_reintroduced_helper(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
bad = Path(tmp) / "mcp_server.py"
|
||||
bad.write_text(
|
||||
"def _trigger_mcp_auto_restart():\n return True\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
with self.assertRaises(AssertionError):
|
||||
rp.assert_auto_restart_helper_absent(tmp)
|
||||
|
||||
|
||||
class TestPkillStaysForbidden(unittest.TestCase):
|
||||
"""AC3: pkill of the daemon remains forbidden/contaminating (#630)."""
|
||||
|
||||
def test_manual_daemon_kill_registered_as_forbidden(self):
|
||||
path = rp.get_restart_path("manual_daemon_kill")
|
||||
self.assertEqual(path.classification, rp.CLASS_FORBIDDEN)
|
||||
|
||||
def test_pkill_classified_as_contamination(self):
|
||||
assessment = runtime_recovery_guard.assess_recovery_command(
|
||||
"pkill -f mcp_server.py"
|
||||
)
|
||||
self.assertTrue(assessment["contaminated"])
|
||||
|
||||
def test_read_only_probe_not_contamination(self):
|
||||
assessment = runtime_recovery_guard.assess_recovery_command(
|
||||
"ps aux | grep mcp_server"
|
||||
)
|
||||
self.assertFalse(assessment["contaminated"])
|
||||
|
||||
|
||||
class TestInventoryDocInSync(unittest.TestCase):
|
||||
def test_doc_exists(self):
|
||||
self.assertTrue(os.path.exists(DOC_PATH), DOC_PATH)
|
||||
|
||||
def test_doc_mentions_every_path_id(self):
|
||||
with open(DOC_PATH, encoding="utf-8") as handle:
|
||||
doc = handle.read()
|
||||
for path in rp.iter_restart_paths():
|
||||
self.assertIn(path.path_id, doc, f"doc missing {path.path_id}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user