"""Regression tests for durable author worktree resolution (#618).""" from __future__ import annotations import os import sys import tempfile import unittest from pathlib import Path from unittest import mock from unittest.mock import MagicMock, patch sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import author_mutation_worktree as amw # noqa: E402 import gitea_mcp_server as srv # noqa: E402 import namespace_workspace_binding as nwb # noqa: E402 FAKE_AUTH = {"Authorization": "token test-token"} 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]) class TestDurableAuthorWorktreeResolution(unittest.TestCase): def test_missing_author_env_fails_closed_no_control_fallback(self): missing = "/nonexistent/branches/mcp-author-clean-ns" result = amw.resolve_durable_author_worktree( process_project_root=CONTROL_CHECKOUT_ROOT, author_worktree_env=missing, canonical_repo_root=CONTROL_CHECKOUT_ROOT, ) self.assertTrue(result["block"]) self.assertTrue(result["bound_worktree_missing"]) self.assertFalse(result["silent_control_fallback"]) self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, result["reasons"][0]) self.assertNotEqual( os.path.realpath(result["workspace_path"]), os.path.realpath(CONTROL_CHECKOUT_ROOT), ) def test_missing_active_env_fails_closed(self): missing = "/nonexistent/branches/deleted-active" result = amw.resolve_durable_author_worktree( process_project_root=CONTROL_CHECKOUT_ROOT, active_worktree_env=missing, canonical_repo_root=CONTROL_CHECKOUT_ROOT, ) self.assertTrue(result["block"]) self.assertTrue(result["bound_worktree_missing"]) self.assertIn(amw.ACTIVE_WORKTREE_ENV, result["workspace_binding_source"]) def test_derives_from_active_author_issue_lock(self): lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock") result = amw.resolve_durable_author_worktree( process_project_root=CONTROL_CHECKOUT_ROOT, session_lock_worktree=lock_wt, canonical_repo_root=CONTROL_CHECKOUT_ROOT, validate=False, ) self.assertEqual( result["workspace_path"], os.path.realpath(os.path.abspath(lock_wt)) ) self.assertIn("issue lock", result["workspace_binding_source"]) def test_explicit_worktree_path_wins_over_lock(self): explicit = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-explicit") lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock") result = amw.resolve_durable_author_worktree( worktree_path=explicit, process_project_root=CONTROL_CHECKOUT_ROOT, session_lock_worktree=lock_wt, canonical_repo_root=CONTROL_CHECKOUT_ROOT, validate=False, ) self.assertEqual( result["workspace_path"], os.path.realpath(os.path.abspath(explicit)) ) self.assertEqual(result["workspace_binding_source"], "worktree_path argument") def test_no_binding_does_not_silently_use_control_checkout(self): result = amw.resolve_durable_author_worktree( process_project_root=CONTROL_CHECKOUT_ROOT, canonical_repo_root=CONTROL_CHECKOUT_ROOT, ) self.assertTrue(result["block"]) self.assertFalse(result["silent_control_fallback"]) blob = " ".join(result["reasons"]) self.assertIn("control checkout", blob) self.assertIn("forbidden", blob) def test_process_root_under_branches_is_allowed(self): branches_root = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "session-wt") result = amw.resolve_durable_author_worktree( process_project_root=branches_root, canonical_repo_root=CONTROL_CHECKOUT_ROOT, validate=False, ) self.assertFalse(result["block"]) self.assertEqual( result["workspace_path"], os.path.realpath(branches_root) ) self.assertIn("branches/", result["workspace_binding_source"]) def test_lock_ownership_mismatch_fails_closed(self): with tempfile.TemporaryDirectory() as tmp: root = tmp branches = os.path.join(root, "branches") os.makedirs(os.path.join(branches, "a")) os.makedirs(os.path.join(branches, "b")) # Seed a fake .git so membership/list may soft-fail without hard error os.makedirs(os.path.join(root, ".git")) result = amw.resolve_durable_author_worktree( worktree_path=os.path.join(branches, "a"), process_project_root=root, session_lock_worktree=os.path.join(branches, "b"), canonical_repo_root=root, validate=True, ) self.assertTrue(result["block"]) self.assertTrue( any("lock" in r.lower() and "match" in r.lower() for r in result["reasons"]) ) def test_traversal_safety_blocks_escape(self): assessment = amw.assess_path_traversal_safety( path="/tmp/other-repo/branches/evil", canonical_repo_root=CONTROL_CHECKOUT_ROOT, ) self.assertTrue(assessment["block"]) self.assertTrue(any("escapes" in r for r in assessment["reasons"])) def test_bound_worktree_existence_reports_null_git_root(self): assessment = amw.assess_bound_worktree_existence( configured_path="/nonexistent/branches/gone", binding_source=f"{amw.AUTHOR_WORKTREE_ENV} environment variable", canonical_repo_root=CONTROL_CHECKOUT_ROOT, profile_name="prgs-author", ) self.assertTrue(assessment["block"]) self.assertIsNone(assessment["inspected_git_root"]) self.assertFalse(assessment["path_exists"]) msg = amw.format_bound_worktree_missing_error(assessment) self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, msg) self.assertIn("prgs-author", msg) self.assertIn("recreate or repoint", msg.lower()) class TestNamespaceAuthorNoDemotion(unittest.TestCase): def test_author_missing_env_not_demoted_to_process_root(self): missing = "/nonexistent/branches/mcp-author-clean-ns" demotions: list[str] = [] path, source = nwb.resolve_namespace_workspace( role_kind="author", process_project_root=CONTROL_CHECKOUT_ROOT, env={amw.AUTHOR_WORKTREE_ENV: missing}, demotions=demotions, verify_paths=True, ) self.assertIn("AUTHOR", source) self.assertNotEqual(os.path.realpath(path), os.path.realpath(CONTROL_CHECKOUT_ROOT)) self.assertTrue(any("not demoted" in d for d in demotions)) def test_reviewer_still_demotes_missing_env(self): """#702 demotion retained for non-author roles.""" demotions: list[str] = [] path, source = nwb.resolve_namespace_workspace( role_kind="reviewer", process_project_root=CONTROL_CHECKOUT_ROOT, env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-gone"}, demotions=demotions, verify_paths=True, ) self.assertEqual(source, "MCP server process root (default)") self.assertEqual(path, os.path.realpath(CONTROL_CHECKOUT_ROOT)) self.assertTrue(demotions) def test_mutation_context_surfaces_missing_binding_health(self): ctx = nwb.resolve_namespace_mutation_context( role_kind="author", worktree_path=None, process_project_root=CONTROL_CHECKOUT_ROOT, env={amw.AUTHOR_WORKTREE_ENV: "/nonexistent/branches/mcp-author-clean-ns"}, profile_name="prgs-author", ) self.assertTrue(ctx.get("bound_worktree_missing")) self.assertTrue(ctx.get("author_worktree_block")) self.assertIsNone(ctx.get("inspected_git_root")) self.assertFalse(ctx.get("path_exists")) class TestCreateIssueAndCommentAgreeOnMissingWorktree(unittest.TestCase): """AC3/AC4: create_issue and create_issue_comment enforce the same rule.""" def setUp(self): srv._preflight_whoami_called = True srv._preflight_capability_called = True srv._preflight_resolved_role = "author" srv._preflight_resolved_task = None srv._preflight_whoami_violation = False srv._preflight_capability_violation = False self._orig_in_test = srv._preflight_in_test_mode srv._preflight_in_test_mode = lambda: False self._lock_patch = patch( "gitea_mcp_server._session_author_lock_worktree", return_value=None ) self._lock_patch.start() self.addCleanup(self._restore) def _restore(self): srv._preflight_in_test_mode = self._orig_in_test srv._preflight_resolved_task = None self._lock_patch.stop() os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None) os.environ.pop(amw.ACTIVE_WORKTREE_ENV, None) def _assert_blocked_missing(self, result_or_exc): if isinstance(result_or_exc, BaseException): blob = str(result_or_exc) else: blob = " ".join( str(x) for x in ( result_or_exc.get("reasons") or [], result_or_exc.get("message"), result_or_exc.get("blocker_kind"), ) if x ) if not blob: blob = str(result_or_exc) self.assertTrue( amw.BOUND_WORKTREE_MISSING_MESSAGE in blob or "does not exist" in blob or "bound worktree" in blob.lower(), msg=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) @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=[]) def test_create_issue_blocked_when_author_env_missing( self, _get_all, mock_api, _role, _ns, _prof, _auth ): missing = os.path.join( CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env" ) os.environ[amw.AUTHOR_WORKTREE_ENV] = missing srv._preflight_resolved_task = "create_issue" with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): try: res = srv.gitea_create_issue(title="Test issue", body="body text here") except RuntimeError as exc: self._assert_blocked_missing(exc) else: self.assertFalse(res.get("success", True) and res.get("number")) self._assert_blocked_missing(res) mock_api.assert_not_called() @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server.api_request") def test_create_issue_comment_blocked_when_author_env_missing(self, mock_api, _auth): missing = os.path.join( CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env" ) os.environ[amw.AUTHOR_WORKTREE_ENV] = missing srv._preflight_resolved_task = "comment_issue" author_env = { "GITEA_PROFILE_NAME": "gitea-author", "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment", amw.AUTHOR_WORKTREE_ENV: missing, } with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): with patch.dict(os.environ, author_env, clear=False): try: res = srv.gitea_create_issue_comment( issue_number=618, body="evidence comment", remote="prgs", ) except RuntimeError as exc: self._assert_blocked_missing(exc) else: self.assertFalse(res.get("success", True)) self._assert_blocked_missing(res) mock_api.assert_not_called() class TestRuntimeContextUnhealthyMissingWorktree(unittest.TestCase): def setUp(self): 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 self._lock_patch = patch( "gitea_mcp_server._session_author_lock_worktree", return_value=None ) self._lock_patch.start() def tearDown(self): self._lock_patch.stop() os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None) def test_assess_preflight_reports_null_git_root_and_missing(self): missing = "/nonexistent/branches/mcp-author-clean-ns" os.environ[amw.AUTHOR_WORKTREE_ENV] = missing with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): with patch("gitea_mcp_server.get_profile", return_value={ "profile_name": "prgs-author", "allowed_operations": ["gitea.pr.create"], "forbidden_operations": [], }): status = srv.assess_preflight_status() self.assertFalse(status["preflight_ready"]) blob = " ".join(status["preflight_block_reasons"]) self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, blob) details = status["preflight_workspace"] self.assertIsNotNone(details) self.assertTrue(details.get("bound_worktree_missing")) self.assertIsNone(details.get("inspected_git_root")) self.assertFalse(details.get("path_exists")) self.assertFalse(details.get("workspace_healthy")) class TestThreadLedgerExample(unittest.TestCase): def test_bound_worktree_missing_ledger_example_exists(self): import thread_state_ledger_examples as examples names = [name for name, _h, _l in examples.EXAMPLES] self.assertIn("bound_worktree_missing_blocker", names) for name, _handoff, ledger in examples.EXAMPLES: if name == "bound_worktree_missing_blocker": self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, ledger) self.assertIn("inspected_git_root", ledger) self.assertIn("operator", ledger.lower()) break if __name__ == "__main__": unittest.main()