"""Tests for namespace-scoped MCP workspace binding (#510).""" from __future__ import annotations import os import sys import unittest from pathlib import Path from unittest import mock from unittest.mock import MagicMock sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import gitea_mcp_server as srv # noqa: E402 import namespace_workspace_binding as nwb # noqa: E402 REPO_ROOT = Path(__file__).resolve().parent.parent if REPO_ROOT.parent.name == "branches": CONTROL_ROOT = str(REPO_ROOT.parent.parent) else: CONTROL_ROOT = str(REPO_ROOT) AUTHOR_DIRTY = f"{CONTROL_ROOT}/branches/mcp-author-worktree" MERGER_CLEAN = f"{CONTROL_ROOT}/branches/merge-pr487-submit" REVIEWER_CLEAN = f"{CONTROL_ROOT}/branches/review-pr487-submit" RECONCILER_CLEAN = f"{CONTROL_ROOT}/branches/reconcile-pr487" MCP_PROCESS_ROOT = CONTROL_ROOT class TestNamespaceWorkspaceModule(unittest.TestCase): def test_author_env_ignored_for_merger_namespace(self): workspace, source = nwb.resolve_namespace_workspace( role_kind="merger", worktree_path=None, process_project_root=MCP_PROCESS_ROOT, env={ nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, nwb.MERGER_WORKTREE_ENV: MERGER_CLEAN, }, profile_name="gitea-merger", ) self.assertEqual(workspace, os.path.realpath(MERGER_CLEAN)) self.assertEqual(source, f"{nwb.MERGER_WORKTREE_ENV} environment variable") def test_author_env_ignored_for_reviewer_namespace(self): workspace, source = nwb.resolve_namespace_workspace( role_kind="reviewer", worktree_path=None, process_project_root=MCP_PROCESS_ROOT, env={ nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, nwb.REVIEWER_WORKTREE_ENV: REVIEWER_CLEAN, }, ) self.assertEqual(workspace, os.path.realpath(REVIEWER_CLEAN)) self.assertEqual(source, f"{nwb.REVIEWER_WORKTREE_ENV} environment variable") def test_author_env_ignored_for_reconciler_namespace(self): workspace, source = nwb.resolve_namespace_workspace( role_kind="reconciler", worktree_path=None, process_project_root=MCP_PROCESS_ROOT, env={ nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, nwb.RECONCILER_WORKTREE_ENV: RECONCILER_CLEAN, }, ) self.assertEqual(workspace, os.path.realpath(RECONCILER_CLEAN)) self.assertEqual(source, f"{nwb.RECONCILER_WORKTREE_ENV} environment variable") def test_merger_profile_maps_to_merger_namespace(self): role = nwb.normalize_role_kind("reviewer", profile_name="gitea-merger") self.assertEqual(role, "merger") def test_error_message_includes_path_and_binding_source(self): msg = nwb.format_namespace_workspace_binding_error( role_kind="merger", workspace_path=AUTHOR_DIRTY, binding_source=f"{nwb.AUTHOR_WORKTREE_ENV} environment variable", dirty_files=["gitea_mcp_server.py"], ignored_bindings=[f"{nwb.AUTHOR_WORKTREE_ENV}={AUTHOR_DIRTY} (ignored for merger namespace)"], ) self.assertIn(AUTHOR_DIRTY, msg) self.assertIn("via", msg.lower()) self.assertIn(nwb.AUTHOR_WORKTREE_ENV, msg) self.assertIn("Do not clean or reset foreign role worktrees", msg) class TestNamespaceWorkspaceIntegration(unittest.TestCase): def setUp(self): self._saved = { "whoami_called": srv._preflight_whoami_called, "capability_called": srv._preflight_capability_called, "resolved_role": srv._preflight_resolved_role, "whoami_violation": srv._preflight_whoami_violation, "capability_violation": srv._preflight_capability_violation, "in_test": srv._preflight_in_test_mode, } srv._preflight_whoami_called = True srv._preflight_capability_called = True srv._preflight_whoami_violation = False srv._preflight_capability_violation = False srv._preflight_in_test_mode = lambda: False self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False) self._env_patch.start() def tearDown(self): srv._preflight_whoami_called = self._saved["whoami_called"] srv._preflight_capability_called = self._saved["capability_called"] srv._preflight_resolved_role = self._saved["resolved_role"] srv._preflight_whoami_violation = self._saved["whoami_violation"] srv._preflight_capability_violation = self._saved["capability_violation"] srv._preflight_in_test_mode = self._saved["in_test"] self._env_patch.stop() for key in ( nwb.AUTHOR_WORKTREE_ENV, nwb.MERGER_WORKTREE_ENV, nwb.REVIEWER_WORKTREE_ENV, nwb.RECONCILER_WORKTREE_ENV, nwb.ACTIVE_WORKTREE_ENV, ): os.environ.pop(key, None) def _merger_profile(self): return { "profile_name": "gitea-merger", "allowed_operations": ["gitea.pr.merge", "gitea.read"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], } def _reviewer_profile(self): return { "profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.approve", "gitea.pr.review", "gitea.read"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], } def _reconciler_profile(self): return { "profile_name": "prgs-reconciler", "allowed_operations": ["gitea.pr.close", "gitea.read"], "forbidden_operations": [ "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.create", ], } def _author_profile(self): return { "profile_name": "prgs-author", "allowed_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.read"], "forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"], } @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_dirty_author_worktree_does_not_block_merger_with_clean_workspace( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN srv._preflight_resolved_role = "reviewer" with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN) @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_dirty_author_worktree_does_not_block_reviewer_with_clean_workspace( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY os.environ[nwb.REVIEWER_WORKTREE_ENV] = REVIEWER_CLEAN srv._preflight_resolved_role = "reviewer" with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch("gitea_mcp_server.get_profile", return_value=self._reviewer_profile()): srv.verify_preflight_purity("prgs", worktree_path=REVIEWER_CLEAN) @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_dirty_author_worktree_does_not_block_reconciler_with_clean_workspace( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY os.environ[nwb.RECONCILER_WORKTREE_ENV] = RECONCILER_CLEAN srv._preflight_resolved_role = "reconciler" with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch("gitea_mcp_server.get_profile", return_value=self._reconciler_profile()): srv.verify_preflight_purity("prgs", worktree_path=RECONCILER_CLEAN) @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_dirty_active_task_workspace_still_blocks_mutations( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN srv._preflight_resolved_role = "reviewer" dirty = " M namespace_workspace_binding.py\n" with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): with mock.patch("gitea_mcp_server._get_workspace_porcelain", return_value=dirty): with self.assertRaises(RuntimeError) as ctx: srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN) self.assertIn(MERGER_CLEAN, str(ctx.exception)) self.assertIn("binding", str(ctx.exception).lower()) def test_root_workspace_mutation_still_blocked_for_author(self): srv._preflight_resolved_role = "author" with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT): with mock.patch("gitea_mcp_server.get_profile", return_value=self._author_profile()): with mock.patch( "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "master"}, ): with self.assertRaises(RuntimeError) as ctx: srv.verify_preflight_purity("prgs") self.assertIn("stable control checkout", str(ctx.exception)) @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_pr487_style_merge_binds_clean_merger_workspace( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY srv._preflight_resolved_role = "reviewer" clean_git_state = { "current_branch": "master", "porcelain_status": "", "head_sha": "abc123", } with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): with mock.patch( "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value=clean_git_state, ): with mock.patch( "gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="abc123", ): resolved = srv._verify_role_mutation_workspace("prgs") self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT))