"""Tests for reviewer/author namespace mismatch walls (#209).""" 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_audit import mcp_server import role_namespace_gate from review_proofs import assess_mutation_namespace_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", }, "prgs-reviewer-issue-writer": { "enabled": True, "context": "ctx", "role": "reviewer", "username": "sysadmin", "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"}, "allowed_operations": [ "gitea.read", "gitea.pr.review", "gitea.issue.create", ], "forbidden_operations": ["gitea.pr.create"], "execution_profile": "prgs-reviewer-issue-writer", }, }, "rules": {"allow_runtime_switching": False}, } ISSUE_WRITE_ENV = {"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create"} class NamespaceGateBase(unittest.TestCase): def setUp(self): 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() 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)) self.audit_path = os.path.join(self._dir.name, "audit.log") def tearDown(self): self._remotes.stop() mcp_server._IDENTITY_CACHE.clear() self._dir.cleanup() def _env(self, profile, *, audit=False): env = { "GITEA_MCP_CONFIG": self.config_path, "GITEA_MCP_PROFILE": profile, "GITEA_TOKEN_AUTHOR": "author-pass", "GITEA_TOKEN_REVIEWER": "reviewer-pass", } if audit: env["GITEA_AUDIT_LOG"] = self.audit_path return env class TestNamespaceInference(unittest.TestCase): def test_reviewer_and_author_namespaces(self): self.assertEqual( role_namespace_gate.infer_mcp_namespace("prgs-reviewer"), "gitea-reviewer", ) self.assertEqual( role_namespace_gate.infer_mcp_namespace("prgs-author"), "gitea-author", ) class TestReviewerAuthorMutationBlocks(NamespaceGateBase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_reviewer_namespace_create_issue_blocked( self, _auth, mock_api, _get_all, _role ): with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True): result = mcp_server.gitea_create_issue( title="should not post", remote="prgs") self.assertFalse(result["success"]) self.assertFalse(result["performed"]) self.assertTrue(result.get("namespace_block")) mock_api.assert_not_called() @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_reviewer_namespace_create_pr_blocked( self, _auth, mock_api, _get_all, _role ): with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True): result = mcp_server.gitea_create_pr( title="feat: x Closes #1", head="feat/issue-1-x", body="Closes #1", remote="prgs", ) self.assertFalse(result.get("success", True)) self.assertFalse(result.get("performed", True)) self.assertTrue(result.get("namespace_block")) mock_api.assert_not_called() @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request", return_value={"number": 12}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_reviewer_with_explicit_issue_create_allowed( self, _auth, mock_api, _get_all, _role ): with patch.dict(os.environ, self._env("prgs-reviewer-issue-writer"), clear=True): result = mcp_server.gitea_create_issue( title="escalation issue", remote="prgs") self.assertEqual(result.get("number"), 12) mock_api.assert_called() class TestHandoffNamespaceParity(unittest.TestCase): def test_author_handoff_with_reviewer_namespace_invalid(self): result = assess_mutation_namespace_handoff( declared_role="author", namespaces_used=["gitea-reviewer"], ) self.assertFalse(result["valid"]) self.assertTrue(result["blocked"]) def test_capability_author_profile_with_reviewer_namespace_blocked(self): result = assess_mutation_namespace_handoff( declared_role="author", namespaces_used=["gitea-reviewer"], capability_profile="prgs-author", ) self.assertFalse(result["valid"]) blob = " ".join(result["reasons"]) self.assertIn("capability proof profile", blob) class TestMutationAuditFields(NamespaceGateBase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request", return_value={"number": 3}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_successful_create_issue_audit_includes_namespace_fields( self, _auth, mock_api, _get_all, _role ): env = {**self._env("prgs-author", audit=True), **ISSUE_WRITE_ENV} with patch.dict(os.environ, env, clear=True): mcp_server.gitea_create_issue(title="audited", remote="prgs") with open(self.audit_path, encoding="utf-8") as fh: record = json.loads(fh.readline()) self.assertEqual(record["mcp_namespace"], "gitea-author") self.assertEqual(record["task_role"], "author") self.assertEqual(record["operation"], "create_issue") self.assertEqual(record["profile_name"], "prgs-author") self.assertEqual(record["remote"], "prgs") if __name__ == "__main__": unittest.main()