"""Tests for reviewer mutation workspace binding (#503).""" 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 reviewer_mutation_workspace as rmw # noqa: E402 import reviewer_pr_lease # 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) BRANCHES_WORKTREE = f"{CONTROL_ROOT}/branches/review-pr503-submit" MCP_PROCESS_ROOT = CONTROL_ROOT class TestReviewerMutationWorkspaceModule(unittest.TestCase): def test_resolve_prefers_explicit_worktree_path(self): resolved = rmw.resolve_reviewer_mutation_workspace( worktree_path=BRANCHES_WORKTREE, worktree="/ignored", process_project_root=MCP_PROCESS_ROOT, active_worktree_env="/env/path", session_lease_worktree="/lease/path", ) self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE)) def test_resolve_falls_back_to_session_lease(self): resolved = rmw.resolve_reviewer_mutation_workspace( worktree_path=None, worktree=None, process_project_root=MCP_PROCESS_ROOT, session_lease_worktree=BRANCHES_WORKTREE, ) self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE)) def test_metadata_only_binding_detected(self): assessment = rmw.assess_metadata_only_worktree_binding( declared_worktree_path=BRANCHES_WORKTREE, mutation_workspace=MCP_PROCESS_ROOT, process_project_root=MCP_PROCESS_ROOT, ) self.assertTrue(assessment["block"]) self.assertTrue(assessment["metadata_only"]) self.assertIn("metadata-only", assessment["reasons"][0]) def test_canonical_error_mentions_fix_paths(self): msg = rmw.format_stale_mcp_workspace_binding_error( process_project_root=MCP_PROCESS_ROOT, mutation_workspace=MCP_PROCESS_ROOT, declared_worktree_path=BRANCHES_WORKTREE, metadata_only=True, ) self.assertIn("GITEA_ACTIVE_WORKTREE", msg) self.assertIn("worktree_path", msg) self.assertIn("branches/", msg) class TestReviewerMutationWorkspaceIntegration(unittest.TestCase): def setUp(self): reviewer_pr_lease.clear_session_lease() 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, "whoami_files": list(srv._preflight_whoami_violation_files), "capability_violation": srv._preflight_capability_violation, "capability_files": list(srv._preflight_capability_violation_files), "reviewer_files": list(srv._preflight_reviewer_violation_files), "in_test": srv._preflight_in_test_mode, } srv._preflight_whoami_called = True srv._preflight_capability_called = True srv._preflight_resolved_role = "reviewer" srv._preflight_whoami_violation = True srv._preflight_whoami_violation_files = ["gitea_mcp_server.py"] srv._preflight_capability_violation = False srv._preflight_capability_violation_files = [] srv._preflight_reviewer_violation_files = [] srv._preflight_in_test_mode = lambda: False self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False) self._env_patch.start() os.environ.pop("GITEA_ACTIVE_WORKTREE", None) os.environ.pop("GITEA_AUTHOR_WORKTREE", None) def tearDown(self): reviewer_pr_lease.clear_session_lease() 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_whoami_violation_files = self._saved["whoami_files"] srv._preflight_capability_violation = self._saved["capability_violation"] srv._preflight_capability_violation_files = self._saved["capability_files"] srv._preflight_reviewer_violation_files = self._saved["reviewer_files"] srv._preflight_in_test_mode = self._saved["in_test"] self._env_patch.stop() @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_reviewer_mutation_uses_declared_branches_worktree( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock( returncode=0, stdout=f"{CONTROL_ROOT}/.git\n", ) with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): resolved = srv._verify_reviewer_mutation_workspace( "prgs", worktree_path=BRANCHES_WORKTREE ) self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE)) def test_control_checkout_blocks_reviewer_mutations(self): with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with self.assertRaises(RuntimeError) as ctx: srv._verify_reviewer_mutation_workspace("prgs") self.assertIn("stable control checkout", str(ctx.exception)) self.assertIn("GITEA_ACTIVE_WORKTREE", str(ctx.exception)) def test_preflight_and_mutation_agree_on_workspace(self): os.environ["GITEA_ACTIVE_WORKTREE"] = BRANCHES_WORKTREE with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch.object( srv, "_get_git_root", return_value=BRANCHES_WORKTREE, ): status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE) resolved = srv._resolve_reviewer_mutation_worktree_path( worktree_path=BRANCHES_WORKTREE ) self.assertTrue(status["preflight_ready"]) self.assertEqual( status["preflight_workspace"]["active_task_workspace_root"], os.path.realpath(BRANCHES_WORKTREE), ) self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE)) def test_metadata_only_worktree_path_blocks_preflight_ready(self): with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE) self.assertFalse(status["preflight_ready"]) self.assertTrue( any("metadata-only" in reason for reason in status["preflight_block_reasons"]) ) @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_dry_run_inherits_session_lease_worktree( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock( returncode=0, stdout=f"{CONTROL_ROOT}/.git\n", ) reviewer_pr_lease.record_session_lease({ "pr_number": 503, "session_id": "sess-503", "worktree": BRANCHES_WORKTREE, }) with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): with mock.patch.object( srv, "gitea_check_pr_eligibility", return_value={ "eligible": False, "reasons": ["stub"], "authenticated_user": "reviewer", "profile_name": "prgs-reviewer", "pr_author": "author", "head_sha": "abc123", }, ): result = srv.gitea_dry_run_pr_review( pr_number=503, action="comment", remote="prgs", ) self.assertNotIn("stable control checkout", " ".join(result.get("reasons") or [])) self.assertNotIn("metadata-only", " ".join(result.get("reasons") or [])) @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) @mock.patch("os.path.exists", return_value=True) def test_dry_run_fails_closed_from_control_checkout_without_binding( self, _exists, _isdir, mock_run ): mock_run.return_value = MagicMock( returncode=0, stdout=f"{CONTROL_ROOT}/.git\n", ) with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): result = srv.gitea_dry_run_pr_review( pr_number=503, action="comment", remote="prgs", ) reasons = " ".join(result.get("reasons") or []) self.assertIn("stable control checkout", reasons) if __name__ == "__main__": unittest.main()