"""Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69).""" import json import os import sys import tempfile import unittest from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import mcp_server import task_capability_map CONFIG = { "version": 2, "contexts": { "ctx": { "enabled": True, "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, } }, "profiles": { "comment-only": { "enabled": True, "context": "ctx", "role": "author", "username": "commenter", "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": ["gitea.read", "gitea.issue.comment"], "forbidden_operations": [ "gitea.issue.create", "gitea.issue.close"], "execution_profile": "comment-only", }, "full-author": { "enabled": True, "context": "ctx", "role": "author", "username": "author-user", "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": [ "gitea.read", "gitea.issue.create", "gitea.issue.close", "gitea.issue.comment"], "forbidden_operations": [], "execution_profile": "full-author", }, }, "rules": {"allow_runtime_switching": False}, } ISSUE_WRITE_TASKS = ( "create_issue", "close_issue", "comment_issue", "mark_issue", "set_issue_labels", ) TOOL_CALLS = { "create_issue": lambda: mcp_server.gitea_create_issue( title="New issue", remote="prgs"), "close_issue": lambda: mcp_server.gitea_close_issue( issue_number=9, remote="prgs"), "comment_issue": lambda: mcp_server.gitea_create_issue_comment( issue_number=9, body="hello", remote="prgs"), "mark_issue": lambda: mcp_server.gitea_mark_issue( issue_number=9, action="start", remote="prgs"), "set_issue_labels": lambda: mcp_server.gitea_set_issue_labels( issue_number=9, labels=["bug"], remote="prgs"), } class IssueWriteGateBase(unittest.TestCase): def setUp(self): self._remotes = patch.dict(mcp_server.REMOTES, { "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}, }) self._remotes.start() mcp_server._IDENTITY_CACHE.clear() self._dir = tempfile.TemporaryDirectory() self.config_path = os.path.join(self._dir.name, "profiles.json") with open(self.config_path, "w", encoding="utf-8") as fh: fh.write(json.dumps(CONFIG)) def tearDown(self): self._remotes.stop() mcp_server._IDENTITY_CACHE.clear() self._dir.cleanup() def _env(self, profile: str) -> dict: return { "GITEA_MCP_CONFIG": self.config_path, "GITEA_MCP_PROFILE": profile, "GITEA_TOKEN_AUTHOR": "author-pass", } class TestResolverToolAlignment(unittest.TestCase): def test_issue_mutation_tools_match_resolver_permissions(self): for tool_name, task_key in ( task_capability_map.ISSUE_MUTATION_TOOL_TASKS.items() ): self.assertEqual( task_capability_map.tool_required_permission(tool_name), task_capability_map.required_permission(task_key), msg=f"{tool_name} gate drift from resolver task {task_key}", ) class TestDeniedIssueWritesFailClosed(IssueWriteGateBase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolver_denied_tasks_block_raw_tools( self, _auth, mock_api, _get_all, _role ): mock_api.return_value = {"number": 1} with patch.dict(os.environ, self._env("comment-only"), clear=True): for task in ISSUE_WRITE_TASKS: with self.subTest(task=task): resolve = mcp_server.gitea_resolve_task_capability( task=task, remote="prgs") if resolve["allowed_in_current_session"]: continue result = TOOL_CALLS[task]() self.assertFalse(result.get("success", True), task) self.assertFalse(result.get("performed", True), task) self.assertTrue(result.get("reasons"), task) self.assertIn("permission_report", result, task) self.assertEqual( result["permission_report"]["missing_permission"], resolve["required_operation_permission"], ) class TestAllowedIssueWritesProceed(IssueWriteGateBase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolver_allowed_create_issue_proceeds( self, _auth, mock_api, _get_all, _role ): mock_api.return_value = {"number": 42, "html_url": "https://x/42"} with patch.dict(os.environ, self._env("full-author"), clear=True): resolve = mcp_server.gitea_resolve_task_capability( task="create_issue", remote="prgs") self.assertTrue(resolve["allowed_in_current_session"]) result = mcp_server.gitea_create_issue(title="Allowed", remote="prgs") self.assertEqual(result.get("number"), 42) post_calls = [ c for c in mock_api.call_args_list if c.args[0] == "POST" ] self.assertTrue(post_calls) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolver_allowed_close_issue_proceeds(self, _auth, mock_api): mock_api.return_value = {"state": "closed"} with patch.dict(os.environ, self._env("full-author"), clear=True): resolve = mcp_server.gitea_resolve_task_capability( task="close_issue", remote="prgs") self.assertTrue(resolve["allowed_in_current_session"]) result = mcp_server.gitea_close_issue(issue_number=9, remote="prgs") self.assertTrue(result.get("success")) @patch("mcp_server.api_get_all", return_value=[{"id": 10, "name": "status:in-progress"}]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api, _labels): def api_side_effect(method, url, auth, payload=None): if method == "GET" and "/issues/9" in url: return {"labels": []} if method == "PUT" and "/issues/9/labels" in url: return [{"name": "status:in-progress"}] if method == "POST" and "/issues/9/comments" in url: return {"id": 501} if method == "GET" and url.endswith("/user"): return {"login": "author-user"} return {} mock_api.side_effect = api_side_effect with patch.dict(os.environ, self._env("full-author"), clear=True): resolve = mcp_server.gitea_resolve_task_capability( task="mark_issue", remote="prgs") self.assertTrue(resolve["allowed_in_current_session"]) result = mcp_server.gitea_mark_issue( issue_number=9, action="start", remote="prgs") self.assertTrue(result.get("success")) class TestCreateIssueMissingPermissionReport(IssueWriteGateBase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_create_issue_without_create_permission_blocked( self, _auth, mock_api, _get_all, _role ): with patch.dict(os.environ, self._env("comment-only"), clear=True): result = mcp_server.gitea_create_issue(title="Blocked", remote="prgs") self.assertFalse(result["success"]) self.assertFalse(result["performed"]) self.assertIsNone(result["number"]) self.assertEqual( result["permission_report"]["missing_permission"], "gitea.issue.create", ) mock_api.assert_not_called() if __name__ == "__main__": unittest.main()