Gitea-Tools/tests/test_issue_760_exact_owner_lease_renewal.py

"""Exact-owner renewal of an expired author issue lease (#760).

Covers the renewal disposition that lets the exact recorded owner re-acquire
its own lock after the wall-clock lease expires — including while the recording
MCP daemon PID is still alive — plus every rejection condition that must keep
failing closed, and the pre-existing dead-PID and live-foreign dispositions
that must remain untouched.
"""

import inspect
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))

import issue_lock_renewal  # noqa: E402
import issue_lock_store  # noqa: E402

ISSUE = 5150
BRANCH = f"fix/issue-{ISSUE}-demo"
WORKTREE = "/scratch/wt-5150"
HEAD = "c" * 40
OTHER_SHA = "d" * 40
IDENTITY = "example-user"
PROFILE = "example-author"
REMOTE = "prgs"
ORG = "ExampleOrg"
REPO = "ExampleRepo"


def dead_pid() -> int:
    """A PID that has certainly exited (spawned, then reaped)."""
    proc = subprocess.Popen([sys.executable, "-c", "pass"])
    proc.wait()
    return proc.pid


def past_ts(hours: int = 1) -> str:
    return (
        (datetime.now(timezone.utc) - timedelta(hours=hours))
        .isoformat()
        .replace("+00:00", "Z")
    )


def future_ts(hours: int = 4) -> str:
    return (
        (datetime.now(timezone.utc) + timedelta(hours=hours))
        .isoformat()
        .replace("+00:00", "Z")
    )


def make_lock(*, expires_at: str | None = None, pid: int | None = None, **overrides):
    """An expired lock owned by a still-alive daemon PID — the #760 condition."""
    lock = {
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "worktree_path": WORKTREE,
        "remote": REMOTE,
        "org": ORG,
        "repo": REPO,
        # os.getpid() is unambiguously alive: the whole point of #760 is that
        # daemon liveness is not evidence of an active author task.
        "session_pid": os.getpid() if pid is None else pid,
        "lock_generation": 3,
        "work_lease": {
            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
            "issue_number": ISSUE,
            "branch": BRANCH,
            "worktree_path": WORKTREE,
            "claimant": {"username": IDENTITY, "profile": PROFILE},
            "created_at": past_ts(5),
            "expires_at": expires_at or past_ts(),
        },
    }
    lease_overrides = overrides.pop("work_lease", None)
    if lease_overrides:
        lock["work_lease"].update(lease_overrides)
    lock.update(overrides)
    return lock


def assess(lock=None, **overrides):
    """Run the assessor with all-passing evidence unless overridden."""
    kwargs = {
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "worktree_path": WORKTREE,
        "remote": REMOTE,
        "org": ORG,
        "repo": REPO,
        "identity": IDENTITY,
        "profile": PROFILE,
        "current_branch": BRANCH,
        "porcelain_status": "",
        "worktree_exists": True,
        "head_sha": HEAD,
        "remote_head_sha": HEAD,
        "pr_head_sha": None,
        "pr_number": None,
        "competing_live_locks": [],
        "candidate_branches": [BRANCH],
        "current_pid": 4242,
    }
    kwargs.update(overrides)
    return issue_lock_renewal.assess_exact_owner_lease_renewal(
        make_lock() if lock is None else lock, **kwargs
    )


class ExactOwnerRenewalGranted(unittest.TestCase):
    """AC1/AC3-AC7: the positive path."""

    def test_expired_lease_alive_pid_exact_owner_is_renewable(self):
        result = assess()
        self.assertEqual(result["outcome"], issue_lock_renewal.RENEWAL_SANCTIONED)
        self.assertTrue(result["renewal_sanctioned"])
        self.assertTrue(result["is_candidate"])

    def test_renewal_holds_when_owning_pr_head_matches(self):
        result = assess(pr_number=999, pr_head_sha=HEAD)
        self.assertTrue(result["renewal_sanctioned"])

    def test_evidence_records_both_sides_of_the_transition(self):
        result = assess()
        evidence = result["evidence"]
        self.assertEqual(evidence["prior_pid"], os.getpid())
        self.assertTrue(evidence["prior_pid_alive"])
        self.assertEqual(evidence["replacement_pid"], 4242)
        self.assertTrue(evidence["prior_expires_at"])


