Gitea-Tools/tests/test_issue_945_owning_pr_renewal_continuation.py

import sys as _sys
from pathlib import Path as _Path
_sys.path.insert(0, str(_Path(__file__).resolve().parent))
from mutation_profile_fixture import shared_mutation_env  # noqa: F401,E402
"""Exact-owner renewal keeps its owning-PR waiver past lock_issue (#945).

#755 taught the duplicate-work gate that a sanctioned *dead-session recovery*
owns its open PR, and #768 taught the later gates to rebuild that proof from the
durable lock. #760 added the exact-owner *renewal* disposition and granted it
the same waiver inside ``gitea_lock_issue`` — but never added the matching
rebuild. So an ordinary renewal held the waiver only for the duration of the
lock call: ``_enforce_locked_issue_duplicate_recheck`` asked
``recovered_owning_pr_from_lock``, which reads only ``dead_session_recovery``,
and the very next commit was refused ``duplicate_commit_prevented`` with
``owning_pr_recovery_exempted: false`` on the PR the renewal had just proved.

``TestPreFixReproduction`` pins that defect directly: the recovery-only rebuild
still returns ``None`` for a renewal lock, which is exactly why the gates lost
the waiver. Everything else proves the renewal half now survives, that recovery
is unchanged, and that no path grants an exemption on weaker evidence.

Every fixture here is an in-memory mapping. Nothing writes a branch, worktree,
lock file, lease, comment, or PR (#945 AC18).
"""
import copy
import sys
import unittest
from pathlib import Path

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

import gitea_mcp_server  # noqa: E402
import issue_lock_recovery  # noqa: E402
import issue_lock_renewal  # noqa: E402
from issue_work_duplicate_gate import (  # noqa: E402
    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
    PHASE_COMMIT,
    PHASE_CREATE_PR,
    PHASE_LOCK,
    PHASE_PUSH,
    assess_work_issue_duplicate_gate,
)

ISSUE = 4945
OWNING_PR = 4946
OTHER_PR = 4947
BRANCH = f"fix/issue-{ISSUE}-owning-pr-renewal"
OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
HEAD = "a" * 40
OTHER_HEAD = "b" * 40
IDENTITY = "example-user"
PROFILE = "test-author-prgs"


def renewal_record(**overrides):
    """The ``lease_renewal`` block ``build_renewal_record`` writes on success."""
    record = {
        "renewed": True,
        "renewed_at": "2026-01-01T00:00:00Z",
        "prior_pid": 4242,
        "prior_pid_alive": True,
        "prior_expires_at": "2026-01-01T00:00:00Z",
        "replacement_pid": 4243,
        "new_expires_at": "2026-01-01T00:10:00Z",
        "identity": IDENTITY,
        "profile": PROFILE,
        "branch_name": BRANCH,
        "worktree_path": f"branches/issue-{ISSUE}-owning-pr-renewal",
        "head_sha": HEAD,
        "remote_head_sha": HEAD,
        "pr_head_sha": HEAD,
        "pr_number": OWNING_PR,
        "reason": "expired lease renewed by its exact recorded owner",
        "proof": [],
    }
    record.update(overrides)
    return record


def renewal_lock(record=None, *, issue_number=ISSUE, claimant=True, **lock_overrides):
    lock = {
        "issue_number": issue_number,
        "branch_name": BRANCH,
        "lease_renewal": renewal_record() if record is None else record,
    }
    if claimant:
        lock["claimant"] = {"username": IDENTITY, "profile": PROFILE}
    lock.update(lock_overrides)
    return lock


def recovery_lock(pr_number=OWNING_PR, head=HEAD):
    """A lock carrying sanctioned dead-session recovery evidence (#755/#768)."""
    return {
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "claimant": {"username": IDENTITY, "profile": PROFILE},
        "dead_session_recovery": {
            "recovered": True,
            "branch_name": BRANCH,
            "pr_number": pr_number,
            "pr_head": head,
            "recorded_head": head,
            "accepted_head": head,
            "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
        },
    }


def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
    return {
        "number": number,
        "title": f"fix: something (Closes #{issue})",
        "body": f"Closes #{issue}.",
        "head": {"ref": ref, "sha": sha},
    }


