Gitea-Tools/tests/test_issue_lock_recovery.py

"""Integration tests for non-destructive lock/PR recovery (#440)."""
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import patch

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

import issue_lock_adoption  # noqa: E402
import issue_lock_store as ils  # noqa: E402
import mcp_server  # noqa: E402
from mcp_server import gitea_create_pr, gitea_lock_issue  # noqa: E402
from tests.test_mcp_server import CREATE_PR_ENV, ISSUE_WRITE_ENV  # noqa: E402

FAKE_AUTH = "token fake"
BRANCH = "feat/issue-420-server-code-parity"


def _lock_record(**overrides):
    record = {
        "issue_number": 420,
        "branch_name": BRANCH,
        "remote": "prgs",
        "org": "Scaled-Tech-Consulting",
        "repo": mcp_server.REMOTES["prgs"]["repo"],
        "worktree_path": "/tmp/wt-420",
        "work_lease": {
            "operation_type": "author_issue_work",
            "expires_at": "2999-01-01T00:00:00Z",
        },
    }
    record.update(overrides)
    return record


def _clean_git_state():
    return {
        "current_branch": BRANCH,
        "porcelain_status": "",
        "base_equivalent": True,
        "inspected_git_root": os.getcwd(),
        "base_branch": "master",
    }


class TestIssueLockRecovery(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.TemporaryDirectory()
        self._lock_dir = self._tmpdir.name
        self._env = {
            **ISSUE_WRITE_ENV,
            "GITEA_ISSUE_LOCK_DIR": self._lock_dir,
        }
        self._create_pr_env = {
            **CREATE_PR_ENV,
            "GITEA_ISSUE_LOCK_DIR": self._lock_dir,
        }

    def tearDown(self):
        self._tmpdir.cleanup()

    def test_adoption_after_restart_without_session_pointer(self):
        with patch.dict(os.environ, self._env, clear=True):
            with mock.patch("os.getpid", return_value=9999):
                self.assertIsNone(ils.read_session_issue_lock())

            with mock.patch(
                "mcp_server.issue_lock_worktree.read_worktree_git_state",
                return_value=_clean_git_state(),
            ), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
                "mcp_server.get_auth_header", return_value=FAKE_AUTH
            ):
                mock_api.side_effect = [
                    [],
                    [{"name": BRANCH, "commit": {"id": "934688a"}}],
                ]
                res = gitea_lock_issue(
                    issue_number=420,
                    branch_name=BRANCH,
                    remote="prgs",
                    worktree_path=os.path.realpath(os.getcwd()),
                )
        self.assertTrue(res["success"])
        self.assertIn("adoption", res)
        self.assertEqual(res["adoption"]["branch_head_commit"], "934688a")

    @mock.patch(
        "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
        return_value=(True, []),
    )
    @mock.patch("mcp_server.api_request")
    @mock.patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_resolves_keyed_lock_after_restart(self, _auth, mock_api, _role):
        mock_api.return_value = {"number": 501, "html_url": "https://example/pr/501"}
        worktree = os.path.realpath(os.getcwd())
        with patch.dict(os.environ, self._create_pr_env, clear=True):
            path = ils.lock_file_path(
                remote="prgs",
                org=mcp_server.REMOTES["prgs"]["org"],
                repo=mcp_server.REMOTES["prgs"]["repo"],
                issue_number=420,
                lock_dir=self._lock_dir,
            )
            ils.save_lock_file(path, _lock_record(worktree_path=worktree))

            with mock.patch("os.getpid", return_value=4242):
                self.assertIsNone(ils.read_session_issue_lock())

            res = gitea_create_pr(
                title="feat: recovery Closes #420",
                head=BRANCH,
                base="master",
                remote="prgs",
                worktree_path=worktree,
            )
        self.assertEqual(res["number"], 501)

    def test_open_pr_blocks_adoption(self):
        with patch.dict(os.environ, self._env, clear=True):
            with mock.patch(
                "mcp_server.issue_lock_worktree.read_worktree_git_state",
                return_value=_clean_git_state(),
            ), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
                "mcp_server.get_auth_header", return_value=FAKE_AUTH
            ):
                mock_api.side_effect = [
                    [{"number": 99, "head": {"ref": BRANCH}, "title": "", "body": ""}],
                    [{"name": BRANCH, "commit": {"id": "934688a"}}],
                ]
                with self.assertRaises(ValueError) as ctx:
                    gitea_lock_issue(
                        issue_number=420,
                        branch_name=BRANCH,
                        remote="prgs",
                        worktree_path=os.path.realpath(os.getcwd()),
                    )
        self.assertIn("already tied to an open PR", str(ctx.exception))

    def test_structured_ownership_ignores_issue_4200_collision(self):
        result = issue_lock_adoption.assess_own_branch_adoption(
            issue_number=420,
            requested_branch=BRANCH,
            existing_branches=[{"name": "feat/issue-4200-unrelated"}],
        )
        self.assertEqual(result["outcome"], issue_lock_adoption.NO_MATCH)


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