Gitea-Tools/tests/test_issue_lock_recovery.py

"""End-to-end lock recovery scenarios for issue #440."""
import os
import sys
import tempfile
import unittest
from pathlib import Path
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

PRGS_REPO = mcp_server.REMOTES["prgs"]["repo"]

FAKE_AUTH = "Basic dGVzdDp0ZXN0"
BRANCH = "feat/issue-420-server-code-parity"
ISSUE_WRITE_ENV = {
    "GITEA_ALLOWED_OPERATIONS": (
        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
    ),
}
CREATE_PR_ENV = {
    "GITEA_PROFILE_NAME": "author-test",
    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.create,gitea.branch.push",
    "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.approve,gitea.pr.merge,gitea.pr.review",
}


def _clean_master_git_state_for_lock():
    return {
        "current_branch": "master",
        "porcelain_status": "",
        "base_equivalent": True,
        "inspected_git_root": "/tmp/repo",
        "base_branch": "master",
    }


def _lock_record(**overrides):
    import issue_lock_provenance

    work_lease = {
        "operation_type": "author_issue_work",
        "expires_at": "2999-01-01T00:00:00Z",
    }
    record = {
        "issue_number": 420,
        "branch_name": BRANCH,
        "remote": "prgs",
        "org": "Scaled-Tech-Consulting",
        "repo": PRGS_REPO,
        "worktree_path": os.path.realpath(os.getcwd()),
        "work_lease": work_lease,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease.get("claimant"),
        ),
    }
    record.update(overrides)
    return record


class TestIssueLockRecovery(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.TemporaryDirectory()
        self.addCleanup(self._tmpdir.cleanup)
        self.lock_dir = self._tmpdir.name

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

    def test_competing_actor_branch_blocks_adoption(self):
        result = issue_lock_adoption.assess_own_branch_adoption(
            issue_number=420,
            requested_branch=BRANCH,
            existing_branches=[{"name": "feat/issue-420-other-work"}],
        )
        self.assertTrue(result["block"])

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    def test_lock_recovery_after_restart_adopts_own_branch(self, _dup_fetcher, _auth, mock_api, _git):
        mock_api.return_value = [{"name": BRANCH, "commit": {"id": "934688a"}}]
        env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
        with patch.dict(os.environ, env, clear=True):
            with patch("os.getpid", return_value=9999):
                res = gitea_lock_issue(
                    issue_number=420,
                    branch_name=BRANCH,
                    remote="prgs",
                )
        self.assertTrue(res["success"])
        self.assertIn("adoption", res)
        found = ils.find_lock_for_branch(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo=PRGS_REPO,
            branch_name=BRANCH,
            lock_dir=self.lock_dir,
        )
        self.assertEqual(found["branch_name"], BRANCH)

    @patch("mcp_server.api_request")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_resolves_durable_lock_after_session_loss(self, _auth, _role, mock_api):
        mock_api.return_value = {"number": 421, "html_url": "https://example/pr/421"}
        worktree = os.path.realpath(os.getcwd())
        path = ils.lock_file_path(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo=PRGS_REPO,
            issue_number=420,
            lock_dir=self.lock_dir,
        )
        ils.save_lock_file(path, _lock_record(worktree_path=worktree))

        env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
        with patch.dict(os.environ, env, clear=True):
            with patch("os.getpid", return_value=8888):
                self.assertIsNone(ils.read_session_issue_lock(lock_dir=self.lock_dir))
                res = gitea_create_pr(
                    title=f"feat: server parity Closes #420",
                    head=BRANCH,
                    remote="prgs",
                    worktree_path=worktree,
                )
        self.assertEqual(res["number"], 421)

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([{
            "number": 99,
            "head": {"ref": BRANCH},
            "title": "WIP",
            "body": "",
        }], [], {"status": "not_claimed"}),
    )
    def test_open_pr_blocks_recovery_lock(self, _dup_fetcher, _auth, mock_api, _git):
        env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
        with patch.dict(os.environ, env, clear=True):
            with self.assertRaises(ValueError) as ctx:
                gitea_lock_issue(issue_number=420, branch_name=BRANCH, remote="prgs")
        self.assertIn("open PR #99 already covers issue", str(ctx.exception))


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