"""Tests for gitea_resolve_task_capability tool. Covers Issue #141 requirements and #145 regression tests for permission boundaries, self blocks, and structured guidance. """ import os import sys import json import tempfile import unittest from unittest.mock import patch, MagicMock sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import gitea_config import gitea_auth import mcp_server CONFIG_RESOLVER = { "version": 2, "contexts": { "ctx": { "enabled": True, "gitea": { "enabled": True, "base_url": "https://gitea.example.com" } } }, "profiles": { "author-profile": { "enabled": True, "context": "ctx", "role": "author", "username": "author-user", "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], "execution_profile": "author-profile" }, "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.approve", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"], "execution_profile": "reviewer-profile" }, "merger-profile": { "enabled": True, "context": "ctx", "role": "merger", "username": "merger-user", "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"}, "allowed_operations": ["gitea.read", "gitea.pr.merge", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.approve"], "execution_profile": "merger-profile" } }, "rules": { "allow_runtime_switching": False } } class TestResolveTaskCapability(unittest.TestCase): def setUp(self): self._remotes_patch = patch.dict(mcp_server.REMOTES, { "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}, "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"} }) self._remotes_patch.start() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None self._dir = tempfile.TemporaryDirectory() self.config_path = os.path.join(self._dir.name, "profiles.json") self._write_config(CONFIG_RESOLVER) def tearDown(self): self._remotes_patch.stop() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None self._dir.cleanup() def _write_config(self, obj): with open(self.config_path, "w", encoding="utf-8") as fh: fh.write(json.dumps(obj)) def _env(self, profile="author-profile"): return { "GITEA_MCP_CONFIG": self.config_path, "GITEA_MCP_PROFILE": profile, "GITEA_TOKEN_AUTHOR": "author-pass", "GITEA_TOKEN_REVIEWER": "reviewer-pass", "GITEA_TOKEN_MERGER": "merger-pass", } @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_author_capable_task(self, _auth, _api): with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs") self.assertEqual(res["requested_task"], "create_issue") self.assertEqual(res["required_operation_permission"], "gitea.issue.create") self.assertEqual(res["required_role_kind"], "author") self.assertEqual(res["active_profile"], "author-profile") self.assertTrue(res["allowed_in_current_session"]) self.assertIn("author-profile", res["matching_configured_profile"]) self.assertFalse(res["different_mcp_namespace_required"]) self.assertIn("ready for operations", res["exact_safe_next_action"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_reviewer_capable_task_blocked(self, _auth, _api): with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") self.assertEqual(res["requested_task"], "review_pr") self.assertEqual(res["required_operation_permission"], "gitea.pr.review") self.assertEqual(res["required_role_kind"], "reviewer") self.assertEqual(res["active_profile"], "author-profile") self.assertFalse(res["allowed_in_current_session"]) self.assertIn("reviewer-profile", res["matching_configured_profile"]) self.assertTrue(res["different_mcp_namespace_required"]) self.assertIn("gitea-reviewer", res["exact_safe_next_action"]) @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_resolve_reviewer_capable_task_allowed(self, _auth, _api): with patch.dict(os.environ, self._env("reviewer-profile")): res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") self.assertTrue(res["allowed_in_current_session"]) self.assertFalse(res["different_mcp_namespace_required"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_work_issue_author_profile_allowed(self, _auth, _api): with patch.dict(os.environ, self._env("author-profile")): for task in ("work_issue", "work-issue"): res = mcp_server.gitea_resolve_task_capability( task=task, remote="prgs" ) self.assertEqual(res["requested_task"], task) self.assertEqual( res["required_operation_permission"], "gitea.pr.create" ) self.assertEqual(res["required_role_kind"], "author") self.assertTrue(res["allowed_in_current_session"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api): with patch.dict(os.environ, self._env("reviewer-profile")): res = mcp_server.gitea_resolve_task_capability( task="work_issue", remote="prgs" ) self.assertFalse(res["allowed_in_current_session"]) self.assertEqual(res["required_role_kind"], "author") self.assertTrue(res["different_mcp_namespace_required"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_issue_comment_task(self, _auth, _api): with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs") self.assertTrue(res["allowed_in_current_session"]) self.assertIn("author-profile", res["matching_configured_profile"]) self.assertIn("reviewer-profile", res["matching_configured_profile"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_mark_issue_author_profile_allowed(self, _auth, _api): # #183: mark_issue must map to gitea.issue.comment (prgs-author allowed op). with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="mark_issue", remote="prgs") self.assertEqual(res["requested_task"], "mark_issue") self.assertEqual(res["required_operation_permission"], "gitea.issue.comment") self.assertTrue(res["allowed_in_current_session"]) self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api): # #275: lock_issue must be an exact resolver task for claim/lock gates. with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs") self.assertEqual(res["requested_task"], "lock_issue") self.assertEqual(res["required_operation_permission"], "gitea.issue.comment") self.assertTrue(res["allowed_in_current_session"]) self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) def test_resolve_unknown_task_fails_closed(self): with patch.dict(os.environ, self._env("author-profile")): with self.assertRaises(ValueError): mcp_server.gitea_resolve_task_capability(task="invalid_task_xyz", remote="prgs") # Additional regression tests per #145 for permission boundaries and structured guidance def test_issue_comment_does_not_imply_close(self): # Author profile has issue.comment but not issue.close with patch.dict(os.environ, self._env("author-profile")): res_comment = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs") res_close = mcp_server.gitea_resolve_task_capability(task="close_issue", remote="prgs") self.assertTrue(res_comment["allowed_in_current_session"]) self.assertIn("gitea.issue.comment", res_comment["active_profile_allowed_operations"]) self.assertFalse(res_close["allowed_in_current_session"]) self.assertIn("gitea.issue.close", res_close.get("required_operation_permission", "")) def test_pr_comment_does_not_imply_issue_comment(self): # Create a profile that has pr.comment but not issue.comment config = dict(CONFIG_RESOLVER) config["profiles"]["pr-only"] = { "enabled": True, "context": "ctx", "role": "author", "username": "pr-user", "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": ["gitea.read", "gitea.pr.comment"], "forbidden_operations": ["gitea.issue.comment"], "execution_profile": "pr-only" } self._write_config(config) with patch.dict(os.environ, {**self._env("pr-only"), "GITEA_MCP_PROFILE": "pr-only"}): res_pr = mcp_server.gitea_resolve_task_capability(task="comment_pr", remote="prgs") res_issue = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs") self.assertTrue(res_pr["allowed_in_current_session"]) self.assertFalse(res_issue["allowed_in_current_session"]) self.assertIn("gitea.pr.comment", res_pr["active_profile_allowed_operations"]) self.assertNotIn("gitea.issue.comment", res_pr["active_profile_allowed_operations"]) @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_reviewer_cannot_push_branch(self, _auth, _api): with patch.dict(os.environ, self._env("reviewer-profile")): res = mcp_server.gitea_resolve_task_capability(task="push_branch", remote="prgs") self.assertFalse(res["allowed_in_current_session"]) self.assertIn("author-profile", res["matching_configured_profile"]) self.assertTrue(res["different_mcp_namespace_required"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_structured_guidance_for_missing_permission(self, _auth, _api): with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="merge_pr", remote="prgs") self.assertFalse(res["allowed_in_current_session"]) self.assertIn("merger", res["exact_safe_next_action"].lower()) self.assertEqual(res["required_operation_permission"], "gitea.pr.merge") self.assertEqual(res["required_role_kind"], "merger") @patch("mcp_server.api_request") def test_self_review_blocked_structured(self, mock_api): # Test that self-approve (self-review gate) is blocked and returns structured reasons def side_api(method, url, auth): if "/user" in url: return {"login": "author-user"} if "/pulls/123" in url: return {"user": {"login": "author-user"}, "state": "open", "head": {"sha": "abc"}, "mergeable": True} return {} mock_api.side_effect = side_api with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_check_pr_eligibility(pr_number=123, action="approve", remote="prgs") self.assertFalse(res.get("eligible", True)) reasons = res.get("reasons", []) self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0) # Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close. @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api): # Default author profile has PR create/comment but not close: the # close capability must never be implied by broader author ops. with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") self.assertEqual(res["requested_task"], "close_pr") self.assertEqual(res["required_operation_permission"], "gitea.pr.close") self.assertEqual(res["required_role_kind"], "author") self.assertFalse(res["allowed_in_current_session"]) self.assertTrue(res["stop_required"]) self.assertEqual(res["matching_configured_profile"], []) self.assertTrue(res["different_mcp_namespace_required"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api): # Operator-granted close capability resolves cleanly under an author profile. config = json.loads(json.dumps(CONFIG_RESOLVER)) config["profiles"]["author-closer"] = { "enabled": True, "context": "ctx", "role": "author", "username": "author-user", "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": ["gitea.read", "gitea.pr.close"], "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], "execution_profile": "author-closer", } self._write_config(config) with patch.dict(os.environ, self._env("author-closer")): res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") self.assertTrue(res["allowed_in_current_session"]) self.assertFalse(res["stop_required"]) self.assertIn("author-closer", res["matching_configured_profile"]) self.assertIn("ready for operations", res["exact_safe_next_action"]) @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api): # close_pr is author-side: a reviewer profile must be told to stop. with patch.dict(os.environ, self._env("reviewer-profile")): res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") self.assertFalse(res["allowed_in_current_session"]) self.assertTrue(res["stop_required"]) self.assertEqual(res["required_role_kind"], "author") self.assertIn("author", res["exact_safe_next_action"].lower()) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_reconcile_already_landed_pr_requires_close(self, _auth, _api): with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability( task="reconcile_already_landed_pr", remote="prgs" ) self.assertEqual(res["requested_task"], "reconcile_already_landed_pr") self.assertEqual(res["required_operation_permission"], "gitea.pr.close") self.assertEqual(res["required_role_kind"], "reconciler") self.assertFalse(res["allowed_in_current_session"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_pr_queue_cleanup_author_profile_blocked(self, _auth, _api): with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability( task="pr_queue_cleanup", remote="prgs" ) self.assertEqual(res["required_role_kind"], "reviewer") self.assertFalse(res["allowed_in_current_session"]) self.assertTrue(res["stop_required"]) @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_resolve_pr_queue_cleanup_reviewer_profile_allowed(self, _auth, _api): with patch.dict(os.environ, self._env("reviewer-profile")): res = mcp_server.gitea_resolve_task_capability( task="pr_queue_cleanup", remote="prgs" ) self.assertEqual(res["required_operation_permission"], "gitea.pr.review") self.assertTrue(res["allowed_in_current_session"]) self.assertFalse(res["stop_required"]) @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api): # close_pr resolves (no Unknown-task error); near-miss spellings keep # failing closed so no untracked close fallback can be rationalized. with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") self.assertEqual(res["required_operation_permission"], "gitea.pr.close") for unknown in ("close_pull_request", "close", "pr_close"): with self.assertRaises(ValueError): mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs") if __name__ == "__main__": unittest.main()