Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
72b18143f8 | ||
|
|
8886ce201f | ||
|
|
bee24e2b10 | ||
|
|
a7aac0df1d | ||
|
|
ec903b0d61 |
@@ -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:
|
||||||
|
|||||||
@@ -29,6 +29,11 @@ class UnsanctionedRuntimeError(RuntimeError):
|
|||||||
|
|
||||||
|
|
||||||
def is_pytest_runtime() -> bool:
|
def is_pytest_runtime() -> bool:
|
||||||
|
if os.environ.get("GITEA_TEST_FORCE_UNSANCTIONED") == "1":
|
||||||
|
return False
|
||||||
|
import sys
|
||||||
|
if "pytest" in sys.modules:
|
||||||
|
return True
|
||||||
return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())
|
return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+2
-1
@@ -6,7 +6,8 @@ Runs over stdio. All tools authenticate via macOS keychain (git credential fill)
|
|||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
sys.stderr = open("/tmp/mcp_server_stderr.log", "a", buffering=1)
|
if "PYTEST_CURRENT_TEST" not in os.environ:
|
||||||
|
sys.stderr = open("/tmp/mcp_server_stderr.log", "a", buffering=1)
|
||||||
sys.stderr.write(f"\n--- MCP SERVER STARTUP (PID {os.getpid()}) ---\n")
|
sys.stderr.write(f"\n--- MCP SERVER STARTUP (PID {os.getpid()}) ---\n")
|
||||||
|
|
||||||
from role_session_router import (
|
from role_session_router import (
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|||||||
@@ -82,13 +82,14 @@ class TestPreflightIntegration(unittest.TestCase):
|
|||||||
control_root = "/repo/Gitea-Tools"
|
control_root = "/repo/Gitea-Tools"
|
||||||
with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root):
|
with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root):
|
||||||
with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"):
|
with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"):
|
||||||
with mock.patch.dict(
|
with mock.patch("gitea_auth.get_profile", return_value={"profile_name": "gitea-author"}):
|
||||||
"os.environ",
|
with mock.patch.dict(
|
||||||
{"GITEA_TEST_PORCELAIN": ""},
|
"os.environ",
|
||||||
clear=False,
|
{"GITEA_TEST_PORCELAIN": ""},
|
||||||
):
|
clear=False,
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
):
|
||||||
mcp_server.verify_preflight_purity()
|
with self.assertRaises(RuntimeError) as ctx:
|
||||||
|
mcp_server.verify_preflight_purity()
|
||||||
self.assertIn("Branches-only mutation guard", str(ctx.exception))
|
self.assertIn("Branches-only mutation guard", str(ctx.exception))
|
||||||
|
|
||||||
def test_verify_preflight_allows_branches_worktree(self):
|
def test_verify_preflight_allows_branches_worktree(self):
|
||||||
|
|||||||
@@ -11,7 +11,11 @@ import gitea_mcp_server as srv
|
|||||||
|
|
||||||
FAKE_AUTH = {"Authorization": "token test-token"}
|
FAKE_AUTH = {"Authorization": "token test-token"}
|
||||||
# Stable control checkout (parent of branches/), not the MCP server worktree root.
|
# Stable control checkout (parent of branches/), not the MCP server worktree root.
|
||||||
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
|
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])
|
||||||
PROJECT_ROOT = srv.PROJECT_ROOT
|
PROJECT_ROOT = srv.PROJECT_ROOT
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -25,7 +25,11 @@ import gitea_mcp_server as srv # noqa: E402
|
|||||||
import root_checkout_guard as rcg # noqa: E402
|
import root_checkout_guard as rcg # noqa: E402
|
||||||
|
|
||||||
FAKE_AUTH = "token test"
|
FAKE_AUTH = "token test"
|
||||||
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
|
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])
|
||||||
MASTER_SHA = "a" * 40
|
MASTER_SHA = "a" * 40
|
||||||
OTHER_SHA = "b" * 40
|
OTHER_SHA = "b" * 40
|
||||||
|
|
||||||
|
|||||||
@@ -10,7 +10,11 @@ import gitea_mcp_server as srv
|
|||||||
|
|
||||||
|
|
||||||
FAKE_AUTH = {"Authorization": "token test-token"}
|
FAKE_AUTH = {"Authorization": "token test-token"}
|
||||||
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
|
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 TestIssueCommentWorkspaceGuard(unittest.TestCase):
|
class TestIssueCommentWorkspaceGuard(unittest.TestCase):
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ class TestMcpDaemonGuard(unittest.TestCase):
|
|||||||
mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV,
|
mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV,
|
||||||
"PYTEST_CURRENT_TEST",
|
"PYTEST_CURRENT_TEST",
|
||||||
}}
|
}}
|
||||||
|
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
||||||
with patch.dict(os.environ, env, clear=True):
|
with patch.dict(os.environ, env, clear=True):
|
||||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("test")
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("test")
|
||||||
@@ -43,6 +44,7 @@ class TestMcpDaemonGuard(unittest.TestCase):
|
|||||||
"PYTEST_CURRENT_TEST",
|
"PYTEST_CURRENT_TEST",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
||||||
with patch.dict(os.environ, env, clear=True):
|
with patch.dict(os.environ, env, clear=True):
|
||||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||||
mcp_daemon_guard.assert_keychain_access_allowed()
|
mcp_daemon_guard.assert_keychain_access_allowed()
|
||||||
@@ -58,6 +60,7 @@ class TestMcpDaemonGuard(unittest.TestCase):
|
|||||||
"PYTEST_CURRENT_TEST",
|
"PYTEST_CURRENT_TEST",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
||||||
with patch.dict(os.environ, env, clear=True):
|
with patch.dict(os.environ, env, clear=True):
|
||||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||||
gitea_auth.get_auth_header("gitea.prgs.cc")
|
gitea_auth.get_auth_header("gitea.prgs.cc")
|
||||||
|
|||||||
@@ -48,6 +48,22 @@ import gitea_config # noqa: E402
|
|||||||
|
|
||||||
import mcp_server
|
import mcp_server
|
||||||
import issue_lock_store
|
import issue_lock_store
|
||||||
|
import gitea_auth
|
||||||
|
|
||||||
|
_orig_api_request = gitea_auth.api_request
|
||||||
|
def mockable_api_request(*args, **kwargs):
|
||||||
|
import mcp_server
|
||||||
|
return mcp_server.api_request(*args, **kwargs)
|
||||||
|
|
||||||
|
def setUpModule():
|
||||||
|
gitea_auth.api_request = mockable_api_request
|
||||||
|
patch("mcp_server._enforce_root_checkout_guard").start()
|
||||||
|
patch("mcp_server._enforce_branches_only_author_mutation").start()
|
||||||
|
|
||||||
|
def tearDownModule():
|
||||||
|
gitea_auth.api_request = _orig_api_request
|
||||||
|
patch.stopall()
|
||||||
|
|
||||||
|
|
||||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||||
FULL_HEAD_SHA = "a" * 40
|
FULL_HEAD_SHA = "a" * 40
|
||||||
@@ -1364,11 +1380,11 @@ class TestReviewPR(unittest.TestCase):
|
|||||||
self.assertIn("Review/Merge Blocked", result["message"])
|
self.assertIn("Review/Merge Blocked", result["message"])
|
||||||
self.assertIn("Author profile", result["message"])
|
self.assertIn("Author profile", result["message"])
|
||||||
|
|
||||||
@patch("mcp_server.api_get_all")
|
@patch("mcp_server.api_fetch_page")
|
||||||
@patch("mcp_server.api_request")
|
@patch("mcp_server.api_request")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@patch("mcp_server.get_profile")
|
@patch("mcp_server.get_profile")
|
||||||
def test_legacy_review_pr_self_approval_blocked(self, mock_get_profile, _auth, mock_api, mock_get_all):
|
def test_legacy_review_pr_self_approval_blocked(self, mock_get_profile, _auth, mock_api, mock_fetch):
|
||||||
mock_get_profile.return_value = {
|
mock_get_profile.return_value = {
|
||||||
"profile_name": "gitea-reviewer",
|
"profile_name": "gitea-reviewer",
|
||||||
"allowed_operations": ["read", "approve"],
|
"allowed_operations": ["read", "approve"],
|
||||||
@@ -1376,7 +1392,10 @@ class TestReviewPR(unittest.TestCase):
|
|||||||
"base_url": None,
|
"base_url": None,
|
||||||
}
|
}
|
||||||
head_sha = FULL_HEAD_SHA
|
head_sha = FULL_HEAD_SHA
|
||||||
mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
|
mock_fetch.return_value = (
|
||||||
|
[{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}],
|
||||||
|
{"page": 1, "per_page": 50, "returned_count": 1, "has_more": False, "next_page": None, "is_final_page": True}
|
||||||
|
)
|
||||||
# mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
|
# mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
|
||||||
mock_api.side_effect = [
|
mock_api.side_effect = [
|
||||||
{"login": "jcwalker3"}, # /api/v1/user (inventory)
|
{"login": "jcwalker3"}, # /api/v1/user (inventory)
|
||||||
|
|||||||
@@ -10,8 +10,11 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|||||||
import gitea_mcp_server as srv
|
import gitea_mcp_server as srv
|
||||||
|
|
||||||
FAKE_AUTH = "token test"
|
FAKE_AUTH = "token test"
|
||||||
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
|
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])
|
||||||
RECONCILER_PROFILE = {
|
RECONCILER_PROFILE = {
|
||||||
"profile_name": "prgs-reconciler",
|
"profile_name": "prgs-reconciler",
|
||||||
"allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"],
|
"allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"],
|
||||||
|
|||||||
@@ -13,8 +13,13 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|||||||
import gitea_mcp_server as srv # noqa: E402
|
import gitea_mcp_server as srv # noqa: E402
|
||||||
import root_checkout_guard as rcg # noqa: E402
|
import root_checkout_guard as rcg # noqa: E402
|
||||||
|
|
||||||
CONTROL_ROOT = str(Path(__file__).resolve().parents[3])
|
current_file_path = Path(__file__).resolve()
|
||||||
BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1])
|
if "branches" in current_file_path.parts:
|
||||||
|
CONTROL_ROOT = str(current_file_path.parents[3])
|
||||||
|
BRANCHES_WORKTREE = str(current_file_path.parents[1])
|
||||||
|
else:
|
||||||
|
CONTROL_ROOT = str(current_file_path.parents[1])
|
||||||
|
BRANCHES_WORKTREE = str(current_file_path.parents[1] / "branches" / "mock-worktree")
|
||||||
MASTER_SHA = "a" * 40
|
MASTER_SHA = "a" * 40
|
||||||
OTHER_SHA = "b" * 40
|
OTHER_SHA = "b" * 40
|
||||||
|
|
||||||
@@ -136,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,
|
||||||
@@ -156,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()
|
||||||
@@ -14,8 +14,13 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|||||||
import author_mutation_worktree as amw # noqa: E402
|
import author_mutation_worktree as amw # noqa: E402
|
||||||
import gitea_mcp_server as srv # noqa: E402
|
import gitea_mcp_server as srv # noqa: E402
|
||||||
|
|
||||||
CONTROL_ROOT = str(Path(__file__).resolve().parents[3])
|
current_file_path = Path(__file__).resolve()
|
||||||
BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1])
|
if "branches" in current_file_path.parts:
|
||||||
|
CONTROL_ROOT = str(current_file_path.parents[3])
|
||||||
|
BRANCHES_WORKTREE = str(current_file_path.parents[1])
|
||||||
|
else:
|
||||||
|
CONTROL_ROOT = str(current_file_path.parents[1])
|
||||||
|
BRANCHES_WORKTREE = str(current_file_path.parents[1] / "branches" / "mock-worktree")
|
||||||
MCP_PROCESS_ROOT = BRANCHES_WORKTREE
|
MCP_PROCESS_ROOT = BRANCHES_WORKTREE
|
||||||
|
|
||||||
|
|
||||||
@@ -95,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",
|
||||||
|
|||||||
@@ -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