jcwalker3andClaude Opus 4.8
5ed2ab8a38
Author mutation tools now resolve workspace via explicit worktree_path,
env bindings, or the active author issue lock — never silent fallback to
the control checkout/master. Missing configured bindings fail closed with
operator recovery; create_issue and create_issue_comment agree.
Recovered onto 0568f44 from preserved candidate cbf56ccd (AUTHOR_RECOVERY).
LLM_LOCK_ID=author-618-recovery-508eb3162d01-1784569919
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
350 lines
15 KiB
Python
350 lines
15 KiB
Python
"""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()
|