def gate(phase, *, token, open_prs=None, branch_names=None, locked_branch=BRANCH):
    return assess_work_issue_duplicate_gate(
        ISSUE,
        open_prs=[owning_pr()] if open_prs is None else open_prs,
        branch_names=branch_names or [],
        claim_entry={},
        locked_branch=locked_branch,
        phase=phase,
        recovered_owning_pr=token,
    )


# ───────────────────── the defect this issue exists to fix ─────────────────────


class TestPreFixReproduction(unittest.TestCase):
    """The exact wiring gap: renewal evidence was invisible to later gates."""

    def test_recovery_only_rebuild_cannot_see_a_renewal_lock(self):
        # This is the pre-fix behaviour of every enforcement path. It is correct
        # for the recovery rebuild to ignore a renewal block -- the defect was
        # that nothing else looked at it.
        self.assertIsNone(
            issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
        )

    def test_renewal_lock_produced_no_exemption_before_the_fix(self):
        # Feeding the gate what the pre-fix code fed it (recovery rebuild only)
        # reproduces the reported refusal at the commit phase.
        token = issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
        result = gate(PHASE_COMMIT, token=token)
        self.assertTrue(result["block"])
        self.assertEqual(result["outcome"], "duplicate_commit_prevented")
        self.assertFalse(result["owning_pr_recovery_exempted"])
        self.assertEqual(result["owning_pr_recovery_notes"], [])

    def test_shared_resolver_now_sees_it(self):
        self.assertIsNotNone(
            gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
        )


# ───────────────────────── rebuild: the granted case ─────────────────────────


class TestRenewalRebuildGranted(unittest.TestCase):
    def test_sanctioned_renewal_rebuilds_owning_pr_evidence(self):
        token = issue_lock_renewal.owning_pr_renewal_from_lock(renewal_lock())
        self.assertEqual(
            token,
            {
                "issue_number": ISSUE,
                "pr_number": OWNING_PR,
                "branch_name": BRANCH,
                "head_sha": HEAD,
                "recorded_head": HEAD,
                "accepted_head": HEAD,
                "head_relation": "equal",
            },
        )

    def test_branch_falls_back_to_the_lock_branch(self):
        lock = renewal_lock(renewal_record(branch_name=""))
        token = issue_lock_renewal.owning_pr_renewal_from_lock(lock)
        self.assertEqual(token["branch_name"], BRANCH)

    def test_claimant_may_live_under_work_lease(self):
        lock = renewal_lock(claimant=False)
        lock["work_lease"] = {"claimant": {"username": IDENTITY, "profile": PROFILE}}
        self.assertIsNotNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))

    def test_rebuild_does_not_mutate_the_lock(self):
        lock = renewal_lock()
        before = copy.deepcopy(lock)
        issue_lock_renewal.owning_pr_renewal_from_lock(lock)
        self.assertEqual(lock, before)


# ───────────────────────── rebuild: fails closed ─────────────────────────


