sysadminandClaude Opus 4.8
00394d43b7
Extract TASK_CAPABILITY_MAP into task_capability_map.py as the single source of truth shared by gitea_resolve_task_capability and issue-mutating tools. Gate gitea_create_issue, gitea_close_issue, gitea_mark_issue, and gitea_set_issue_labels with structured #142 permission reports. Add regression tests proving resolver-denied tasks fail closed at the raw tool layer. Implements mcp-control-plane #69. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
210 lines
8.6 KiB
Python
210 lines
8.6 KiB
Python
"""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_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api):
|
|
def api_side_effect(method, url, auth, payload=None):
|
|
if method == "GET" and url.endswith("/labels?limit=100"):
|
|
return [{"id": 10, "name": "status:in-progress"}]
|
|
if method == "POST" and "/issues/9/labels" in url:
|
|
return [{"name": "status:in-progress"}]
|
|
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() |