From 75ed0632b43620afc3072f98972da47f77e33664 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 11:27:39 -0400 Subject: [PATCH] feat(mcp): implement operation-scoped role selection and automatic dispatch switching (#228) --- gitea_config.py | 10 +- tests/test_operation_scoped_roles.py | 181 +++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 1 deletion(-) create mode 100644 tests/test_operation_scoped_roles.py diff --git a/gitea_config.py b/gitea_config.py index a4b0821..ffbf7ea 100644 --- a/gitea_config.py +++ b/gitea_config.py @@ -205,7 +205,7 @@ def selected_profile_name(): def is_runtime_switching_enabled(path=None): - """Check if runtime profile switching is explicitly enabled in config.""" + """Check if runtime profile switching is enabled in config.""" try: config = load_config(path) except Exception: @@ -213,10 +213,18 @@ def is_runtime_switching_enabled(path=None): if not config: return False rules = config.get("rules") or {} + if rules.get("allow_runtime_switching") is False: + return False + if config.get("allow_runtime_switching") is False: + return False if rules.get("allow_runtime_switching") is True: return True if config.get("allow_runtime_switching") is True: return True + # Default to True if multiple profiles exist in the config + profiles = config.get("profiles") or {} + if len(profiles) > 1: + return True return False diff --git a/tests/test_operation_scoped_roles.py b/tests/test_operation_scoped_roles.py new file mode 100644 index 0000000..c382085 --- /dev/null +++ b/tests/test_operation_scoped_roles.py @@ -0,0 +1,181 @@ +"""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()