Merge pull request 'fix: register gitea_lock_issue as MCP tool (Closes #521)' (#526) from feat/issue-521-lock-issue-registration into master

gitea_mcp_server.py

@@ -1446,7 +1446,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-@mcp.tool()
 def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
     """Fetch all OPEN pull requests for a repo (used for stacked-base proof, #484)."""
     try:
@@ -1457,6 +1456,7 @@ def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
         )
 
 
+@mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
     branch_name: str,

tests/test_lock_issue_mcp_registration.py

@@ -0,0 +1,127 @@
+"""Regression tests for gitea_lock_issue MCP tool registration (#521)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import issue_lock_store
+import mcp_server
+from mcp_server import gitea_create_pr, gitea_lock_issue
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+
+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.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+    "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": "/scratch/wt",
+        "base_branch": "origin/master",
+    }
+
+
+def _registered_tool_names() -> set[str]:
+    manager = mcp_server.mcp._tool_manager
+    tools = getattr(manager, "_tools", None) or {}
+    return set(tools.keys())
+
+
+class TestLockIssueMcpRegistration(unittest.TestCase):
+    def test_gitea_lock_issue_registered_as_public_mcp_tool(self):
+        names = _registered_tool_names()
+        self.assertIn("gitea_lock_issue", names)
+
+    def test_internal_list_open_pulls_not_exposed_as_mcp_tool(self):
+        names = _registered_tool_names()
+        self.assertNotIn("_list_open_pulls", names)
+
+
+class TestCreatePrLockRegistrationFlow(unittest.TestCase):
+    def setUp(self):
+        self._lock_dir = tempfile.TemporaryDirectory()
+        self._env_patcher = patch.dict(
+            os.environ,
+            {
+                **CREATE_PR_ENV,
+                "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
+            },
+            clear=True,
+        )
+        self._env_patcher.start()
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self._dup_fetcher_patcher.start()
+
+    def tearDown(self):
+        self._dup_fetcher_patcher.stop()
+        self._env_patcher.stop()
+        self._lock_dir.cleanup()
+
+    @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_still_fails_closed_without_issue_lock(self, _auth, _role):
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_create_pr(
+                title="feat: X Closes #521",
+                head="feat/issue-521-lock-issue-registration",
+                remote="prgs",
+            )
+        self.assertIn("Issue lock is missing", str(ctx.exception))
+
+    @patch("mcp_server.api_request")
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all", return_value=[])
+    @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_proceeds_after_valid_issue_lock(
+        self, _auth, _role, _api, _git_state, mock_api_request
+    ):
+        worktree = os.path.realpath(os.getcwd())
+        mock_api_request.return_value = {"number": 521, "html_url": "https://example/pr/521"}
+        lock_res = gitea_lock_issue(
+            issue_number=521,
+            branch_name="feat/issue-521-lock-issue-registration",
+            remote="prgs",
+            worktree_path=worktree,
+        )
+        self.assertTrue(lock_res["success"])
+        lock_path = lock_res["lock_file_path"]
+        self.assertTrue(os.path.exists(lock_path))
+        lock = issue_lock_store.read_lock_file(lock_path)
+        self.assertEqual(lock["issue_number"], 521)
+
+        res = gitea_create_pr(
+            title="fix: restore lock tool registration Closes #521",
+            head="feat/issue-521-lock-issue-registration",
+            remote="prgs",
+            worktree_path=worktree,
+        )
+        self.assertEqual(res["number"], 521)