Gitea-Tools/tests/test_issue_855_expired_reviewer_reclaim.py

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