Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 101 Pull Requests 11 Actions Packages Projects Releases Wiki Activity

Fix review 646 blockers B1-B7 for cross-repo canonical roots (#973)

Browse Source
This commit is contained in:
sysadmin
2026-07-29 14:01:17 -04:00
parent 6a53308473
commit 7978008709
4 changed files with 380 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+51 -11
View File
@@ -500,10 +500,38 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
return workspace
def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
def _resolve_expected_repository_slug(
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> str | None:
"""Resolve expected repository slug from parameters, session context, or remote URL."""
if org and repo:
return session_ctx.format_repository_slug(org, repo)
bound = session_ctx.get_session_context() or {}
b_org = bound.get("org")
b_repo = bound.get("repository")
if b_org and b_repo:
return session_ctx.format_repository_slug(b_org, b_repo)
eff_remote = remote or bound.get("remote") or _effective_remote()
parsed = remote_repo_guard.parse_org_repo_from_remote_url(
_local_git_remote_url(eff_remote)
)
if parsed:
return session_ctx.format_repository_slug(parsed[0], parsed[1])
return None
def _resolve_namespace_mutation_context(
worktree_path: str | None = None,
remote: str | None = None,
) -> dict:
"""Canonical namespace workspace + repository root for guards (#460/#510/#706/#618)."""
role = _effective_workspace_role()
configured_root, _source = _configured_canonical_root()
bound = session_ctx.get_session_context() or {}
eff_remote = remote or bound.get("remote") or _effective_remote()
expected_slug = _resolve_expected_repository_slug(eff_remote)
return nwb.resolve_namespace_mutation_context(
role_kind=role,
worktree_path=worktree_path,
@@ -516,9 +544,12 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic
),
profile_name=get_profile().get("profile_name"),
configured_canonical_root=configured_root,
expected_slug=expected_slug,
remote=eff_remote,
)
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
"""Backward-compatible alias for namespace workspace context."""
return _resolve_namespace_mutation_context(worktree_path)
@@ -620,6 +651,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
inspected_root = _get_git_root(workspace)
process_root = ctx["process_project_root"]
canonical_root = ctx["canonical_repo_root"]
crr_assessment = ctx.get("canonical_root_assessment") or {}
resolved_slug = crr_assessment.get("resolved_slug")
expected_slug = ctx.get("expected_slug")
active_root = os.path.realpath(inspected_root or workspace)
if active_root == canonical_root:
dirty_scope = "control checkout"
@@ -649,6 +683,10 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
"workspace_healthy": not bool(
ctx.get("bound_worktree_missing") or ctx.get("author_worktree_block")
),
"canonical_root_assessment": crr_assessment,
"expected_repository_slug": expected_slug,
"observed_repository_identity": resolved_slug,
"worktree_registration_result": ctx.get("in_git_worktree_list"),
}
if not ctx["roots_aligned"]:
details["workspace_root_mismatch"] = (
@@ -658,6 +696,7 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
return details
def _format_preflight_workspace_details(details: dict) -> str:
parts = [
f"MCP server process root: {details.get('mcp_server_process_root')}",
@@ -1866,6 +1905,9 @@ def _verify_role_mutation_workspace(
# back to the install checkout and validated Gitea-Tools/branches/ instead
# of the target repository the namespace is actually bound to.
_configured_root, _configured_source = _configured_canonical_root()
bound = session_ctx.get_session_context() or {}
eff_remote = remote or bound.get("remote") or _effective_remote()
expected_slug = _resolve_expected_repository_slug(eff_remote, org=org, repo=repo)
assessment = nwb.assess_namespace_mutation_workspace(
role_kind=role,
worktree_path=worktree_path,
@@ -1880,6 +1922,8 @@ def _verify_role_mutation_workspace(
profile_name=get_profile().get("profile_name"),
current_branch=git_state.get("current_branch"),
configured_canonical_root=_configured_root,
expected_slug=expected_slug,
remote=eff_remote,
)
if assessment["block"]:
raise RuntimeError(
@@ -3414,7 +3458,7 @@ def cleanup_in_progress_for_pr(
# ── Helpers ───────────────────────────────────────────────────────────────────
def _effective_remote(remote: str) -> str:
def _effective_remote(remote: str = "dadeschools") -> str:
"""If remote is the default ('dadeschools') but the active profile base_url maps to a known remote, use that remote instead."""
try:
profile = get_profile()
@@ -15277,22 +15321,17 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
workspace_root = None
aligned = None
canonical_root = None
resolved_slug = None
try:
ctx = _resolve_namespace_mutation_context(None)
workspace_root = ctx.get("workspace_path")
canonical_root = ctx.get("canonical_repo_root")
# Alignment keeps its established repository-level meaning (#615 F1):
# does this namespace target the repository the process is installed in,
# i.e. canonical_repo_root == process_project_root. It is deliberately
# NOT path equality between the task workspace and the process root --
# the global worktree rule requires task work to live in a branches/
# worktree, so that comparison would classify every correctly bound
# author, reviewer, and merger session as unsafe.
aligned = ctx.get("roots_aligned")
crr_assessment = ctx.get("canonical_root_assessment") or {}
resolved_slug = crr_assessment.get("resolved_slug")
except Exception:
# An unresolvable binding is reported as unknown alignment, never as
# proof of alignment.
aligned = None
resolved_slug = None
try:
profile_name = get_profile()["profile_name"]
except Exception:
@@ -15305,6 +15344,7 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
dirty_files=dirty_files,
active_task_workspace=workspace_root,
canonical_repository_root=canonical_root,
repository_slug=resolved_slug,
workspace_roots_aligned=aligned,
profile=profile_name,
declared_mode=stable_control_runtime.declared_runtime_mode(),
namespace_workspace_binding.py
+54 -40
View File
@@ -189,23 +189,23 @@ def verify_git_common_directory_membership(
real_common = os.path.realpath(common_dir)
canonical_git = os.path.realpath(os.path.join(real_root, ".git"))
if real_common in (canonical_git, real_root) or os.path.dirname(real_common) == real_root:
if real_common in (canonical_git, real_root):
return True, None
return (
False,
f"workspace '{real_ws}' git common directory '{real_common}' does not match "
f"canonical repository root '{real_root}' (.git at '{canonical_git}')"
)
except Exception:
pass
if amw.is_path_under_branches(real_ws, real_root):
return True, None
return (
False,
f"workspace '{real_ws}' does not belong to canonical repository root '{real_root}'"
)
else:
return (
False,
f"workspace '{real_ws}' is not a valid git repository or git rev-parse failed"
)
except Exception as exc:
return (
False,
f"failed to inspect git common directory for workspace '{real_ws}': {exc}"
)
def resolve_namespace_mutation_context(
@@ -313,6 +313,8 @@ def resolve_namespace_mutation_context(
"canonical_repo_root": canonical_root,
"roots_aligned": roots_aligned,
"canonical_root_assessment": crr_assessment,
"expected_slug": expected_slug,
"remote": remote,
}
if durable is not None:
result["author_worktree_resolution"] = durable
@@ -324,6 +326,15 @@ def resolve_namespace_mutation_context(
result["author_worktree_reasons"] = list(durable.get("reasons") or [])
result["author_worktree_blocker_kind"] = durable.get("blocker_kind")
result["operator_recovery"] = durable.get("operator_recovery")
else:
path_exists = os.path.exists(workspace)
result["path_exists"] = path_exists
result["in_git_worktree_list"] = (
amw.path_in_git_worktree_list(workspace, canonical_root)
if path_exists
else False
)
result["bound_worktree_missing"] = not path_exists
return result
@@ -456,8 +467,8 @@ def format_namespace_workspace_binding_error(
def assess_namespace_mutation_workspace(
*,
role_kind: str,
worktree_path: str | None,
worktree: str | None,
worktree_path: str | None = None,
worktree: str | None = None,
process_project_root: str,
env: dict[str, str] | os._Environ | None = None,
session_lease_worktree: str | None = None,
@@ -509,14 +520,16 @@ def assess_namespace_mutation_workspace(
if crr_reasons:
reasons.extend(crr_reasons)
if not ctx.get("roots_aligned"):
if not crr_reasons:
reasons.append(
f"unsafe_process_root_workspace_alignment: process root '{process_root}' "
f"and canonical root '{ctx['canonical_repo_root']}' disagree"
)
path_exists = ctx.get("path_exists")
if path_exists is None:
path_exists = os.path.exists(mutation_workspace)
if os.path.exists(mutation_workspace):
if not path_exists:
if role != "author":
reasons.append(
f"{role} mutation blocked: configured workspace directory '{mutation_workspace}' does not exist (nonexistent worktree)"
)
else:
valid_common, common_err = verify_git_common_directory_membership(
mutation_workspace, ctx["canonical_repo_root"]
)
@@ -539,26 +552,27 @@ def assess_namespace_mutation_workspace(
)
if branches["block"]:
reasons.extend(branches["reasons"])
elif (
role == "reviewer"
and mutation_workspace == process_root
and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
):
reasons.append(
f"{role} mutation blocked: workspace is the stable control checkout; "
f"create or reconnect to a session-owned worktree under branches/ "
f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
f"{ACTIVE_WORKTREE_ENV}"
)
elif (
role in {"reviewer", "merger"}
and mutation_workspace != process_root
and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
):
reasons.append(
f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
f"'{ctx['canonical_repo_root']}/branches/'"
)
elif role in {"reviewer", "merger"}:
if mutation_workspace == process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
reasons.append(
f"{role} mutation blocked: workspace is the stable control checkout; "
f"create or reconnect to a session-owned worktree under branches/ "
f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
f"{ACTIVE_WORKTREE_ENV}"
)
elif mutation_workspace != process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
reasons.append(
f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
f"'{ctx['canonical_repo_root']}/branches/'"
)
if path_exists and amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
in_list = ctx.get("in_git_worktree_list")
if in_list is False:
reasons.append(
f"{role} mutation blocked: workspace '{mutation_workspace}' is under branches/ "
f"but is not registered in git worktree list for '{ctx['canonical_repo_root']}'"
)
block = bool(reasons)
return {
tests/test_issue_973_cross_repo_canonical_roots.py
+268 -27
View File
@@ -6,12 +6,15 @@ import os
import shutil
import tempfile
import unittest
from unittest.mock import patch, MagicMock
from pathlib import Path
import subprocess
import gitea_config
import namespace_workspace_binding as nwb
import canonical_repository_root as crr
import stable_control_runtime
import gitea_mcp_server as mcp_server
class TestIssue973RecognizedEnvKeys(unittest.TestCase):
@@ -53,29 +56,44 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase):
self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools")
os.makedirs(self.install_root)
subprocess.run(["git", "init", "-b", "master"], cwd=self.install_root, check=True)
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.install_root, check=True)
subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.install_root, check=True)
with open(os.path.join(self.install_root, "README.md"), "w") as f:
f.write("install\n")
subprocess.run(["git", "add", "README.md"], cwd=self.install_root, check=True)
subprocess.run(["git", "commit", "-m", "initial"], cwd=self.install_root, check=True)
# Create simulated target repository root
self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane")
os.makedirs(self.target_root)
subprocess.run(["git", "init", "-b", "master"], cwd=self.target_root, check=True)
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.target_root, check=True)
subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.target_root, check=True)
with open(os.path.join(self.target_root, "README.md"), "w") as f:
f.write("target\n")
subprocess.run(["git", "add", "README.md"], cwd=self.target_root, check=True)
subprocess.run(["git", "commit", "-m", "initial"], cwd=self.target_root, check=True)
# Create branches/ directory and a valid worktree in target repository
# Create branches/ directory and a valid registered worktree in target repository
self.target_branches = os.path.join(self.target_root, "branches")
os.makedirs(self.target_branches)
self.target_worktree = os.path.join(self.target_branches, "rev-pr-99")
os.makedirs(self.target_worktree)
# Create git common dir linking target_worktree to target_root
dot_git_file = os.path.join(self.target_worktree, ".git")
git_dir_target = os.path.join(self.target_root, ".git", "worktrees", "rev-pr-99")
os.makedirs(git_dir_target, exist_ok=True)
with open(dot_git_file, "w") as f:
f.write(f"gitdir: {git_dir_target}\n")
with open(os.path.join(git_dir_target, "commondir"), "w") as f:
f.write("../../..\n")
subprocess.run(["git", "worktree", "add", "-b", "rev-pr-99", self.target_worktree], cwd=self.target_root, check=True)
def tearDown(self):
self._tmp.cleanup()
def test_valid_same_repository_configuration(self):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="reviewer",
worktree_path=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=None,
)
self.assertEqual(ctx["canonical_repo_root"], self.install_root)
self.assertTrue(ctx["roots_aligned"])
self.assertTrue(ctx["canonical_root_assessment"]["proven"])
def test_valid_cross_repo_canonical_root(self):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="reviewer",
@@ -88,6 +106,54 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase):
self.assertTrue(ctx["roots_aligned"])
self.assertTrue(ctx["canonical_root_assessment"]["proven"])
def test_expected_repository_identity_match(self):
with patch("canonical_repository_root.repository_identity_slug", return_value="Scaled-Tech-Consulting/Gitea-Tools"):
assessment = crr.assess_canonical_repository_root(
configured_value=self.target_root,
source="test",
expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
process_project_root=self.install_root,
)
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
def test_foreign_repository_identity_mismatch(self):
with patch("canonical_repository_root.repository_identity_slug", return_value="Someone-Else/Evil-Repo"):
assessment = crr.assess_canonical_repository_root(
configured_value=self.target_root,
source="test",
expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
process_project_root=self.install_root,
)
self.assertFalse(assessment["proven"])
self.assertTrue(assessment["block"])
self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"]))
def test_native_repository_binding_mismatch(self):
with patch("canonical_repository_root.repository_identity_slug", return_value="Foreign/Repo"):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="reviewer",
worktree_path=self.target_worktree,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
)
self.assertFalse(ctx["roots_aligned"])
self.assertFalse(ctx["canonical_root_assessment"]["proven"])
self.assertTrue(any("identity mismatch" in r for r in ctx["canonical_root_assessment"]["reasons"]))
def test_missing_canonical_root(self):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="author",
worktree_path=None,
process_project_root=self.install_root,
env={},
configured_canonical_root="",
)
self.assertEqual(ctx["canonical_repo_root"], self.install_root)
self.assertTrue(ctx["roots_aligned"])
def test_nonexistent_configured_canonical_root(self):
nonexistent = os.path.join(self.tmp_dir, "nonexistent-repo")
ctx = nwb.resolve_namespace_mutation_context(
@@ -122,38 +188,213 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase):
self.assertTrue(valid, err)
self.assertIsNone(err)
def test_git_common_directory_membership_foreign_repo_blocks(self):
def test_foreign_git_common_directory(self):
# Foreign worktree created under install_root
foreign_wt = os.path.join(self.install_root, "branches", "foreign-wt")
os.makedirs(foreign_wt)
dot_git_file = os.path.join(foreign_wt, ".git")
git_dir_install = os.path.join(self.install_root, ".git", "worktrees", "foreign-wt")
os.makedirs(git_dir_install, exist_ok=True)
with open(dot_git_file, "w") as f:
f.write(f"gitdir: {git_dir_install}\n")
with open(os.path.join(git_dir_install, "commondir"), "w") as f:
f.write("../../..\n")
install_branches = os.path.join(self.install_root, "branches")
foreign_wt = os.path.join(install_branches, "foreign-wt")
subprocess.run(["git", "worktree", "add", "-b", "foreign-wt", foreign_wt], cwd=self.install_root, check=True)
valid, err = nwb.verify_git_common_directory_membership(
foreign_wt, self.target_root
)
self.assertFalse(valid)
self.assertIn("does not belong", err)
self.assertIn("does not match", err)
def test_reviewer_worktree_outside_branches_blocks(self):
outside_wt = os.path.join(self.target_root, "outside_branches_wt")
os.makedirs(outside_wt)
def test_normalized_path_aliases(self):
alias_path = self.target_worktree + "/../rev-pr-99/./"
valid, err = nwb.verify_git_common_directory_membership(
alias_path, self.target_root
)
self.assertTrue(valid, err)
def test_safe_symlink_identity(self):
link_path = os.path.join(self.target_branches, "symlink-rev-99")
try:
os.symlink(self.target_worktree, link_path)
valid, err = nwb.verify_git_common_directory_membership(
link_path, self.target_root
)
self.assertTrue(valid, err)
finally:
if os.path.exists(link_path):
os.unlink(link_path)
def test_symlink_escape_or_foreign_alias(self):
outside_dir = os.path.join(self.tmp_dir, "outside-target")
os.makedirs(outside_dir)
link_escape = os.path.join(self.target_branches, "escape-link")
try:
os.symlink(outside_dir, link_escape)
assessment = nwb.assess_namespace_mutation_workspace(
role_kind="reviewer",
worktree_path=link_escape,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertTrue(assessment["block"])
finally:
if os.path.exists(link_escape):
os.unlink(link_escape)
def test_reviewer_worktree_registered_and_valid(self):
assessment = nwb.assess_namespace_mutation_workspace(
role_kind="reviewer",
worktree_path=outside_wt,
worktree_path=self.target_worktree,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertFalse(assessment["block"])
def test_reviewer_worktree_unregistered_blocks(self):
unreg_wt = os.path.join(self.target_branches, "unregistered-reviewer")
os.makedirs(unreg_wt)
assessment = nwb.assess_namespace_mutation_workspace(
role_kind="reviewer",
worktree_path=unreg_wt,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertTrue(assessment["block"])
self.assertTrue(any("is not under" in r for r in assessment["reasons"]))
self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"]))
def test_merger_worktree_registered_and_valid(self):
assessment = nwb.assess_namespace_mutation_workspace(
role_kind="merger",
worktree_path=self.target_worktree,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertFalse(assessment["block"])
def test_merger_worktree_unregistered_blocks(self):
unreg_wt = os.path.join(self.target_branches, "unregistered-merger")
os.makedirs(unreg_wt)
assessment = nwb.assess_namespace_mutation_workspace(
role_kind="merger",
worktree_path=unreg_wt,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertTrue(assessment["block"])
self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"]))
def test_reviewer_or_merger_worktree_outside_branches_blocks(self):
outside_wt = os.path.join(self.target_root, "outside_branches_wt")
os.makedirs(outside_wt)
for r_kind in ("reviewer", "merger"):
assessment = nwb.assess_namespace_mutation_workspace(
role_kind=r_kind,
worktree_path=outside_wt,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertTrue(assessment["block"])
self.assertTrue(any("is not under" in r for r in assessment["reasons"]))
def test_nonexistent_reviewer_or_merger_worktree_blocks(self):
nonexistent_wt = os.path.join(self.target_branches, "nonexistent-wt")
for r_kind in ("reviewer", "merger"):
assessment = nwb.assess_namespace_mutation_workspace(
role_kind=r_kind,
worktree_path=nonexistent_wt,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertTrue(assessment["block"])
self.assertTrue(any("does not exist" in r for r in assessment["reasons"]))
def test_safe_and_unsafe_mutation_alignment_outcomes(self):
# Safe alignment (same repo)
report_safe = stable_control_runtime.build_runtime_report(
process_root=self.install_root,
checkout_branch="master",
runtime_head="abcdef123456",
active_task_workspace=self.install_root,
canonical_repository_root=self.install_root,
workspace_roots_aligned=True,
)
gate_safe = stable_control_runtime.assess_runtime_mutation_gate(report_safe)
self.assertFalse(gate_safe["block"])
# Unsafe alignment
report_unsafe = stable_control_runtime.build_runtime_report(
process_root=self.install_root,
checkout_branch="master",
runtime_head="abcdef123456",
active_task_workspace=self.target_worktree,
canonical_repository_root=self.target_root,
workspace_roots_aligned=False,
)
gate_unsafe = stable_control_runtime.assess_runtime_mutation_gate(report_unsafe)
self.assertTrue(gate_unsafe["block"])
def test_reviewer_lease_lifecycle_production_path(self):
"""B5: Automated regression for reviewer lease acquire and release through production path."""
mock_whoami = {
"authenticated": True,
"username": "sysadmin",
"remote": "prgs",
"profile": {
"profile_name": "prgs-reviewer",
"role": "reviewer",
"role_kind": "reviewer",
"allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.approve", "gitea.pr.request_changes"],
"forbidden_operations": [],
},
}
def mock_api_request(method, url, auth=None, json_data=None):
if method == "GET":
return {"number": 99, "head": {"sha": "abc1234"}, "state": "open", "merged": False, "merged_at": None}
elif method == "POST":
return {"id": 9999, "body": (json_data or {}).get("body", "")}
return {}
with patch.object(mcp_server, "gitea_whoami", return_value=mock_whoami), \
patch.object(mcp_server, "get_profile", return_value=mock_whoami["profile"]), \
patch.object(mcp_server, "_effective_workspace_role", return_value="reviewer"), \
patch.object(mcp_server, "_configured_canonical_root", return_value=(self.target_root, "env")), \
patch.object(mcp_server, "_reviewer_session_worktree", return_value=self.target_worktree), \
patch.object(mcp_server, "_auth", return_value="token mock-token"), \
patch.object(mcp_server, "_fetch_pr_comments", return_value=[]), \
patch.object(mcp_server, "api_request", side_effect=mock_api_request), \
patch("reviewer_pr_lease.assess_acquire_lease", return_value={"acquire_allowed": True, "reasons": [], "lease_body": "<!-- LEASE -->"}), \
patch("reviewer_pr_lease.find_active_reviewer_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \
patch("reviewer_pr_lease.get_session_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \
patch("reviewer_pr_lease.clear_session_lease") as mock_clear:
acq_res = mcp_server.gitea_acquire_reviewer_pr_lease(
pr_number=99,
remote="prgs",
worktree=self.target_worktree,
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertTrue(acq_res.get("success"), acq_res)
rel_res = mcp_server.gitea_release_reviewer_pr_lease(
pr_number=99,
worktree=self.target_worktree,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertTrue(rel_res.get("success"), rel_res)
mock_clear.assert_called_once()
if __name__ == "__main__":
tests/test_namespace_workspace_binding.py
+7 -2
View File
@@ -245,10 +245,15 @@ class TestNamespaceWorkspaceIntegration(unittest.TestCase):
def test_pr487_style_merge_binds_clean_merger_workspace(
self, _exists, _isdir, mock_run
):
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
def mock_git(cmd, *args, **kwargs):
if "rev-parse" in cmd:
return MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
return MagicMock(returncode=0, stdout=f"worktree {CONTROL_ROOT}\nworktree {MERGER_CLEAN}\n")
mock_run.side_effect = mock_git
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
srv._preflight_resolved_role = "reviewer"
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
resolved = srv._verify_role_mutation_workspace("prgs")
self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT))
self.assertEqual(resolved, os.path.realpath(MERGER_CLEAN))