Gitea-Tools/tests/test_issue_760_mcp_renewal_path.py

"""MCP-level exact-owner lease renewal through ``gitea_lock_issue`` (#760).

The unit suite in ``test_issue_760_exact_owner_lease_renewal`` proves the
renewal *disposition*. It cannot prove the disposition survives the rest of the
tool, and it did not: the waiver was computed and then discarded before
``assess_issue_lock_worktree``, so every real renewal still failed on
base-equivalence. A branch being renewed always carries committed work, so it is
never base-equivalent by construction — exactly the argument #753 already makes
for recovery.

These tests drive the public tool end to end against a real git repository and a
real durable lock file, composing every gate in the production order.
"""

from __future__ import annotations

import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from mutation_profile_fixture import shared_mutation_env  # noqa: E402

import issue_lock_provenance  # noqa: E402
import issue_lock_store  # noqa: E402
import mcp_server  # noqa: E402

ISSUE = 9760
BRANCH = f"fix/issue-{ISSUE}-renewal-mcp"
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"


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


class _RenewalMcpBase(unittest.TestCase):
    """Real git repo + durable expired lock owned by a live PID.

    The recorded PID is ``os.getpid()`` — unambiguously alive. That is the whole
    point of #760: the PID belongs to the long-lived MCP daemon, so its liveness
    says nothing about whether the authoring task still holds the work.
    """

    def setUp(self):
        self.lock_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.lock_dir.cleanup)
        self.repo = tempfile.mkdtemp(prefix="issue760-mcp-")
        self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False))
        self._init_worktree()
        self.remotes = patch.dict(
            mcp_server.REMOTES,
            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
        )
        self.remotes.start()
        self.addCleanup(patch.stopall)
        mcp_server._IDENTITY_CACHE.clear()

    def _git(self, *args):
        return subprocess.run(
            ["git", "-C", self.repo, *args],
            capture_output=True,
            text=True,
            check=True,
        )

    def _init_worktree(self):
        self._git("init", "-q", "-b", "master")
        self._git("config", "user.email", "[email protected]")
        self._git("config", "user.name", "Test")
        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
            fh.write("seed\n")
        self._git("add", "seed.txt")
        self._git("commit", "-q", "-m", "seed")
        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
        # The branch carries committed work, so it is NOT base-equivalent.
        self._git("checkout", "-q", "-b", BRANCH)
        with open(os.path.join(self.repo, "work.txt"), "w") as fh:
            fh.write("author work\n")
        self._git("add", "work.txt")
        self._git("commit", "-q", "-m", "author work")
        self.head_sha = self._git("rev-parse", "HEAD").stdout.strip()
        self.worktree = os.path.realpath(self.repo)

    def write_expired_lock(self, **overrides):
        path = issue_lock_store.lock_file_path(
            remote="prgs",
            org=ORG,
            repo=REPO,
            issue_number=ISSUE,
            lock_dir=self.lock_dir.name,
        )
        claimant = {"username": IDENTITY, "profile": PROFILE}
        pid = overrides.pop("session_pid", os.getpid())
        overrides.pop("pid", None)
        lease_overrides = overrides.pop("work_lease", {})
        data = {
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "remote": "prgs",
            "org": ORG,
            "repo": REPO,
            "worktree_path": self.worktree,
            "session_pid": pid,
            "pid": pid,
            "lock_generation": 3,
            "work_lease": {
                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
                "issue_number": ISSUE,
                "pr_number": None,
                "branch": BRANCH,
                "worktree_path": self.worktree,
                "claimant": claimant,
                "created_at": _past_ts(5),
                "last_heartbeat_at": _past_ts(5),
                "expires_at": _past_ts(),          # already expired
            },
            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
                tool="gitea_lock_issue",
                claimant=claimant,
            ),
        }
        data["work_lease"].update(lease_overrides)
        data.update(overrides)
        data["session_pid"] = pid
        data["pid"] = pid
        data["lock_file_path"] = path
        issue_lock_store.save_lock_file(path, data)
        return path

    def _tool_env(self):
        env = shared_mutation_env(
            PROFILE,
            include_example_repo=True,
            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
        )
        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
        return env

    def _git_state(self, *, porcelain="", branch=BRANCH, head=None):
        return {
            "current_branch": branch,
            "porcelain_status": porcelain,
            # The decisive fact: a branch carrying work is never base-equivalent.
            "base_equivalent": False,
            "head_sha": head or self.head_sha,
            "inspected_git_root": self.worktree,
            "base_branch": "master",
        }

    def run_lock_issue(
        self,
        *,
        branch_entries=None,
        open_prs=None,
        git_state=None,
        identity=IDENTITY,
        profile=PROFILE,
    ):
        """Drive the public tool for the published exact-owner renewal shape."""
        if branch_entries is None:
            branch_entries = [{"name": BRANCH, "commit": {"id": self.head_sha}}]
        if open_prs is None:
            open_prs = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]
        if git_state is None:
            git_state = self._git_state()
        env = self._tool_env()
        with patch(
            "mcp_server.api_get_all", return_value=list(branch_entries)
        ), patch(
            "mcp_server._list_open_pulls", return_value=list(open_prs)
        ), patch(
            "mcp_server.get_auth_header", return_value="token x"
        ), patch(
            "mcp_server._work_lease_claimant",
            return_value={"username": identity, "profile": profile},
        ), patch(
            "mcp_server.issue_lock_worktree.read_worktree_git_state",
            return_value=git_state,
        ), patch(
            "mcp_server.issue_duplicate_context_fetcher",
            side_effect=lambda h, o, r, auth, issue_number: (
                list(open_prs),
                [b.get("name") for b in branch_entries if isinstance(b, dict)],
                {"status": "not_claimed"},
            ),
        ), patch.dict(os.environ, env, clear=True):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_lock_issue(
                issue_number=ISSUE,
                branch_name=BRANCH,
                remote="prgs",
                worktree_path=self.worktree,
            )


