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