class ExactOwnerRenewalRefused(unittest.TestCase):
    """AC3-AC8: every near-match must fail closed, one reason at a time."""

    def _refused(self, **overrides):
        result = assess(**overrides)
        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
        self.assertFalse(result["renewal_sanctioned"])
        self.assertTrue(result["reasons"])
        return result

    def test_different_branch_refused(self):
        result = self._refused(branch_name=f"fix/issue-{ISSUE}-other")
        self.assertTrue(any("branch" in r for r in result["reasons"]))

    def test_different_worktree_refused(self):
        result = self._refused(worktree_path="/scratch/somewhere-else")
        self.assertTrue(any("worktree" in r for r in result["reasons"]))

    def test_different_claimant_refused(self):
        result = self._refused(identity="someone-else")
        self.assertTrue(any("claimant" in r for r in result["reasons"]))

    def test_different_profile_refused(self):
        result = self._refused(profile="other-author")
        self.assertTrue(any("profile" in r for r in result["reasons"]))

    def test_different_remote_org_or_repo_refused(self):
        self._refused(remote="dadeschools")
        self._refused(org="OtherOrg")
        self._refused(repo="OtherRepo")

    def test_dirty_worktree_refused(self):
        result = self._refused(porcelain_status=" M gitea_mcp_server.py\n")
        self.assertTrue(any("uncommitted" in r for r in result["reasons"]))

    def test_missing_worktree_refused(self):
        result = self._refused(worktree_exists=False)
        self.assertTrue(any("does not exist" in r for r in result["reasons"]))

    def test_worktree_on_wrong_branch_refused(self):
        self._refused(current_branch="master")

    def test_local_and_remote_head_mismatch_refused(self):
        result = self._refused(remote_head_sha=OTHER_SHA)
        self.assertTrue(
            any("does not equal remote head" in r for r in result["reasons"])
        )

    def test_unpublished_branch_refused(self):
        result = self._refused(remote_head_sha=None)
        self.assertTrue(any("remote branch head" in r for r in result["reasons"]))

    def test_pr_head_mismatch_refused(self):
        result = self._refused(pr_number=999, pr_head_sha=OTHER_SHA)
        self.assertTrue(any("does not equal local" in r for r in result["reasons"]))

    def test_unobservable_pr_head_refused(self):
        self._refused(pr_number=999, pr_head_sha=None)

    def test_competing_live_lock_on_same_issue_refused(self):
        result = self._refused(
            competing_live_locks=[
                {"issue_number": ISSUE, "branch_name": BRANCH, "pid": 777}
            ]
        )
        self.assertTrue(any("live lock" in r for r in result["reasons"]))

    def test_competing_live_lock_holding_the_branch_refused(self):
        self._refused(
            competing_live_locks=[
                {"issue_number": 111, "branch_name": BRANCH, "worktree_path": ""}
            ]
        )

    def test_competing_branch_claim_refused(self):
        result = self._refused(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"])
        self.assertTrue(any("issue marker" in r for r in result["reasons"]))

    def test_malformed_durable_lock_refused(self):
        lock = make_lock()
        lock["worktree_path"] = ""
        result = assess(lock)
        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)

    def test_lock_without_recorded_claimant_refused(self):
        lock = make_lock()
        lock["work_lease"]["claimant"] = {}
        result = assess(lock)
        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)


