"""#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()