"""Tests for operation-scoped role selection and automatic dispatch switching (#228).""" 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 from reviewer_worktree import assess_author_worktree_continuity CONFIG_TEST = { "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.pr.merge", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], "execution_profile": "reviewer-profile" } } } class TestOperationScopedRoles(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 mcp_server._MUTATION_AUTHORITY = None self._dir = tempfile.TemporaryDirectory() self.config_path = os.path.join(self._dir.name, "profiles.json") self._write_config(CONFIG_TEST) def tearDown(self): self._remotes_patch.stop() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None mcp_server._MUTATION_AUTHORITY = 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", with_reviewer_token=True): env = { "GITEA_MCP_CONFIG": self.config_path, "GITEA_MCP_PROFILE": profile, "GITEA_TOKEN_AUTHOR": "author-pass", } if with_reviewer_token: env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass" return env @patch("mcp_server.api_request") def test_auto_switch_to_reviewer(self, mock_api): # mock identity resolution to return username matching profile mock_api.side_effect = lambda method, url, header: ( {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} ) with patch.dict(os.environ, self._env("author-profile")): # initially we are author-profile self.assertEqual(gitea_config.selected_profile_name(), "author-profile") self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile") # resolve a reviewer task (review_pr) res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") # verify it automatically switched to reviewer-profile self.assertTrue(res["allowed_in_current_session"]) self.assertTrue(res["available_in_session"]) self.assertEqual(res["active_profile"], "reviewer-profile") self.assertEqual(res["active_identity"], "reviewer-user") self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") @patch("mcp_server.api_request") def test_auto_switch_to_author(self, mock_api): mock_api.side_effect = lambda method, url, header: ( {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} ) with patch.dict(os.environ, self._env("reviewer-profile")): # initially we are reviewer-profile self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") # resolve an author task (create_issue) res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs") # verify it automatically switched to author-profile self.assertTrue(res["allowed_in_current_session"]) self.assertTrue(res["available_in_session"]) self.assertEqual(res["active_profile"], "author-profile") self.assertEqual(res["active_identity"], "author-user") self.assertEqual(gitea_config.selected_profile_name(), "author-profile") @patch("mcp_server.api_request") def test_restart_required_when_unattached(self, mock_api): mock_api.side_effect = lambda method, url, header: {"login": "author-user"} # launch without reviewer token in env with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)): self.assertEqual(gitea_config.selected_profile_name(), "author-profile") # resolve a reviewer task (review_pr) res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") # verify it did NOT switch and reports restart_required self.assertFalse(res["allowed_in_current_session"]) self.assertFalse(res["available_in_session"]) self.assertTrue(res["configured"]) self.assertTrue(res["restart_required"]) self.assertTrue(res["stop_required"]) self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"]) def test_author_continuity_dirty_worktree(self): # author is allowed to keep dirty worktree res = assess_author_worktree_continuity({ "task_role": "author", "dirty_files": ["review_proofs.py"], }) self.assertTrue(res["allowed"]) # reviewer is blocked by reviewer worktree proof res2 = assess_author_worktree_continuity({ "task_role": "reviewer", "worktree_path": "/repo", "dirty_files": ["review_proofs.py"], "pr_scope_files": ["docs/wiki/Repositories.md"], "scratch_used": False, }) self.assertFalse(res2["proven"]) @patch("mcp_server.api_request") def test_mutating_actions_auto_switch(self, mock_api): mock_api.side_effect = lambda method, url, header: ( {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} ) with patch.dict(os.environ, self._env("author-profile")): # verify we are author-profile self.assertEqual(gitea_config.selected_profile_name(), "author-profile") # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission) blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs") self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block) # verify we dynamically switched to reviewer-profile self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") if __name__ == "__main__": unittest.main()