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. CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3]) 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): with self.assertRaises(RuntimeError) as ctx: srv.gitea_create_issue(title="Test issue", body="body text") self.assertIn("stable control checkout", 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) @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): with self.assertRaises(RuntimeError) as ctx: srv.gitea_create_issue( title="Test issue", body="body", worktree_path=missing_path ) self.assertIn("does not exist (fail closed)", 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) @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=""): with self.assertRaises(RuntimeError) as ctx: srv.gitea_create_issue( title="Test issue", body="body", worktree_path=wrong_repo_path ) self.assertIn("does not belong to the target repository", 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_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()