sysadmin
d56baa635e
Move @mcp.tool() from internal _list_open_pulls helper onto gitea_lock_issue so author sessions can discover and call the issue lock required by gitea_create_pr. Add regression tests for MCP registration and create_pr lock flow.
127 lines
4.2 KiB
Python
127 lines
4.2 KiB
Python
"""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) |