diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 4ffaff7..7d3fa38 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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(), diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index 911bf78..9025be9 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -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 { diff --git a/tests/test_issue_973_cross_repo_canonical_roots.py b/tests/test_issue_973_cross_repo_canonical_roots.py index 15ca7bf..47e8ca6 100644 --- a/tests/test_issue_973_cross_repo_canonical_roots.py +++ b/tests/test_issue_973_cross_repo_canonical_roots.py @@ -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", "test@example.com"], 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", "test@example.com"], 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": ""}), \ + 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__": diff --git a/tests/test_namespace_workspace_binding.py b/tests/test_namespace_workspace_binding.py index 46e0b19..fedfe4e 100644 --- a/tests/test_namespace_workspace_binding.py +++ b/tests/test_namespace_workspace_binding.py @@ -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)) \ No newline at end of file + self.assertEqual(resolved, os.path.realpath(MERGER_CLEAN)) \ No newline at end of file