class TestRenewalReachableThroughTool(_RenewalMcpBase):
    """F1: the sanctioned renewal must survive every downstream gate."""

    def test_expired_lease_live_pid_exact_owner_renews_through_the_tool(self):
        prior = issue_lock_store.read_lock_file(self.write_expired_lock())
        self.assertTrue(issue_lock_store.is_lease_expired(prior))
        self.assertTrue(issue_lock_store.is_process_alive(prior["session_pid"]))

        result = self.run_lock_issue()

        self.assertTrue(result["success"], result)
        self.assertEqual(result["issue_number"], ISSUE)
        self.assertEqual(result["branch_name"], BRANCH)
        # The renewal is reported natively, so no lock-file inspection is needed.
        self.assertIn("lease_renewal", result)
        self.assertTrue(result["lease_renewal"]["renewed"])
        self.assertIn("Renewed the expired", result["message"])

    def test_renewed_lock_records_prior_and_replacement_evidence(self):
        prior = issue_lock_store.read_lock_file(self.write_expired_lock())
        prior_expiry = prior["work_lease"]["expires_at"]
        prior_generation = issue_lock_store.lock_generation(prior)

        result = self.run_lock_issue()
        written = issue_lock_store.read_lock_file(result["lock_file_path"])

        renewal = written["lease_renewal"]
        self.assertTrue(renewal["renewed"])
        self.assertEqual(renewal["prior_pid"], prior["session_pid"])
        self.assertTrue(renewal["prior_pid_alive"])
        self.assertEqual(renewal["prior_expires_at"], prior_expiry)
        self.assertEqual(renewal["identity"], IDENTITY)
        self.assertEqual(renewal["profile"], PROFILE)
        self.assertEqual(renewal["head_sha"], self.head_sha)
        self.assertTrue(renewal["proof"])
        # New expiry is a fresh absolute stamp, later than the one it replaced.
        self.assertEqual(renewal["new_expires_at"], written["work_lease"]["expires_at"])
        self.assertGreater(renewal["new_expires_at"], prior_expiry)
        # Compare-and-swap advanced the generation exactly once.
        self.assertEqual(
            issue_lock_store.lock_generation(written), prior_generation + 1
        )

    def test_renewed_lock_is_live_and_satisfies_mutation_ownership(self):
        self.write_expired_lock()
        result = self.run_lock_issue()
        written = issue_lock_store.read_lock_file(result["lock_file_path"])

        self.assertTrue(issue_lock_store.assess_lock_freshness(written)["live"])
        verdict = issue_lock_store.verify_lock_for_mutation(
            written,
            issue_number=ISSUE,
            branch_name=BRANCH,
            worktree_path=self.worktree,
        )
        self.assertTrue(verdict["proven"], verdict)
        self.assertFalse(verdict["block"])

    def test_recovery_record_is_not_written_for_a_live_owner_renewal(self):
        """#753 recovery must not be claimed when the recorded PID is alive."""
        self.write_expired_lock()
        result = self.run_lock_issue()
        written = issue_lock_store.read_lock_file(result["lock_file_path"])
        self.assertNotIn("dead_session_recovery", written)


