"""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()