sysadmin
d302602567
Add shared workflow_scope_guard with typed blockers (blocker_kind +
exact_next_action) and force-on production enforcement under pytest.
Wire root/branches/scope checks through verify_preflight_purity and
mutation entrypoints (create_issue, comment_issue) without adopting the
rejected 300a4ca patterns (test-mode early-return, porcelain *.py filter).
Regression suite covers out-of-scope issue ownership, root diagnostic
edits, worktree bind success path, force-on under pytest, porcelain
integrity, monkeypatch resistance, real entrypoint fail-closed proof,
and durable failure recording.
201 lines
9.8 KiB
Python
201 lines
9.8 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
# Ensure we import from the repo root
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import gitea_mcp_server as srv
|
|
|
|
FAKE_AUTH = {"Authorization": "token test-token"}
|
|
# Stable control checkout (parent of branches/), not the MCP server worktree root.
|
|
current_file_path = Path(__file__).resolve()
|
|
if "branches" in current_file_path.parts:
|
|
CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3])
|
|
else:
|
|
CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1])
|
|
PROJECT_ROOT = srv.PROJECT_ROOT
|
|
|
|
|
|
class TestCreateIssueWorkspaceGuard(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
# Reset preflight flags
|
|
srv._preflight_whoami_called = True
|
|
srv._preflight_capability_called = True
|
|
srv._preflight_resolved_role = "author"
|
|
srv._preflight_whoami_violation = False
|
|
srv._preflight_capability_violation = False
|
|
|
|
# Disable early return in verify_preflight_purity for testing
|
|
self._orig_in_test = srv._preflight_in_test_mode
|
|
srv._preflight_in_test_mode = lambda: False
|
|
|
|
def tearDown(self):
|
|
srv._preflight_in_test_mode = self._orig_in_test
|
|
|
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
|
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
|
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
|
@patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
|
|
@patch("gitea_mcp_server.api_request")
|
|
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
|
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
|
|
@patch(
|
|
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
|
return_value={
|
|
"current_branch": "master",
|
|
"head_sha": "a" * 40,
|
|
"porcelain_status": "",
|
|
},
|
|
)
|
|
def test_create_issue_stable_checkout_rejected(
|
|
self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
|
|
):
|
|
# Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
|
|
# path is the stable control checkout (not under branches/), mutation must fail.
|
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
|
try:
|
|
res = srv.gitea_create_issue(title="Test issue", body="body text")
|
|
except RuntimeError as exc:
|
|
self.assertIn("stable control checkout", str(exc))
|
|
else:
|
|
# #683: production guards return typed blockers at entrypoints
|
|
self.assertFalse(res.get("success"))
|
|
self.assertFalse(res.get("performed"))
|
|
blob = " ".join(res.get("reasons") or []) + " " + str(
|
|
res.get("blocker_kind") or ""
|
|
)
|
|
self.assertTrue(
|
|
"stable control checkout" in blob
|
|
or "missing_issue_worktree" in blob
|
|
or "control checkout" in blob.lower()
|
|
)
|
|
self.assertTrue(res.get("exact_next_action"))
|
|
|
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
|
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
|
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
|
@patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
|
|
@patch("gitea_mcp_server.api_request")
|
|
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
|
@patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
|
|
@patch("os.path.exists", return_value=True)
|
|
@patch("os.path.isdir", return_value=True)
|
|
@patch("subprocess.run")
|
|
def test_create_issue_valid_worktree_succeeds(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
|
|
# Mock subprocess.run for git --git-common-dir to return PROJECT_ROOT/.git
|
|
mock_res = MagicMock()
|
|
mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
|
|
mock_run.return_value = mock_res
|
|
|
|
mock_api.return_value = {"number": 42, "html_url": "https://gitea.example.com/issues/42"}
|
|
|
|
# Provide a valid branches path under the control checkout root
|
|
valid_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
|
|
|
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
|
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
|
|
res = srv.gitea_create_issue(
|
|
title="Test issue", body="body", worktree_path=valid_path
|
|
)
|
|
|
|
self.assertEqual(res["number"], 42)
|
|
mock_api.assert_called_once()
|
|
|
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
|
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
|
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
|
@patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
|
|
@patch("gitea_mcp_server.api_request")
|
|
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
|
@patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
|
|
def test_create_issue_missing_worktree_fails_closed(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
|
|
# Path under branches/ but doesn't exist
|
|
missing_path = os.path.join(
|
|
CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-worktree-path-999"
|
|
)
|
|
|
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
|
try:
|
|
res = srv.gitea_create_issue(
|
|
title="Test issue", body="body", worktree_path=missing_path
|
|
)
|
|
except RuntimeError as exc:
|
|
self.assertIn("does not exist", str(exc))
|
|
else:
|
|
self.assertFalse(res.get("success"))
|
|
blob = " ".join(res.get("reasons") or [])
|
|
self.assertIn("does not exist", blob)
|
|
self.assertTrue(res.get("exact_next_action") or res.get("reasons"))
|
|
|
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
|
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
|
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
|
@patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
|
|
@patch("gitea_mcp_server.api_request")
|
|
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
|
@patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
|
|
@patch("os.path.exists", return_value=True)
|
|
@patch("os.path.isdir", return_value=True)
|
|
@patch("author_mutation_worktree.subprocess.run")
|
|
@patch("subprocess.run")
|
|
def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_amw_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
|
|
wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
|
|
|
|
def _subprocess_side_effect(cmd, *args, **kwargs):
|
|
mock_res = MagicMock(returncode=0)
|
|
if "--git-common-dir" in cmd:
|
|
cwd = cmd[cmd.index("-C") + 1] if "-C" in cmd else ""
|
|
if cwd == wrong_repo_path:
|
|
mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
|
|
else:
|
|
mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
|
|
else:
|
|
mock_res.stdout = ""
|
|
return mock_res
|
|
|
|
mock_run.side_effect = _subprocess_side_effect
|
|
mock_amw_run.side_effect = _subprocess_side_effect
|
|
|
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
|
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
|
|
try:
|
|
res = srv.gitea_create_issue(
|
|
title="Test issue",
|
|
body="body",
|
|
worktree_path=wrong_repo_path,
|
|
)
|
|
except RuntimeError as exc:
|
|
self.assertIn(
|
|
"does not belong to the target repository", str(exc)
|
|
)
|
|
else:
|
|
self.assertFalse(res.get("success"))
|
|
blob = " ".join(res.get("reasons") or [])
|
|
self.assertIn("does not belong to the target repository", blob)
|
|
|
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
|
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
|
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
|
def test_create_issue_fails_without_whoami_preflight(self, _ns, _prof, _auth):
|
|
srv._preflight_whoami_called = False
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
srv.gitea_create_issue(title="Test issue", body="body")
|
|
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
|
|
|
|
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
|
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
|
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
|
def test_create_issue_fails_without_capability_preflight(self, _ns, _prof, _auth):
|
|
srv._preflight_capability_called = False
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
srv.gitea_create_issue(title="Test issue", body="body")
|
|
self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx.exception))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|