class NotARenewalCandidate(unittest.TestCase):
    """AC12 and scope: situations renewal must decline to judge at all."""

    def test_live_foreign_lease_is_never_a_candidate(self):
        lock = make_lock(expires_at=future_ts())
        result = assess(lock, identity="someone-else")
        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
        self.assertFalse(result["renewal_sanctioned"])

    def test_unexpired_lease_is_never_a_candidate(self):
        lock = make_lock(expires_at=future_ts())
        result = assess(lock)
        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)

    def test_dead_pid_under_unexpired_lease_stays_with_753(self):
        """The opposite trigger; #760 must not re-own it."""
        lock = make_lock(expires_at=future_ts(), pid=dead_pid())
        result = assess(lock)
        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)

    def test_absent_lock_is_not_a_candidate(self):
        result = assess({})
        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)

    def test_different_issue_is_not_a_candidate(self):
        lock = make_lock()
        lock["issue_number"] = ISSUE + 1
        result = assess(lock)
        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)

    def test_different_operation_type_is_not_a_candidate(self):
        lock = make_lock()
        lock["work_lease"]["operation_type"] = "review_pr_work"
        result = assess(lock)
        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)


class DaemonPidIsNotTaskLiveness(unittest.TestCase):
    """AC16: a live recorded PID is never, by itself, authorization."""

    def test_alive_pid_alone_does_not_authorize_renewal(self):
        # Every ownership fact except the live PID is wrong.
        result = assess(identity="someone-else", branch_name="fix/issue-1-nope")
        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
        self.assertTrue(result["evidence"]["prior_pid_alive"])

    def test_renewal_does_not_require_a_dead_pid(self):
        result = assess()
        self.assertTrue(result["evidence"]["prior_pid_alive"])
        self.assertTrue(result["renewal_sanctioned"])

    def test_dead_pid_does_not_block_an_otherwise_exact_owner(self):
        lock = make_lock(pid=dead_pid())
        result = assess(lock)
        self.assertTrue(result["renewal_sanctioned"])


class ConflictGateOrdering(unittest.TestCase):
    """AC2: the same-owner allowance is reachable on an expired lease.

    These cases need a worktree that genuinely exists on disk. The #601 reclaim
    affordance already permits takeover when the recorded worktree is missing,
    so a fictional path would satisfy the gate for the wrong reason and never
    exercise the ordering defect this issue is about.
    """

    @classmethod
    def setUpClass(cls):
        cls._tmp = tempfile.TemporaryDirectory()
        cls.worktree = cls._tmp.name

    @classmethod
    def tearDownClass(cls):
        cls._tmp.cleanup()

    def present_lock(self, **overrides):
        return make_lock(worktree_path=self.worktree, **overrides)

    def test_expired_same_owner_is_allowed_when_renewal_is_sanctioned(self):
        block = issue_lock_store.assess_same_issue_lease_conflict(
            self.present_lock(),
            issue_number=ISSUE,
            branch_name=BRANCH,
            worktree_path=self.worktree,
            renewal_sanctioned=True,
        )
        self.assertIsNone(block)

    def test_expired_same_owner_still_blocks_without_the_waiver(self):
        """Regression for the ordering defect: no waiver, no change in behavior.

        Live PID and a present worktree, so the #601 reclaim affordance refuses;
        before #760 this was the permanent dead end for an exact owner.
        """
        lock = self.present_lock()
        self.assertFalse(
            issue_lock_store.assess_expired_lock_reclaim(lock)["reclaim_allowed"]
        )
        block = issue_lock_store.assess_same_issue_lease_conflict(
            lock,
            issue_number=ISSUE,
            branch_name=BRANCH,
            worktree_path=self.worktree,
        )
        self.assertIsNotNone(block)
        self.assertIn("Recovery review is required", block)

    def test_waiver_does_not_unlock_a_different_owner(self):
        """AC11: the waiver is scoped by same_owner, not merely by its own flag."""
        block = issue_lock_store.assess_same_issue_lease_conflict(
            self.present_lock(),
            issue_number=ISSUE,
            branch_name=f"fix/issue-{ISSUE}-someone-else",
            worktree_path=self.worktree,
            renewal_sanctioned=True,
        )
        self.assertIsNotNone(block)
        self.assertIn("Recovery review is required", block)

    def test_live_lease_disposition_is_unchanged(self):
        """AC12: a live foreign lease still blocks, waiver or not."""
        block = issue_lock_store.assess_same_issue_lease_conflict(
            self.present_lock(expires_at=future_ts()),
            issue_number=ISSUE,
            branch_name=f"fix/issue-{ISSUE}-someone-else",
            worktree_path="/scratch/other",
            renewal_sanctioned=True,
        )
        self.assertIsNotNone(block)
        self.assertIn("already has an active", block)

    def test_dead_pid_reclaim_path_is_unchanged(self):
        """AC11: expired + dead PID still reclaims through the #601 affordance."""
        lock = self.present_lock(pid=dead_pid())
        reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
        self.assertTrue(reclaim["reclaim_allowed"])
        block = issue_lock_store.assess_same_issue_lease_conflict(
            lock,
            issue_number=ISSUE,
            branch_name=BRANCH,
            worktree_path=self.worktree,
        )
        self.assertIsNone(block)