class TestRenewalWaiverIsNarrow(_RenewalMcpBase):
    """The waiver relaxes base-equivalence and nothing else."""

    def test_dirty_worktree_still_blocks_a_would_be_renewal(self):
        """Cleanliness is never waived; the renewal assessor refuses first.

        A dirty worktree makes the renewal refuse, so no waiver is issued and
        the lease-conflict gate fails closed ahead of the worktree gate. The
        refusal names the uncommitted files, so the owner still learns why.
        """
        self.write_expired_lock()
        with self.assertRaises(Exception) as ctx:
            self.run_lock_issue(
                git_state=self._git_state(porcelain=" M gitea_mcp_server.py\n")
            )
        message = str(ctx.exception)
        self.assertIn("Recovery review is required before takeover", message)
        self.assertIn("worktree has uncommitted tracked changes", message)
        self.assertIn("gitea_mcp_server.py", message)

    def test_foreign_claimant_cannot_use_the_waiver(self):
        """A near-match owner gets no renewal and no base-equivalence waiver."""
        self.write_expired_lock()
        with self.assertRaises(Exception) as ctx:
            self.run_lock_issue(identity="someone-else")
        message = str(ctx.exception)
        self.assertIn("Recovery review is required before takeover", message)
        # The refusal names the missing ownership evidence (#760 diagnostics).
        self.assertIn("does not match active identity", message)

    def test_foreign_profile_cannot_use_the_waiver(self):
        self.write_expired_lock()
        with self.assertRaises(Exception) as ctx:
            self.run_lock_issue(profile="other-author")
        self.assertIn(
            "Recovery review is required before takeover", str(ctx.exception)
        )

    def test_unpublished_branch_cannot_use_the_waiver(self):
        """No remote head to agree with, so exact-owner renewal is refused."""
        self.write_expired_lock()
        with self.assertRaises(Exception) as ctx:
            self.run_lock_issue(branch_entries=[], open_prs=[])
        self.assertIn(
            "Recovery review is required before takeover", str(ctx.exception)
        )

    def test_pr_head_mismatch_cannot_use_the_waiver(self):
        self.write_expired_lock()
        other = "9" * 40
        with self.assertRaises(Exception) as ctx:
            self.run_lock_issue(
                open_prs=[{"number": 4242, "head": {"ref": BRANCH, "sha": other}}]
            )
        self.assertIn(
            "Recovery review is required before takeover", str(ctx.exception)
        )

    def test_non_base_equivalent_branch_still_blocks_without_any_waiver(self):
        """No durable lock at all: the ordinary base-equivalence rule applies."""
        with self.assertRaises(Exception) as ctx:
            self.run_lock_issue()
        self.assertIn("must be base-equivalent", str(ctx.exception))


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