Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
72b18143f8 | ||
|
|
8886ce201f |
@@ -724,6 +724,7 @@ def _verify_role_mutation_workspace(
|
|||||||
task: str | None = None,
|
task: str | None = None,
|
||||||
) -> str:
|
) -> str:
|
||||||
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
|
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
|
||||||
|
|
||||||
# Check running runtimes to prevent stale mutations
|
# Check running runtimes to prevent stale mutations
|
||||||
try:
|
try:
|
||||||
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
|
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
|
||||||
|
|||||||
@@ -27,6 +27,20 @@ _READONLY_REVIEWER_GIT = re.compile(
|
|||||||
_GIT_INVOCATION = re.compile(r"\bgit\b", re.IGNORECASE)
|
_GIT_INVOCATION = re.compile(r"\bgit\b", re.IGNORECASE)
|
||||||
|
|
||||||
|
|
||||||
|
# #673: Canonical pattern for identifying review worktrees under branches/.
|
||||||
|
REVIEW_WORKTREE_RE = re.compile(
|
||||||
|
r"branches/(?:review-pr\d+[\w/-]*|merge-simulation-pr\d+|review-[\w-]+)",
|
||||||
|
re.IGNORECASE,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def is_review_worktree_path(path: str) -> bool:
|
||||||
|
"""True when the path belongs to a reviewer or simulation worktree."""
|
||||||
|
normalized = (path or "").replace("\\", "/")
|
||||||
|
return bool(REVIEW_WORKTREE_RE.search(normalized))
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def parse_dirty_tracked_files(porcelain: str) -> list[str]:
|
def parse_dirty_tracked_files(porcelain: str) -> list[str]:
|
||||||
"""Return tracked paths with local modifications from ``git status --porcelain``.
|
"""Return tracked paths with local modifications from ``git status --porcelain``.
|
||||||
|
|
||||||
|
|||||||
@@ -5,10 +5,9 @@ from __future__ import annotations
|
|||||||
import re
|
import re
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
_SESSION_OWNED_RE = re.compile(
|
from reviewer_worktree import REVIEW_WORKTREE_RE
|
||||||
r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
|
|
||||||
re.IGNORECASE,
|
_SESSION_OWNED_RE = REVIEW_WORKTREE_RE
|
||||||
)
|
|
||||||
_WORKTREE_PATH_RE = re.compile(
|
_WORKTREE_PATH_RE = re.compile(
|
||||||
r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)",
|
r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)",
|
||||||
re.IGNORECASE,
|
re.IGNORECASE,
|
||||||
|
|||||||
+10
-1
@@ -26,7 +26,16 @@ def _reset_mutation_authority(monkeypatch):
|
|||||||
Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
|
Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
|
||||||
so durable load/save never touches host state even after env clears.
|
so durable load/save never touches host state even after env clears.
|
||||||
"""
|
"""
|
||||||
monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
|
for env_key in [
|
||||||
|
"GITEA_SESSION_PROFILE_LOCK",
|
||||||
|
"GITEA_ACTIVE_WORKTREE",
|
||||||
|
"GITEA_AUTHOR_WORKTREE",
|
||||||
|
"GITEA_REVIEWER_WORKTREE",
|
||||||
|
"GITEA_MERGER_WORKTREE",
|
||||||
|
"GITEA_RECONCILER_WORKTREE",
|
||||||
|
]:
|
||||||
|
monkeypatch.delenv(env_key, raising=False)
|
||||||
|
|
||||||
# Isolate durable session-state files so tests never share host cache (#559).
|
# Isolate durable session-state files so tests never share host cache (#559).
|
||||||
import tempfile
|
import tempfile
|
||||||
|
|
||||||
|
|||||||
@@ -141,13 +141,21 @@ class TestVerifyPreflightRootGuardIntegration(unittest.TestCase):
|
|||||||
self.assertIn("Root checkout guard (#475)", str(ctx.exception))
|
self.assertIn("Root checkout guard (#475)", str(ctx.exception))
|
||||||
self.assertIn(rcg.REMEDIATION, str(ctx.exception))
|
self.assertIn(rcg.REMEDIATION, str(ctx.exception))
|
||||||
|
|
||||||
|
@patch("os.path.isdir", return_value=True)
|
||||||
|
@patch("os.path.exists", return_value=True)
|
||||||
|
@patch("subprocess.run")
|
||||||
@patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
|
@patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
|
||||||
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
|
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
|
||||||
@patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
|
@patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
|
||||||
@patch("gitea_mcp_server._resolve_author_mutation_context")
|
@patch("gitea_mcp_server._resolve_author_mutation_context")
|
||||||
def test_reviewer_from_branches_worktree_allowed(
|
def test_reviewer_from_branches_worktree_allowed(
|
||||||
self, mock_ctx, mock_git, _remote_sha, _porcelain,
|
self, mock_ctx, mock_git, _remote_sha, _porcelain, mock_run, _exists, _isdir,
|
||||||
):
|
):
|
||||||
|
import unittest.mock
|
||||||
|
mock_run.return_value = unittest.mock.MagicMock(
|
||||||
|
returncode=0,
|
||||||
|
stdout=f"{CONTROL_ROOT}/.git\n",
|
||||||
|
)
|
||||||
srv._preflight_capability_baseline_porcelain = ""
|
srv._preflight_capability_baseline_porcelain = ""
|
||||||
mock_ctx.return_value = {
|
mock_ctx.return_value = {
|
||||||
"workspace_path": BRANCHES_WORKTREE,
|
"workspace_path": BRANCHES_WORKTREE,
|
||||||
@@ -161,6 +169,5 @@ class TestVerifyPreflightRootGuardIntegration(unittest.TestCase):
|
|||||||
}
|
}
|
||||||
srv.verify_preflight_purity("prgs", worktree_path=BRANCHES_WORKTREE)
|
srv.verify_preflight_purity("prgs", worktree_path=BRANCHES_WORKTREE)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
@@ -100,7 +100,23 @@ class TestRuntimeContextGuardAlignment(unittest.TestCase):
|
|||||||
srv._preflight_in_test_mode = self._orig_in_test
|
srv._preflight_in_test_mode = self._orig_in_test
|
||||||
self._env_patch.stop()
|
self._env_patch.stop()
|
||||||
|
|
||||||
def test_runtime_context_and_guard_share_resolved_workspace(self):
|
@mock.patch("subprocess.run")
|
||||||
|
@mock.patch("os.path.isdir", return_value=True)
|
||||||
|
@mock.patch("os.path.exists", return_value=True)
|
||||||
|
def test_runtime_context_and_guard_share_resolved_workspace(
|
||||||
|
self, _exists, _isdir, mock_run
|
||||||
|
):
|
||||||
|
def run_side_effect(cmd, *args, **kwargs):
|
||||||
|
res = MagicMock(returncode=0)
|
||||||
|
if "--git-common-dir" in cmd:
|
||||||
|
res.stdout = f"{CONTROL_ROOT}/.git\n"
|
||||||
|
elif "--show-toplevel" in cmd:
|
||||||
|
cwd = cmd[cmd.index("-C") + 1] if "-C" in cmd else ""
|
||||||
|
res.stdout = f"{cwd}\n"
|
||||||
|
else:
|
||||||
|
res.stdout = f"{CONTROL_ROOT}\n"
|
||||||
|
return res
|
||||||
|
mock_run.side_effect = run_side_effect
|
||||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||||
ctx = srv._resolve_author_mutation_context(BRANCHES_WORKTREE)
|
ctx = srv._resolve_author_mutation_context(BRANCHES_WORKTREE)
|
||||||
status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
|
status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
|
||||||
|
|||||||
@@ -14,11 +14,7 @@ from merged_cleanup_reconcile import (
|
|||||||
read_issue_lock,
|
read_issue_lock,
|
||||||
read_local_worktree_state,
|
read_local_worktree_state,
|
||||||
)
|
)
|
||||||
|
from reviewer_worktree import REVIEW_WORKTREE_RE
|
||||||
_REVIEW_WORKTREE_RE = re.compile(
|
|
||||||
r"branches/(?:review-pr\d+|merge-simulation-pr\d+|review-[\w-]+)",
|
|
||||||
re.IGNORECASE,
|
|
||||||
)
|
|
||||||
|
|
||||||
CLASSIFICATIONS = frozenset({
|
CLASSIFICATIONS = frozenset({
|
||||||
"active-pr",
|
"active-pr",
|
||||||
@@ -147,7 +143,7 @@ def classify_entry(
|
|||||||
if not record:
|
if not record:
|
||||||
return "orphan", "Directory exists but is not a registered git worktree"
|
return "orphan", "Directory exists but is not a registered git worktree"
|
||||||
|
|
||||||
if record.get("detached") and _REVIEW_WORKTREE_RE.search(rel_path):
|
if record.get("detached") and REVIEW_WORKTREE_RE.search(rel_path):
|
||||||
return "detached-review", "Detached HEAD in reviewer/simulation worktree"
|
return "detached-review", "Detached HEAD in reviewer/simulation worktree"
|
||||||
|
|
||||||
if state.get("exists") and state.get("clean"):
|
if state.get("exists") and state.get("clean"):
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ from datetime import datetime, timezone
|
|||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from merged_cleanup_reconcile import branch_worktree_folder, read_local_worktree_state
|
from merged_cleanup_reconcile import branch_worktree_folder, read_local_worktree_state
|
||||||
from reviewer_worktree import parse_dirty_tracked_files
|
from reviewer_worktree import parse_dirty_tracked_files, REVIEW_WORKTREE_RE
|
||||||
|
|
||||||
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
|
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
|
||||||
DEFAULT_TTL_HOURS = float(os.environ.get("GITEA_WORKTREE_TTL_HOURS", "24") or 24)
|
DEFAULT_TTL_HOURS = float(os.environ.get("GITEA_WORKTREE_TTL_HOURS", "24") or 24)
|
||||||
@@ -508,10 +508,8 @@ DISPOSITIONS = frozenset({
|
|||||||
"unsafe_unknown",
|
"unsafe_unknown",
|
||||||
})
|
})
|
||||||
|
|
||||||
REVIEW_WORKTREE_RE = re.compile(
|
|
||||||
r"branches/(?:review-pr\d+|merge-simulation-pr\d+|review-[\w-]+)",
|
|
||||||
re.IGNORECASE,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def normalize_path(path: str) -> str:
|
def normalize_path(path: str) -> str:
|
||||||
|
|||||||
Reference in New Issue
Block a user