class RenewalRecordAndDownstream(unittest.TestCase):
    """AC9/AC10: durable audit trail, and a renewed lock that actually works."""

    def test_record_captures_prior_and_replacement_state(self):
        assessment = assess()
        record = issue_lock_renewal.build_renewal_record(
            assessment,
            renewed_at="2026-01-01T00:00:00Z",
            new_expires_at="2026-01-01T04:00:00Z",
        )
        self.assertTrue(record["renewed"])
        self.assertEqual(record["prior_pid"], os.getpid())
        self.assertEqual(record["new_expires_at"], "2026-01-01T04:00:00Z")
        self.assertEqual(record["renewed_at"], "2026-01-01T00:00:00Z")
        self.assertEqual(record["identity"], IDENTITY)
        self.assertEqual(record["profile"], PROFILE)
        self.assertTrue(record["prior_expires_at"])
        self.assertTrue(record["proof"])

    def test_renewed_lock_satisfies_verify_lock_for_mutation(self):
        renewed = make_lock(expires_at=future_ts())
        renewed["session_pid"] = os.getpid()
        renewed["lease_renewal"] = {"renewed": True}
        verdict = issue_lock_store.verify_lock_for_mutation(
            renewed,
            issue_number=ISSUE,
            branch_name=BRANCH,
        )
        self.assertTrue(verdict["proven"])
        self.assertFalse(verdict["block"])

    def test_refusal_message_names_the_missing_evidence(self):
        assessment = assess(porcelain_status=" M gitea_mcp_server.py\n")
        message = issue_lock_renewal.format_renewal_refusal(assessment)
        self.assertIn("refused", message)
        self.assertIn("uncommitted", message)


class NoCallerControlledRenewalFlag(unittest.TestCase):
    """AC14: renewal eligibility is never declarable by a caller."""

    def test_lock_issue_tool_exposes_no_renewal_parameter(self):
        import gitea_mcp_server

        target = gitea_mcp_server.gitea_lock_issue
        target = getattr(target, "fn", getattr(target, "__wrapped__", target))
        params = set(inspect.signature(target).parameters)
        for forbidden in ("renewal_sanctioned", "renew", "allow_renewal", "is_owner"):
            self.assertNotIn(forbidden, params)

    def test_store_defaults_to_no_waiver(self):
        params = inspect.signature(
            issue_lock_store.assess_same_issue_lease_conflict
        ).parameters
        self.assertIs(params["renewal_sanctioned"].default, False)
        bind_params = inspect.signature(issue_lock_store.bind_session_lock).parameters
        self.assertIs(bind_params["renewal_sanctioned"].default, False)


class NoIssueNumberSpecialCasing(unittest.TestCase):
    """AC17: no repository issue or PR number is special-cased."""

    def test_module_contains_no_hardcoded_issue_special_cases(self):
        source = inspect.getsource(issue_lock_renewal)
        code = "\n".join(
            line for line in source.splitlines() if not line.strip().startswith("#")
        )
        for literal in ("757", "759", "760"):
            self.assertNotIn(f"== {literal}", code)
            self.assertNotIn(f"issue_number == {literal}", code)


if __name__ == "__main__":
    unittest.main()