feat: map commit_files in task capability resolver (#262)
Add commit_files and gitea_commit_files resolver entries for gitea.repo.commit, align gitea_commit_files tool gates with other issue-write mutations, and add regression tests for author allow, reviewer deny, preflight order, and resolver/tool alignment. Refs #262 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
@@ -0,0 +1,199 @@
|
||||
"""Tests for commit_files task capability resolver mapping (#262)."""
|
||||
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": {
|
||||
"commit-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": ["gitea.read", "gitea.repo.commit"],
|
||||
"forbidden_operations": ["gitea.pr.create"],
|
||||
"execution_profile": "commit-author",
|
||||
},
|
||||
"pr-only-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "pr-author",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.create"],
|
||||
"forbidden_operations": ["gitea.repo.commit"],
|
||||
"execution_profile": "pr-only-author",
|
||||
},
|
||||
"reviewer-profile": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "reviewer",
|
||||
"username": "reviewer-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.merge"],
|
||||
"forbidden_operations": [
|
||||
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push",
|
||||
],
|
||||
"execution_profile": "reviewer-profile",
|
||||
},
|
||||
},
|
||||
"rules": {"allow_runtime_switching": False},
|
||||
}
|
||||
|
||||
|
||||
class CommitFilesCapabilityBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._preflight_snapshot = (
|
||||
mcp_server._preflight_whoami_called,
|
||||
mcp_server._preflight_capability_called,
|
||||
)
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
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()
|
||||
mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
|
||||
self._preflight_snapshot
|
||||
)
|
||||
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",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
|
||||
|
||||
class TestCommitFilesResolver(CommitFilesCapabilityBase):
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_resolves_to_repo_commit(self, _auth, _api):
|
||||
with patch.dict(os.environ, self._env("commit-author"), clear=True):
|
||||
for task in ("commit_files", "gitea_commit_files"):
|
||||
with self.subTest(task=task):
|
||||
res = mcp_server.gitea_resolve_task_capability(
|
||||
task=task, remote="prgs")
|
||||
self.assertEqual(res["required_operation_permission"],
|
||||
"gitea.repo.commit")
|
||||
self.assertEqual(res["required_role_kind"], "author")
|
||||
self.assertTrue(res["allowed_in_current_session"])
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "pr-author"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_create_pr_does_not_satisfy_commit_files(self, _auth, _api):
|
||||
with patch.dict(os.environ, self._env("pr-only-author"), clear=True):
|
||||
commit_res = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs")
|
||||
pr_res = mcp_server.gitea_resolve_task_capability(
|
||||
task="create_pr", remote="prgs")
|
||||
self.assertFalse(commit_res["allowed_in_current_session"])
|
||||
self.assertTrue(pr_res["allowed_in_current_session"])
|
||||
self.assertEqual(commit_res["required_operation_permission"],
|
||||
"gitea.repo.commit")
|
||||
self.assertEqual(pr_res["required_operation_permission"],
|
||||
"gitea.pr.create")
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_reviewer_denied_commit_files(self, _auth, _api):
|
||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
||||
res = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs")
|
||||
self.assertFalse(res["allowed_in_current_session"])
|
||||
self.assertEqual(res["required_operation_permission"],
|
||||
"gitea.repo.commit")
|
||||
self.assertTrue(res["different_mcp_namespace_required"])
|
||||
|
||||
|
||||
class TestCommitFilesToolGate(CommitFilesCapabilityBase):
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_allowed_author_proceeds(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "abc"},
|
||||
"branch": {"name": "feat/x"},
|
||||
}
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
with patch.dict(os.environ, self._env("commit-author"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
|
||||
message="test",
|
||||
new_branch="feat/x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_reviewer_blocked_with_permission_report(self, _auth, _role):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
|
||||
message="test",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res.get("performed", True))
|
||||
self.assertIn("permission_report", res)
|
||||
self.assertEqual(res["permission_report"]["missing_permission"],
|
||||
"gitea.repo.commit")
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
def test_preflight_blocks_before_api(self, _role):
|
||||
env = {**self._env("commit-author"), "GITEA_TEST_FORCE_DIRTY": "1"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
|
||||
message="test",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("gitea_whoami", str(ctx.exception))
|
||||
|
||||
|
||||
class TestCommitFilesMapAlignment(unittest.TestCase):
|
||||
def test_tool_gate_matches_resolver(self):
|
||||
self.assertEqual(
|
||||
task_capability_map.tool_required_permission("gitea_commit_files"),
|
||||
task_capability_map.required_permission("commit_files"),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1069,11 +1069,22 @@ class TestGetFile(unittest.TestCase):
|
||||
# ---------------------------------------------------------------------------
|
||||
# Commit Files
|
||||
# ---------------------------------------------------------------------------
|
||||
COMMIT_FILES_ENV = {
|
||||
"GITEA_PROFILE_NAME": "commit-author",
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.repo.commit",
|
||||
"GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.approve,gitea.pr.merge",
|
||||
}
|
||||
|
||||
|
||||
class TestCommitFiles(unittest.TestCase):
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_commit_files_success(self, _auth, mock_api):
|
||||
def test_commit_files_success(self, _auth, mock_api, _role):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "commit-sha-123"},
|
||||
"branch": {"name": "test-branch"}
|
||||
@@ -1081,11 +1092,12 @@ class TestCommitFiles(unittest.TestCase):
|
||||
files = [
|
||||
{"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
|
||||
]
|
||||
result = gitea_commit_files(
|
||||
files=files,
|
||||
message="Initial commit",
|
||||
new_branch="test-branch"
|
||||
)
|
||||
with patch.dict(os.environ, COMMIT_FILES_ENV, clear=False):
|
||||
result = gitea_commit_files(
|
||||
files=files,
|
||||
message="Initial commit",
|
||||
new_branch="test-branch"
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(result["commit"], "commit-sha-123")
|
||||
self.assertEqual(result["branch"], "test-branch")
|
||||
|
||||
Reference in New Issue
Block a user