class TestRenewalRebuildFailsClosed(unittest.TestCase):
    def assertNoEvidence(self, lock):
        self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))

    def test_no_lock_at_all(self):
        self.assertNoEvidence(None)
        self.assertNoEvidence({})
        self.assertNoEvidence("not-a-mapping")

    def test_lock_without_renewal_block(self):
        # A fresh claim, or a lock whose renewal block was replaced.
        self.assertNoEvidence({"issue_number": ISSUE, "branch_name": BRANCH})

    def test_renewal_not_granted(self):
        self.assertNoEvidence(renewal_lock(renewal_record(renewed=False)))

    def test_renewal_flag_missing(self):
        record = renewal_record()
        del record["renewed"]
        self.assertNoEvidence(renewal_lock(record))

    def test_renewal_block_malformed(self):
        self.assertNoEvidence(renewal_lock("not-a-mapping"))

    def test_local_head_diverged_from_pr_head(self):
        self.assertNoEvidence(renewal_lock(renewal_record(head_sha=OTHER_HEAD)))

    def test_remote_head_diverged_from_pr_head(self):
        # Force-push or unrelated remote movement.
        self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha=OTHER_HEAD)))

    def test_local_head_missing(self):
        self.assertNoEvidence(renewal_lock(renewal_record(head_sha="")))

    def test_remote_head_missing(self):
        self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha="")))

    def test_pr_head_missing(self):
        self.assertNoEvidence(renewal_lock(renewal_record(pr_head_sha="")))

    def test_pr_number_missing(self):
        self.assertNoEvidence(renewal_lock(renewal_record(pr_number=None)))

    def test_pr_number_malformed(self):
        self.assertNoEvidence(renewal_lock(renewal_record(pr_number="not-a-number")))

    def test_issue_number_missing_from_lock(self):
        self.assertNoEvidence(renewal_lock(issue_number=None))

    def test_branch_unknown_everywhere(self):
        lock = renewal_lock(renewal_record(branch_name=""))
        lock["branch_name"] = ""
        self.assertNoEvidence(lock)

    def test_identity_mismatch(self):
        self.assertNoEvidence(renewal_lock(renewal_record(identity="someone-else")))

    def test_profile_mismatch(self):
        self.assertNoEvidence(renewal_lock(renewal_record(profile="other-profile")))

    def test_identity_missing(self):
        self.assertNoEvidence(renewal_lock(renewal_record(identity="")))

    def test_profile_missing(self):
        self.assertNoEvidence(renewal_lock(renewal_record(profile="")))

    def test_claimant_absent(self):
        self.assertNoEvidence(renewal_lock(claimant=False))

    def test_evidence_from_a_different_session_is_not_reusable(self):
        # A renewal block left by another workflow session names another
        # claimant, so the lock it is found on cannot inherit its authority.
        lock = renewal_lock()
        lock["claimant"] = {"username": "other-session-user", "profile": PROFILE}
        self.assertNoEvidence(lock)


# ───────────────────────── the shared resolver ─────────────────────────


class TestSharedResolver(unittest.TestCase):
    def test_recovery_lock_resolves_to_recovery_evidence(self):
        token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
        self.assertEqual(token["pr_number"], OWNING_PR)

    def test_renewal_lock_resolves_to_renewal_evidence(self):
        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
        self.assertEqual(token["pr_number"], OWNING_PR)

    def test_recovery_takes_precedence_over_renewal(self):
        # Same precedence gitea_lock_issue applies when granting the waiver, so
        # the answer cannot differ between the granting and enforcing paths.
        lock = recovery_lock(pr_number=OTHER_PR, head=OTHER_HEAD)
        lock["lease_renewal"] = renewal_record()
        token = gitea_mcp_server._owning_pr_continuation_from_lock(lock)
        self.assertEqual(token["pr_number"], OTHER_PR)

    def test_no_evidence_resolves_to_none(self):
        self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock(None))
        self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock({}))
        self.assertIsNone(
            gitea_mcp_server._owning_pr_continuation_from_lock(
                {"issue_number": ISSUE, "branch_name": BRANCH}
            )
        )


# ────────────── every enforcement path uses the same decision ──────────────


class TestEnforcementPathsShareOneDecision(unittest.TestCase):
    """AC: commit, push and create-PR gates consume one authoritative token."""

    def setUp(self):
        self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())

    def test_commit_phase_permits_continuation(self):
        result = gate(PHASE_COMMIT, token=self.token)
        self.assertFalse(result["block"])
        self.assertTrue(result["owning_pr_recovery_exempted"])
        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)

    def test_create_pr_phase_permits_continuation(self):
        result = gate(PHASE_CREATE_PR, token=self.token)
        self.assertFalse(result["block"])
        self.assertTrue(result["owning_pr_recovery_exempted"])

    def test_push_phase_permits_continuation(self):
        result = gate(PHASE_PUSH, token=self.token)
        self.assertFalse(result["block"])
        self.assertTrue(result["owning_pr_recovery_exempted"])

    def test_lock_phase_permits_continuation(self):
        result = gate(PHASE_LOCK, token=self.token)
        self.assertFalse(result["block"])

    def test_all_phases_agree(self):
        outcomes = {
            phase: gate(phase, token=self.token)["block"]
            for phase in (PHASE_LOCK, PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR)
        }
        self.assertEqual(set(outcomes.values()), {False}, outcomes)

    def test_dead_session_recovery_still_permits_continuation(self):
        token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
        for phase in (PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR):
            with self.subTest(phase=phase):
                result = gate(phase, token=token)
                self.assertFalse(result["block"])
                self.assertTrue(result["owning_pr_recovery_exempted"])


