Merge pull request 'feat(workflow): pre-task role/session router blocks reviewer tasks in author sessions (Issue #206)' (#217) from feat/issue-206-role-session-router into master

tests/test_role_session_router.py

@@ -0,0 +1,188 @@
+"""Tests for pre-task role/session router (#206)."""
+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 gitea_config
+import mcp_server
+import role_session_router
+from review_proofs import assess_role_route_handoff
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "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", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+        "prgs-reviewer": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "sysadmin",
+            "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", "gitea.issue.create",
+            ],
+            "execution_profile": "prgs-reviewer",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestRoleSessionRouter(unittest.TestCase):
+    def setUp(self):
+        role_session_router.clear_route_state()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.example.com",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Gitea-Tools",
+            },
+        })
+        self._remotes.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")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        role_session_router.clear_route_state()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self, profile):
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+
+    def test_reviewer_task_under_author_profile_wrong_role_stop(self):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="review_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
+        self.assertFalse(route["downstream_allowed"])
+        self.assertIn("Wrong role/session for reviewer task", route["message"])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_issue_creation_blocked_after_reviewer_wrong_role_stop(
+        self, _auth, mock_api
+    ):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
+            result = mcp_server.gitea_create_issue(
+                title="process issue fallback",
+                body="should not post",
+                remote="prgs",
+            )
+        self.assertFalse(result.get("success", True))
+        self.assertFalse(result.get("performed", True))
+        issue_posts = [
+            c for c in mock_api.call_args_list
+            if c.args[0] == "POST" and str(c.args[1]).endswith("/issues")
+        ]
+        self.assertEqual(issue_posts, [])
+
+    @patch("mcp_server.api_request", return_value={"number": 999})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_author_task_allowed_after_explicit_author_route(
+        self, _auth, _api
+    ):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="create_issue", remote="prgs"
+            )
+            self.assertEqual(
+                route["route_result"], role_session_router.ROUTE_ALLOWED
+            )
+            result = mcp_server.gitea_create_issue(
+                title="legit author task",
+                remote="prgs",
+            )
+        self.assertEqual(result.get("number"), 999)
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="review_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+        self.assertTrue(route["downstream_allowed"])
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="create_issue", remote="prgs"
+            )
+        self.assertEqual(
+            route["route_result"], role_session_router.ROUTE_TO_AUTHOR
+        )
+        self.assertFalse(route["downstream_allowed"])
+
+    def test_activate_profile_blocked_in_static_mode(self):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            result = mcp_server.gitea_activate_profile(
+                profile_name="prgs-reviewer", remote="prgs"
+            )
+        self.assertFalse(result["success"])
+        self.assertIn("switching is disabled", result["message"])
+
+    def test_handoff_requires_route_fields(self):
+        incomplete = assess_role_route_handoff("Task done.")
+        self.assertFalse(incomplete["complete"])
+        self.assertIn("Task type", incomplete["missing_fields"])
+
+        complete = assess_role_route_handoff(
+            "\n".join([
+                "Task type: review_pr",
+                "Required role: reviewer",
+                "Active role: author",
+                "Route result: wrong_role_stop",
+            ]),
+            route_result={"route_result": "wrong_role_stop"},
+        )
+        self.assertTrue(complete["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()