sysadmin
3d11e1f12b
Namespace-scoped workspace resolution prevents GITEA_AUTHOR_WORKTREE from poisoning reviewer, merger, and reconciler purity checks. Each role uses its own worktree env var with actionable binding-source errors and safe reconnect guidance. Preserves #274/#475 author and control-checkout guards.
240 lines
11 KiB
Python
240 lines
11 KiB
Python
"""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"
|
|
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
|
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
|
|
resolved = srv._verify_role_mutation_workspace("prgs")
|
|
self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT)) |