# ───────────────── the exemption cannot be widened ─────────────────


class TestExemptionCannotBeWidened(unittest.TestCase):
    def setUp(self):
        self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())

    def test_an_open_pr_alone_grants_nothing(self):
        result = gate(PHASE_COMMIT, token=None)
        self.assertTrue(result["block"])
        self.assertFalse(result["owning_pr_recovery_exempted"])

    def test_a_second_pr_is_refused(self):
        result = gate(
            PHASE_CREATE_PR,
            token=self.token,
            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
        )
        self.assertTrue(result["block"])
        self.assertFalse(result["owning_pr_recovery_exempted"])

    def test_a_different_pr_is_refused(self):
        result = gate(
            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(number=OTHER_PR)]
        )
        self.assertTrue(result["block"])

    def test_a_different_branch_is_refused(self):
        result = gate(
            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(ref=OTHER_BRANCH)]
        )
        self.assertTrue(result["block"])

    def test_locked_branch_mismatch_is_refused(self):
        result = gate(PHASE_COMMIT, token=self.token, locked_branch=OTHER_BRANCH)
        self.assertTrue(result["block"])

    def test_live_pr_head_divergence_is_refused(self):
        # Force-push or unrelated remote movement after renewal.
        result = gate(
            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(sha=OTHER_HEAD)]
        )
        self.assertTrue(result["block"])

    def test_evidence_for_another_issue_is_refused(self):
        foreign = gitea_mcp_server._owning_pr_continuation_from_lock(
            renewal_lock(issue_number=ISSUE + 1)
        )
        result = gate(PHASE_COMMIT, token=foreign)
        self.assertTrue(result["block"])

    def test_sequential_tasks_do_not_inherit_continuation(self):
        # One daemon serves many tasks. A renewal proved for issue N must not
        # authorize continuation for the next task's issue.
        prior_task = gitea_mcp_server._owning_pr_continuation_from_lock(
            renewal_lock(issue_number=ISSUE + 7)
        )
        self.assertIsNotNone(prior_task)
        self.assertTrue(gate(PHASE_COMMIT, token=prior_task)["block"])


# ───────────────── ordinary duplicate prevention is intact ─────────────────


class TestDuplicatePreventionRetained(unittest.TestCase):
    def test_competing_branch_still_blocks(self):
        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
        result = gate(
            PHASE_COMMIT,
            token=token,
            open_prs=[],
            branch_names=[BRANCH, OTHER_BRANCH],
        )
        self.assertTrue(result["block"])

    def test_unrelated_work_without_a_lock_still_blocks(self):
        token = gitea_mcp_server._owning_pr_continuation_from_lock(None)
        self.assertIsNone(token)
        self.assertTrue(gate(PHASE_COMMIT, token=token)["block"])


# ───────────────── refusals stay structured and auditable ─────────────────


class TestRefusalShapePreserved(unittest.TestCase):
    def test_blocked_result_keeps_its_audit_fields(self):
        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
        result = gate(
            PHASE_COMMIT, token=token, open_prs=[owning_pr(number=OTHER_PR)]
        )
        for field in (
            "block",
            "outcome",
            "reasons",
            "owning_pr_recovery_exempted",
            "owning_pr_recovery_notes",
        ):
            with self.subTest(field=field):
                self.assertIn(field, result)
        self.assertTrue(result["reasons"])
        # A rejected token explains which element of ownership disagreed.
        self.assertTrue(result["owning_pr_recovery_notes"])

    def test_granted_result_records_why(self):
        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
        result = gate(PHASE_COMMIT, token=token)
        self.assertTrue(result["owning_pr_recovery_notes"])
        self.assertIn(
            f"#{OWNING_PR}", " ".join(result["owning_pr_recovery_notes"])
        )


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