Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
bc9366c394 |
+139
-81
@@ -517,6 +517,19 @@ def _clear_preflight_capability_state() -> None:
|
||||
_preflight_reviewer_violation_files = []
|
||||
|
||||
|
||||
def _clear_resolved_capability_stamp() -> None:
|
||||
"""#723 AC3: drop only the resolved role/task stamp (denied resolve).
|
||||
|
||||
``record_preflight_check("capability")`` without a role intentionally
|
||||
preserves a prior stamp, so a DENIED resolve must clear it explicitly —
|
||||
otherwise the stale role poisons :func:`_effective_workspace_role` and a
|
||||
later mutation preflight raises instead of failing closed with reasons.
|
||||
"""
|
||||
global _preflight_resolved_role, _preflight_resolved_task
|
||||
_preflight_resolved_role = None
|
||||
_preflight_resolved_task = None
|
||||
|
||||
|
||||
def record_preflight_check(
|
||||
type_name: str,
|
||||
resolved_role: str | None = None,
|
||||
@@ -4096,9 +4109,6 @@ def _evaluate_pr_review_submission(
|
||||
worktree_path: str | None = None,
|
||||
) -> dict:
|
||||
"""Shared gate chain for live submit and dry-run review tools."""
|
||||
_verify_role_mutation_workspace(
|
||||
remote, worktree_path=worktree_path, task="review_pr"
|
||||
)
|
||||
action = (action or "").strip().lower()
|
||||
workflow_blockers = _review_workflow_load_gate_reasons() if live else []
|
||||
result = {
|
||||
@@ -4116,6 +4126,20 @@ def _evaluate_pr_review_submission(
|
||||
"reasons": [],
|
||||
}
|
||||
reasons = result["reasons"]
|
||||
# #723 AC4: workspace/role-binding failures fail closed with structured
|
||||
# reasons — never escape as a generic internal_error (PR #721: a stamped
|
||||
# 'merger' role raised RuntimeError here and the client saw an
|
||||
# unactionable internal_error after mark_final had already succeeded).
|
||||
try:
|
||||
_verify_role_mutation_workspace(
|
||||
remote, worktree_path=worktree_path, task="review_pr"
|
||||
)
|
||||
except RuntimeError as exc:
|
||||
result["blocker_kind"] = "workspace_role_binding"
|
||||
reasons.append(
|
||||
"workspace/role binding failed (fail closed, #723): " f"{exc}"
|
||||
)
|
||||
return result
|
||||
if workflow_blockers:
|
||||
reasons.extend(workflow_blockers)
|
||||
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
|
||||
@@ -11891,8 +11915,16 @@ _RUNTIME_CAPABILITY_TASKS = (
|
||||
def _matching_configured_profiles(
|
||||
config: dict | None,
|
||||
required_permission: str,
|
||||
required_role_kind: str | None = None,
|
||||
) -> list[str]:
|
||||
"""Profile names that allow *required_permission* (redacted metadata only)."""
|
||||
"""Profile names that allow *required_permission* (redacted metadata only).
|
||||
|
||||
#723 AC5: when *required_role_kind* is supplied (role-exclusive tasks),
|
||||
a profile with a declared role must also match it — mirroring the
|
||||
resolver's matching rule so the two lists cannot drift. Profiles without
|
||||
a declared role still match on permission alone, exactly like the
|
||||
resolver.
|
||||
"""
|
||||
if not config or "profiles" not in config:
|
||||
return []
|
||||
matches: list[str] = []
|
||||
@@ -11916,7 +11948,13 @@ def _matching_configured_profiles(
|
||||
ok, _ = gitea_config.check_operation(
|
||||
required_permission, p_allowed_n, p_forbidden_n
|
||||
)
|
||||
if ok:
|
||||
p_role = (p_data.get("role") or "").strip()
|
||||
role_ok = (
|
||||
required_role_kind is None
|
||||
or not p_role
|
||||
or p_role == required_role_kind
|
||||
)
|
||||
if ok and role_ok:
|
||||
matches.append(p_name)
|
||||
return sorted(matches)
|
||||
|
||||
@@ -11925,8 +11963,16 @@ def _build_runtime_task_capabilities(
|
||||
allowed: list[str],
|
||||
forbidden: list[str],
|
||||
config: dict | None,
|
||||
active_role_kind: str | None = None,
|
||||
) -> dict:
|
||||
"""Per-task capability summary for role-aware runtime context (#139)."""
|
||||
"""Per-task capability summary for role-aware runtime context (#139).
|
||||
|
||||
#723 AC5: applies the same role-exclusive filter as
|
||||
``gitea_resolve_task_capability`` when *active_role_kind* is supplied, so
|
||||
runtime context can never report a role-exclusive task as allowed while
|
||||
the resolver fail-closes it in the same session (incident #722). Without
|
||||
an *active_role_kind* the view is permission-only and each entry says so.
|
||||
"""
|
||||
task_entries = []
|
||||
flags: dict[str, bool] = {}
|
||||
flag_keys = {
|
||||
@@ -11939,18 +11985,35 @@ def _build_runtime_task_capabilities(
|
||||
"close_issue": "can_close_issues",
|
||||
"reconcile_already_landed_pr": "can_reconcile_already_landed_prs",
|
||||
}
|
||||
role_filter_applied = active_role_kind is not None
|
||||
for task in _RUNTIME_CAPABILITY_TASKS:
|
||||
permission = task_capability_map.required_permission(task)
|
||||
allowed_here, _ = gitea_config.check_operation(
|
||||
required_role_kind = task_capability_map.required_role(task)
|
||||
role_exclusive = task in task_capability_map.ROLE_EXCLUSIVE_TASKS
|
||||
permission_allowed, _ = gitea_config.check_operation(
|
||||
permission, allowed, forbidden
|
||||
)
|
||||
role_ok = (
|
||||
not role_exclusive
|
||||
or not role_filter_applied
|
||||
or active_role_kind == required_role_kind
|
||||
)
|
||||
allowed_here = permission_allowed and role_ok
|
||||
entry = {
|
||||
"task": task,
|
||||
"required_permission": permission,
|
||||
"required_role_kind": task_capability_map.required_role(task),
|
||||
"required_role_kind": required_role_kind,
|
||||
"role_exclusive": role_exclusive,
|
||||
"capability_view": (
|
||||
"role_filtered" if role_filter_applied else "permission_only"
|
||||
),
|
||||
"allowed_in_current_session": allowed_here,
|
||||
"matching_configured_profiles": _matching_configured_profiles(
|
||||
config, permission
|
||||
config,
|
||||
permission,
|
||||
required_role_kind=(
|
||||
required_role_kind if role_exclusive else None
|
||||
),
|
||||
),
|
||||
}
|
||||
task_entries.append(entry)
|
||||
@@ -12403,7 +12466,10 @@ def gitea_get_runtime_context(
|
||||
)
|
||||
|
||||
session_capabilities = _build_runtime_task_capabilities(
|
||||
allowed, forbidden, config
|
||||
allowed,
|
||||
forbidden,
|
||||
config,
|
||||
active_role_kind=_profile_role_kind(profile),
|
||||
)
|
||||
|
||||
preflight = assess_preflight_status(worktree_path)
|
||||
@@ -13659,18 +13725,36 @@ def gitea_route_task_session(
|
||||
)
|
||||
|
||||
|
||||
# #685: resolver stale-runtime detection is report-only. Config touch / os._exit
|
||||
# self-recovery was removed from the read-only path (was _trigger_mcp_auto_restart).
|
||||
# Recovery is owned exclusively by the IDE/client reconnect path.
|
||||
_restart_triggered = False
|
||||
|
||||
|
||||
def _trigger_mcp_auto_restart():
|
||||
global _restart_triggered
|
||||
if _restart_triggered or _preflight_in_test_mode():
|
||||
return
|
||||
_restart_triggered = True
|
||||
|
||||
config_path = os.environ.get(
|
||||
"MCP_CONFIG_PATH",
|
||||
os.path.expanduser("~/.gemini/config/mcp_config.json")
|
||||
)
|
||||
|
||||
try:
|
||||
if os.path.exists(config_path):
|
||||
os.utime(config_path, None)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
import threading
|
||||
import time
|
||||
def delayed_exit():
|
||||
time.sleep(1.0)
|
||||
os._exit(0)
|
||||
threading.Thread(target=delayed_exit, daemon=True).start()
|
||||
|
||||
|
||||
def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
|
||||
"""Read-only: report missing or stale MCP runtimes (no config or process mutation).
|
||||
|
||||
#685: Never touches MCP client config, never spawns recovery threads, never
|
||||
calls ``os._exit``. Stale detection remains fail-closed via returned reasons
|
||||
only; the IDE/client owns reconnect/reload.
|
||||
"""
|
||||
"""Check running runtimes and return errors if they are missing or stale."""
|
||||
import subprocess
|
||||
import re
|
||||
from datetime import datetime
|
||||
@@ -13750,13 +13834,11 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
|
||||
}
|
||||
|
||||
if self_stale:
|
||||
# #685: report-only — no config utime, no thread, no os._exit.
|
||||
_trigger_mcp_auto_restart()
|
||||
reasons.append(
|
||||
"stale-runtime: The active Gitea MCP server process is stale "
|
||||
"(running code from before changes were merged). "
|
||||
"Reconnect the IDE/client-managed MCP namespace for this profile "
|
||||
"so it reloads current master. The resolver does not touch "
|
||||
"mcp_config.json, spawn recovery threads, or terminate this process."
|
||||
"stale-runtime: The active Gitea MCP server process is stale (running code from before changes were merged). "
|
||||
"Auto-restart has been triggered: touched mcp_config.json to reload the daemon. "
|
||||
"The current process will cleanly exit shortly."
|
||||
)
|
||||
|
||||
if matching_profiles:
|
||||
@@ -13788,15 +13870,10 @@ def gitea_resolve_task_capability(
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
) -> dict:
|
||||
"""Read-only / side-effect free: resolve capability, profile, and namespace for a task.
|
||||
"""Read-only: Resolve which capability, profile, and namespace is required for a Gitea task.
|
||||
|
||||
Does **not** mutate MCP client configuration, spawn recovery threads, kill
|
||||
processes, or trigger daemon reloads (#685). Stale-runtime detection remains
|
||||
fail-closed: when the serving process is stale the result includes
|
||||
``blocker_kind=runtime_reconnect_required``, ``restart_required=true``,
|
||||
``stop_required=true``, and ``mutation_performed=false`` with a precise
|
||||
``exact_safe_next_action`` pointing at IDE/client reconnect. Recovery is
|
||||
owned by the client reconnect path — never by this resolver.
|
||||
Helps the client or LLM determine the correct namespace or profile before acting,
|
||||
and returns exact next action instructions if the current session is not authorized.
|
||||
|
||||
Args:
|
||||
task: The task/action to check (e.g. review_pr, create_issue).
|
||||
@@ -13810,26 +13887,9 @@ def gitea_resolve_task_capability(
|
||||
|
||||
required_permission = task_capability_map.required_permission(task)
|
||||
required_role = task_capability_map.required_role(task)
|
||||
role_exclusive_tasks = {
|
||||
"review_pr",
|
||||
"approve_pr",
|
||||
"request_changes_pr",
|
||||
"blind_pr_queue_review",
|
||||
"pr_queue_cleanup",
|
||||
"pr-queue-cleanup",
|
||||
"merge_pr",
|
||||
"create_branch",
|
||||
"push_branch",
|
||||
"create_pr",
|
||||
"commit_files",
|
||||
"gitea_commit_files",
|
||||
"address_pr_change_requests",
|
||||
"delete_branch",
|
||||
"cleanup_merged_pr_branch",
|
||||
"reconciliation_cleanup",
|
||||
"work_issue",
|
||||
"work-issue",
|
||||
}
|
||||
# #723 AC5: single shared definition; the runtime-context capability
|
||||
# report applies the same set so the two views cannot drift.
|
||||
role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
|
||||
|
||||
infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
|
||||
if required_role == "reviewer":
|
||||
@@ -13910,7 +13970,12 @@ def gitea_resolve_task_capability(
|
||||
"exact_safe_next_action": next_safe_action,
|
||||
}
|
||||
|
||||
record_preflight_check("capability", required_role, resolved_task=task)
|
||||
# #723 AC3: record the capability purity baseline now, but DEFER the
|
||||
# resolved-role/task stamp until after the allow/deny decision below. A
|
||||
# denied resolve previously stamped ``_preflight_resolved_role`` anyway
|
||||
# (PR #721: a denied review_pr resolve stamped 'merger', and the later
|
||||
# submit escaped as a generic internal_error instead of failing closed).
|
||||
record_preflight_check("capability")
|
||||
|
||||
# Try automatic dispatch switching
|
||||
_ensure_matching_profile(required_permission, required_role, remote, host)
|
||||
@@ -13945,6 +14010,14 @@ def gitea_resolve_task_capability(
|
||||
permission_allowed_in_current_session and role_matches_current_session
|
||||
)
|
||||
|
||||
# #723 AC3: stamp the resolved role/task only for an ALLOWED resolve; a
|
||||
# denied resolve clears any stale stamp so later mutation preflights key
|
||||
# off the real profile role and fail closed with structured reasons.
|
||||
if allowed_in_current_session:
|
||||
record_preflight_check("capability", required_role, resolved_task=task)
|
||||
else:
|
||||
_clear_resolved_capability_stamp()
|
||||
|
||||
switching = gitea_config.is_runtime_switching_enabled()
|
||||
available_in_session = allowed_in_current_session
|
||||
configured = False
|
||||
@@ -13981,32 +14054,31 @@ def gitea_resolve_task_capability(
|
||||
|
||||
configured = len(matching_profiles) > 0
|
||||
available_in_session = allowed_in_current_session
|
||||
runtime_stale_blocker = False
|
||||
|
||||
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
|
||||
runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
|
||||
if runtime_reasons:
|
||||
restart_required = True
|
||||
runtime_stale_blocker = True
|
||||
reason_msg = "; ".join(runtime_reasons)
|
||||
next_safe_action = (
|
||||
"stale-runtime: Gitea MCP runtime conflict or missing process detected. "
|
||||
"Please fully restart the Gitea MCP server and retry."
|
||||
)
|
||||
|
||||
if not allowed_in_current_session:
|
||||
if configured and switching:
|
||||
restart_required = True
|
||||
available_in_session = False
|
||||
if not reason_msg:
|
||||
reason_msg = (
|
||||
f"{required_role.capitalize()} profile exists but MCP server "
|
||||
"was added after session startup and is not attached."
|
||||
)
|
||||
reason_msg = (
|
||||
f"{required_role.capitalize()} profile exists but MCP server "
|
||||
"was added after session startup and is not attached."
|
||||
)
|
||||
elif not configured:
|
||||
if not reason_msg:
|
||||
reason_msg = (
|
||||
f"No profile configured with permission '{required_permission}'."
|
||||
)
|
||||
reason_msg = (
|
||||
f"No profile configured with permission '{required_permission}'."
|
||||
)
|
||||
elif role_mismatch_reason:
|
||||
if not reason_msg:
|
||||
reason_msg = role_mismatch_reason
|
||||
reason_msg = role_mismatch_reason
|
||||
different_namespace_required = False
|
||||
next_safe_action = "None; ready for operations."
|
||||
|
||||
@@ -14032,16 +14104,6 @@ def gitea_resolve_task_capability(
|
||||
"or use the corresponding MCP namespace."
|
||||
)
|
||||
|
||||
# #685: stale-runtime typed remediation wins for exact_next_action when the
|
||||
# serving process/profile inventory is stale — even if permission is OK.
|
||||
if runtime_stale_blocker:
|
||||
next_safe_action = (
|
||||
"blocker_kind=runtime_reconnect_required: reconnect/restart the "
|
||||
"IDE-managed Gitea MCP server for this profile so it reloads current "
|
||||
"master. Do not edit mcp_config.json by hand; the resolver does not "
|
||||
"touch config, spawn recovery threads, or terminate the process."
|
||||
)
|
||||
|
||||
# Task/role alignment guards (#167): the requested task, not the
|
||||
# available credential, decides what the session may do. A review/merge
|
||||
# task under a non-reviewer profile must stop — not silently degrade
|
||||
@@ -14103,16 +14165,12 @@ def gitea_resolve_task_capability(
|
||||
"configured": configured,
|
||||
"restart_required": restart_required,
|
||||
"stop_required": stop_required or restart_required,
|
||||
# #685: resolver is always side-effect free; never claims mutations.
|
||||
"mutation_performed": False,
|
||||
"task_role_guidance": task_role_guidance,
|
||||
"matching_configured_profile": matching_profiles,
|
||||
"runtime_switching_supported": switching,
|
||||
"different_mcp_namespace_required": different_namespace_required,
|
||||
"exact_safe_next_action": next_safe_action,
|
||||
}
|
||||
if runtime_stale_blocker:
|
||||
result["blocker_kind"] = "runtime_reconnect_required"
|
||||
if reason_msg:
|
||||
result["reason"] = reason_msg
|
||||
if task in ("review_pr", "merge_pr"):
|
||||
|
||||
@@ -330,6 +330,32 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
},
|
||||
}
|
||||
|
||||
# #723 AC1/AC5: tasks where permission alone is NOT enough — the profile's
|
||||
# role kind must also match. Shared by ``gitea_resolve_task_capability`` and
|
||||
# the runtime-context capability report so the two can never disagree about
|
||||
# which tasks are role-exclusive (incident #722: runtime context said
|
||||
# review_pr was allowed while the resolver fail-closed the same session).
|
||||
ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset({
|
||||
"review_pr",
|
||||
"approve_pr",
|
||||
"request_changes_pr",
|
||||
"blind_pr_queue_review",
|
||||
"pr_queue_cleanup",
|
||||
"pr-queue-cleanup",
|
||||
"merge_pr",
|
||||
"create_branch",
|
||||
"push_branch",
|
||||
"create_pr",
|
||||
"commit_files",
|
||||
"gitea_commit_files",
|
||||
"address_pr_change_requests",
|
||||
"delete_branch",
|
||||
"cleanup_merged_pr_branch",
|
||||
"reconciliation_cleanup",
|
||||
"work_issue",
|
||||
"work-issue",
|
||||
})
|
||||
|
||||
# Issue-mutating MCP tools and their resolver task keys.
|
||||
ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
|
||||
"gitea_create_issue": "create_issue",
|
||||
|
||||
@@ -1,326 +0,0 @@
|
||||
"""#685: gitea_resolve_task_capability must be side-effect free.
|
||||
|
||||
Stale-runtime detection remains fail-closed, but the resolver must never:
|
||||
* touch mcp_config.json (or any MCP client config)
|
||||
* spawn recovery threads
|
||||
* call os._exit / terminate the serving process
|
||||
* claim that an auto-restart was triggered
|
||||
|
||||
Recovery is owned by the IDE/client reconnect path only.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import sys
|
||||
|
||||
ROOT = str(Path(__file__).resolve().parent.parent)
|
||||
if ROOT not in sys.path:
|
||||
sys.path.insert(0, ROOT)
|
||||
|
||||
import gitea_mcp_server as mcp_server
|
||||
|
||||
|
||||
ROLE_PROFILES = (
|
||||
("create_issue", "prgs-author", "author"),
|
||||
("review_pr", "prgs-reviewer", "reviewer"),
|
||||
("merge_pr", "prgs-merger", "merger"),
|
||||
("reconciliation_cleanup", "prgs-reconciler", "reconciler"),
|
||||
)
|
||||
|
||||
|
||||
def _stale_self_ps_mocks(profile: str = "prgs-author"):
|
||||
"""Build subprocess mocks: self PID is stale vs code mtime."""
|
||||
mock_getpid = MagicMock(return_value=12345)
|
||||
mock_exists = MagicMock(return_value=True)
|
||||
code_time = datetime(2026, 7, 8, 14, 0, 0)
|
||||
mock_getmtime = MagicMock(return_value=code_time.timestamp())
|
||||
|
||||
ps_output = (
|
||||
" PID LSTART COMMAND\n"
|
||||
"12345 Wed Jul 8 13:00:00 2026 /path/to/python mcp_server.py\n"
|
||||
)
|
||||
mock_run_ps = MagicMock()
|
||||
mock_run_ps.stdout = ps_output
|
||||
|
||||
mock_run_env = MagicMock()
|
||||
mock_run_env.stdout = f"GITEA_MCP_PROFILE={profile}"
|
||||
|
||||
mock_run_git = MagicMock()
|
||||
mock_run_git.stdout = "SAME"
|
||||
|
||||
def side_effect(args, **kwargs):
|
||||
if args[0] == "ps" and "eww" in args:
|
||||
return mock_run_env
|
||||
if args[0] == "ps":
|
||||
return mock_run_ps
|
||||
if args[0] == "git":
|
||||
return mock_run_git
|
||||
raise ValueError(f"Unexpected subprocess args: {args}")
|
||||
|
||||
mock_run = MagicMock(side_effect=side_effect)
|
||||
return mock_getpid, mock_exists, mock_getmtime, mock_run
|
||||
|
||||
|
||||
class TestIssue685DiagnosticsNoSideEffects(unittest.TestCase):
|
||||
def setUp(self):
|
||||
mcp_server._process_boot_head_sha = None
|
||||
|
||||
def tearDown(self):
|
||||
mcp_server._process_boot_head_sha = None
|
||||
|
||||
@patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
|
||||
@patch("subprocess.run")
|
||||
@patch("os.path.getmtime")
|
||||
@patch("os.path.exists")
|
||||
@patch("os.getpid")
|
||||
@patch("os.utime")
|
||||
@patch("threading.Thread")
|
||||
@patch("os._exit")
|
||||
def test_stale_self_does_not_touch_config_or_exit(
|
||||
self,
|
||||
mock_exit,
|
||||
mock_thread,
|
||||
mock_utime,
|
||||
mock_getpid,
|
||||
mock_exists,
|
||||
mock_getmtime,
|
||||
mock_run,
|
||||
):
|
||||
mock_getpid.return_value = 12345
|
||||
mock_exists.return_value = True
|
||||
mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
|
||||
mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
|
||||
|
||||
before_threads = threading.active_count()
|
||||
reasons = mcp_server._check_mcp_runtimes_diagnostics(
|
||||
"create_issue", ["prgs-author"]
|
||||
)
|
||||
after_threads = threading.active_count()
|
||||
|
||||
self.assertTrue(
|
||||
any("stale-runtime" in r and "active Gitea MCP server process is stale" in r
|
||||
for r in reasons),
|
||||
reasons,
|
||||
)
|
||||
# Must not claim auto-restart / config touch
|
||||
blob = " ".join(reasons)
|
||||
self.assertNotIn("Auto-restart has been triggered", blob)
|
||||
self.assertNotIn("touched mcp_config", blob)
|
||||
self.assertNotIn("will cleanly exit", blob)
|
||||
|
||||
mock_utime.assert_not_called()
|
||||
mock_thread.assert_not_called()
|
||||
mock_exit.assert_not_called()
|
||||
self.assertEqual(before_threads, after_threads)
|
||||
|
||||
@patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
|
||||
@patch("subprocess.run")
|
||||
@patch("os.path.getmtime")
|
||||
@patch("os.path.exists")
|
||||
@patch("os.getpid")
|
||||
@patch("os.utime")
|
||||
def test_repeated_stale_calls_do_not_trigger_restart_loop(
|
||||
self, mock_utime, mock_getpid, mock_exists, mock_getmtime, mock_run
|
||||
):
|
||||
mock_getpid.return_value = 12345
|
||||
mock_exists.return_value = True
|
||||
mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
|
||||
mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
|
||||
|
||||
for _ in range(5):
|
||||
reasons = mcp_server._check_mcp_runtimes_diagnostics(
|
||||
"create_issue", ["prgs-author"]
|
||||
)
|
||||
self.assertTrue(any("stale-runtime" in r for r in reasons))
|
||||
|
||||
mock_utime.assert_not_called()
|
||||
|
||||
def test_trigger_mcp_auto_restart_removed(self):
|
||||
"""#685 AC: auto-restart helper is removed (unreachable from read-only)."""
|
||||
self.assertFalse(hasattr(mcp_server, "_trigger_mcp_auto_restart"))
|
||||
self.assertFalse(hasattr(mcp_server, "_restart_triggered"))
|
||||
|
||||
|
||||
class TestIssue685ResolverTypedBlocker(unittest.TestCase):
|
||||
def setUp(self):
|
||||
mcp_server._process_boot_head_sha = None
|
||||
|
||||
def tearDown(self):
|
||||
mcp_server._process_boot_head_sha = None
|
||||
if hasattr(mcp_server, "capability_stop_terminal"):
|
||||
mcp_server.capability_stop_terminal.clear()
|
||||
|
||||
def _resolve_with_stale_runtime(self, task: str, profile_name: str, role: str):
|
||||
allowed = [
|
||||
"gitea.read",
|
||||
"gitea.issue.create",
|
||||
"gitea.issue.comment",
|
||||
"gitea.issue.close",
|
||||
"gitea.branch.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.branch.delete",
|
||||
"gitea.pr.create",
|
||||
"gitea.pr.comment",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.close",
|
||||
"gitea.repo.commit",
|
||||
]
|
||||
profile = {
|
||||
"profile_name": profile_name,
|
||||
"role": role,
|
||||
"allowed_operations": allowed,
|
||||
"forbidden_operations": [],
|
||||
}
|
||||
config = {
|
||||
"profiles": {
|
||||
profile_name: {
|
||||
"role": role,
|
||||
"allowed_operations": allowed,
|
||||
"forbidden_operations": [],
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
mock_getpid, mock_exists, mock_getmtime, mock_run = _stale_self_ps_mocks(
|
||||
profile_name
|
||||
)
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"GITEA_FORCE_MCP_RUNTIME_CHECK": "1",
|
||||
"GITEA_MCP_PROFILE": profile_name,
|
||||
},
|
||||
clear=False,
|
||||
), patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server.gitea_config, "load_config", return_value=config
|
||||
), patch.object(
|
||||
mcp_server, "_authenticated_username", return_value="test-user"
|
||||
), patch.object(
|
||||
mcp_server, "_ensure_matching_profile", return_value=None
|
||||
), patch.object(
|
||||
mcp_server, "record_preflight_check", return_value=None
|
||||
), patch.object(
|
||||
mcp_server, "record_mutation_authority", return_value=None
|
||||
), patch.object(
|
||||
mcp_server, "init_review_decision_lock", return_value=None
|
||||
), patch(
|
||||
"subprocess.run", mock_run
|
||||
), patch(
|
||||
"os.path.getmtime", mock_getmtime
|
||||
), patch(
|
||||
"os.path.exists", mock_exists
|
||||
), patch(
|
||||
"os.getpid", mock_getpid
|
||||
), patch(
|
||||
"os.utime"
|
||||
) as mock_utime, patch(
|
||||
"threading.Thread"
|
||||
) as mock_thread, patch(
|
||||
"os._exit"
|
||||
) as mock_exit:
|
||||
result = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
|
||||
return result, mock_utime, mock_thread, mock_exit
|
||||
|
||||
def test_stale_returns_typed_blocker_fields(self):
|
||||
result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
|
||||
"create_issue", "prgs-author", "author"
|
||||
)
|
||||
self.assertTrue(result.get("restart_required"), result)
|
||||
self.assertTrue(result.get("stop_required"), result)
|
||||
self.assertEqual(result.get("blocker_kind"), "runtime_reconnect_required")
|
||||
self.assertIs(result.get("mutation_performed"), False)
|
||||
action = result.get("exact_safe_next_action") or ""
|
||||
self.assertIn("reconnect", action.lower())
|
||||
self.assertNotIn("None; ready for operations", action)
|
||||
reason = result.get("reason") or ""
|
||||
self.assertIn("stale-runtime", reason)
|
||||
self.assertNotIn("Auto-restart has been triggered", reason)
|
||||
mock_utime.assert_not_called()
|
||||
mock_thread.assert_not_called()
|
||||
mock_exit.assert_not_called()
|
||||
|
||||
def test_all_four_role_profiles_get_same_side_effect_free_contract(self):
|
||||
for task, profile, role in ROLE_PROFILES:
|
||||
with self.subTest(task=task, profile=profile):
|
||||
# Skip tasks that may be unknown on this branch
|
||||
try:
|
||||
import task_capability_map as tcm
|
||||
|
||||
tcm.required_permission(task)
|
||||
except Exception:
|
||||
self.skipTest(f"task {task} not in capability map")
|
||||
|
||||
result, mock_utime, mock_thread, mock_exit = (
|
||||
self._resolve_with_stale_runtime(task, profile, role)
|
||||
)
|
||||
self.assertTrue(
|
||||
result.get("restart_required") or result.get("stop_required"),
|
||||
result,
|
||||
)
|
||||
self.assertEqual(
|
||||
result.get("blocker_kind"), "runtime_reconnect_required", result
|
||||
)
|
||||
self.assertIs(result.get("mutation_performed"), False, result)
|
||||
mock_utime.assert_not_called()
|
||||
mock_thread.assert_not_called()
|
||||
mock_exit.assert_not_called()
|
||||
|
||||
def test_config_mtime_and_contents_unchanged(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
cfg = os.path.join(tmp, "mcp_config.json")
|
||||
original = '{"servers": {"gitea-author": {}}}'
|
||||
with open(cfg, "w", encoding="utf-8") as fh:
|
||||
fh.write(original)
|
||||
mtime_before = os.path.getmtime(cfg)
|
||||
|
||||
result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
|
||||
"create_issue", "prgs-author", "author"
|
||||
)
|
||||
# Force-path also must not use real utime when diagnostics runs
|
||||
with open(cfg, encoding="utf-8") as fh:
|
||||
after = fh.read()
|
||||
self.assertEqual(after, original)
|
||||
self.assertEqual(os.path.getmtime(cfg), mtime_before)
|
||||
mock_utime.assert_not_called()
|
||||
self.assertTrue(result.get("restart_required"), result)
|
||||
|
||||
|
||||
class TestIssue685MutationGatesStillFailClosed(unittest.TestCase):
|
||||
def test_parity_stale_still_reports_restart_required(self):
|
||||
"""Mutation-facing parity gate remains fail-closed when heads differ."""
|
||||
import master_parity_gate as mpg
|
||||
|
||||
out = mpg.assess_master_parity(
|
||||
{"startup_head": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
|
||||
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
|
||||
)
|
||||
self.assertFalse(out.get("in_parity"))
|
||||
self.assertTrue(out.get("restart_required") or out.get("stale"))
|
||||
|
||||
|
||||
class TestIssue685DocstringReadOnlyContract(unittest.TestCase):
|
||||
def test_resolve_docstring_declares_side_effect_free(self):
|
||||
doc = mcp_server.gitea_resolve_task_capability.__doc__ or ""
|
||||
lower = doc.lower()
|
||||
self.assertTrue(
|
||||
"side-effect" in lower or "read-only" in lower or "does not mutate" in lower,
|
||||
doc,
|
||||
)
|
||||
self.assertNotIn("auto-restart", lower)
|
||||
self.assertNotIn("os._exit", lower)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,337 @@
|
||||
"""#723 AC3/AC4/AC5: denied resolves must not poison the session role stamp,
|
||||
review-submission workspace failures must fail closed with structured
|
||||
reasons, and runtime-context capabilities must apply the resolver's
|
||||
role-exclusive filter.
|
||||
|
||||
Reproduces the PR #721 sequence: a reviewer session resolved a task whose
|
||||
capability-map role had drifted to ``merger``; the denied resolve stamped
|
||||
``_preflight_resolved_role = "merger"`` anyway, ``mark_final`` succeeded, and
|
||||
``gitea_submit_pr_review`` raised RuntimeError out of the tool as a generic
|
||||
``internal_error``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
ROOT = str(Path(__file__).resolve().parent.parent)
|
||||
if ROOT not in sys.path:
|
||||
sys.path.insert(0, ROOT)
|
||||
|
||||
import gitea_mcp_server as mcp_server
|
||||
import task_capability_map
|
||||
|
||||
|
||||
REVIEWER_PROFILE = {
|
||||
"profile_name": "prgs-reviewer",
|
||||
"role": "reviewer",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.branch.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.repo.commit",
|
||||
"gitea.pr.create",
|
||||
"gitea.pr.merge",
|
||||
],
|
||||
}
|
||||
|
||||
CONFIG = {
|
||||
"profiles": {
|
||||
"prgs-reviewer": {
|
||||
"role": "reviewer",
|
||||
"allowed_operations": REVIEWER_PROFILE["allowed_operations"],
|
||||
"forbidden_operations": REVIEWER_PROFILE["forbidden_operations"],
|
||||
},
|
||||
"prgs-merger": {
|
||||
"role": "merger",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.request_changes",
|
||||
],
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _reset_preflight():
|
||||
mcp_server._clear_preflight_capability_state()
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_whoami_violation = False
|
||||
|
||||
|
||||
class _ResolveHarness(unittest.TestCase):
|
||||
"""Shared patched-resolver harness."""
|
||||
|
||||
def setUp(self):
|
||||
_reset_preflight()
|
||||
|
||||
def tearDown(self):
|
||||
_reset_preflight()
|
||||
|
||||
def _resolve(self, task, profile=REVIEWER_PROFILE, required_role=None):
|
||||
"""Run gitea_resolve_task_capability with a fixed profile/config."""
|
||||
patches = [
|
||||
patch.object(mcp_server, "get_profile", return_value=profile),
|
||||
patch.object(
|
||||
mcp_server.gitea_config, "load_config", return_value=CONFIG
|
||||
),
|
||||
patch.object(
|
||||
mcp_server, "_authenticated_username", return_value="tester"
|
||||
),
|
||||
patch.object(
|
||||
mcp_server, "_ensure_matching_profile", return_value=None
|
||||
),
|
||||
patch.object(
|
||||
mcp_server, "init_review_decision_lock", return_value=None
|
||||
),
|
||||
]
|
||||
if required_role is not None:
|
||||
patches.append(
|
||||
patch.object(
|
||||
mcp_server.task_capability_map,
|
||||
"required_role",
|
||||
side_effect=lambda t: (
|
||||
required_role
|
||||
if t == task
|
||||
else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
|
||||
),
|
||||
)
|
||||
)
|
||||
for p in patches:
|
||||
p.__enter__()
|
||||
try:
|
||||
return mcp_server.gitea_resolve_task_capability(
|
||||
task=task, remote="prgs"
|
||||
)
|
||||
finally:
|
||||
for p in reversed(patches):
|
||||
p.__exit__(None, None, None)
|
||||
|
||||
|
||||
class TestAC3DeniedResolveDoesNotStampRole(_ResolveHarness):
|
||||
def test_allowed_resolve_stamps_role(self):
|
||||
result = self._resolve("review_pr")
|
||||
self.assertTrue(result["allowed_in_current_session"], result)
|
||||
self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
|
||||
self.assertEqual(mcp_server._preflight_resolved_task, "review_pr")
|
||||
|
||||
def test_denied_resolve_does_not_stamp_required_role(self):
|
||||
"""The PR #721 poison: review_pr requiring merger under a reviewer."""
|
||||
result = self._resolve("review_pr", required_role="merger")
|
||||
self.assertFalse(result["allowed_in_current_session"], result)
|
||||
self.assertIsNone(
|
||||
mcp_server._preflight_resolved_role,
|
||||
"denied resolve must not stamp the required role (#723 AC3)",
|
||||
)
|
||||
self.assertIsNone(mcp_server._preflight_resolved_task)
|
||||
|
||||
def test_denied_resolve_clears_prior_stale_stamp(self):
|
||||
allowed = self._resolve("review_pr")
|
||||
self.assertTrue(allowed["allowed_in_current_session"])
|
||||
self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
|
||||
|
||||
denied = self._resolve("merge_pr") # reviewer profile cannot merge
|
||||
self.assertFalse(denied["allowed_in_current_session"], denied)
|
||||
self.assertIsNone(
|
||||
mcp_server._preflight_resolved_role,
|
||||
"a denied resolve must clear the previous stamp, not keep it",
|
||||
)
|
||||
|
||||
def test_denied_resolve_keeps_effective_role_on_actual_profile(self):
|
||||
with patch.object(
|
||||
mcp_server, "get_profile", return_value=REVIEWER_PROFILE
|
||||
):
|
||||
self._resolve("review_pr", required_role="merger")
|
||||
self.assertEqual(
|
||||
mcp_server._effective_workspace_role(), "reviewer"
|
||||
)
|
||||
|
||||
|
||||
class TestAC4SubmissionFailsClosedStructured(unittest.TestCase):
|
||||
def setUp(self):
|
||||
_reset_preflight()
|
||||
|
||||
def tearDown(self):
|
||||
_reset_preflight()
|
||||
|
||||
def test_workspace_binding_failure_returns_reasons_not_raise(self):
|
||||
boom = RuntimeError(
|
||||
"namespace workspace binding blocked: role 'merger' cannot "
|
||||
"mutate from reviewer session workspace"
|
||||
)
|
||||
with patch.object(
|
||||
mcp_server,
|
||||
"_verify_role_mutation_workspace",
|
||||
side_effect=boom,
|
||||
), patch.object(
|
||||
mcp_server, "get_profile", return_value=REVIEWER_PROFILE
|
||||
):
|
||||
result = mcp_server._evaluate_pr_review_submission(
|
||||
pr_number=721,
|
||||
action="approve",
|
||||
expected_head_sha="80f59b334e6671b08006725292c08a8e8b6c823f",
|
||||
remote="prgs",
|
||||
live=True,
|
||||
final_review_decision_ready=True,
|
||||
)
|
||||
self.assertFalse(result["performed"])
|
||||
self.assertEqual(result["blocker_kind"], "workspace_role_binding")
|
||||
self.assertTrue(
|
||||
any("workspace/role binding failed" in r for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
self.assertTrue(
|
||||
any("cannot mutate from reviewer session" in r for r in result["reasons"]),
|
||||
"the underlying binding error text must be preserved",
|
||||
)
|
||||
|
||||
def test_stale_runtime_failure_also_structured(self):
|
||||
boom = RuntimeError(
|
||||
"stale-runtime: The active Gitea MCP server process is stale"
|
||||
)
|
||||
with patch.object(
|
||||
mcp_server,
|
||||
"_verify_role_mutation_workspace",
|
||||
side_effect=boom,
|
||||
), patch.object(
|
||||
mcp_server, "get_profile", return_value=REVIEWER_PROFILE
|
||||
):
|
||||
result = mcp_server._evaluate_pr_review_submission(
|
||||
pr_number=721,
|
||||
action="approve",
|
||||
remote="prgs",
|
||||
live=True,
|
||||
)
|
||||
self.assertFalse(result["performed"])
|
||||
self.assertEqual(result["blocker_kind"], "workspace_role_binding")
|
||||
self.assertTrue(
|
||||
any("stale-runtime" in r for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
def test_dry_run_also_fails_closed_structured(self):
|
||||
boom = RuntimeError("binding blocked")
|
||||
with patch.object(
|
||||
mcp_server,
|
||||
"_verify_role_mutation_workspace",
|
||||
side_effect=boom,
|
||||
), patch.object(
|
||||
mcp_server, "get_profile", return_value=REVIEWER_PROFILE
|
||||
):
|
||||
result = mcp_server._evaluate_pr_review_submission(
|
||||
pr_number=721,
|
||||
action="approve",
|
||||
remote="prgs",
|
||||
live=False,
|
||||
)
|
||||
self.assertFalse(result["would_perform"])
|
||||
self.assertEqual(result["blocker_kind"], "workspace_role_binding")
|
||||
|
||||
|
||||
class TestAC5RuntimeContextRoleFilter(unittest.TestCase):
|
||||
def test_role_filtered_view_matches_resolver_denial(self):
|
||||
"""Reviewer session: merge_pr stays denied under the role filter even
|
||||
when the permission is (pathologically) present."""
|
||||
allowed_ops = [
|
||||
"gitea.read",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
"gitea.pr.merge",
|
||||
]
|
||||
caps = mcp_server._build_runtime_task_capabilities(
|
||||
allowed_ops, [], CONFIG, active_role_kind="reviewer"
|
||||
)
|
||||
merge_entry = next(
|
||||
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
|
||||
)
|
||||
self.assertTrue(merge_entry["role_exclusive"])
|
||||
self.assertEqual(merge_entry["capability_view"], "role_filtered")
|
||||
self.assertFalse(
|
||||
merge_entry["allowed_in_current_session"],
|
||||
"role-exclusive merge_pr must stay denied for a reviewer role "
|
||||
"even when the permission is present (#723 AC5)",
|
||||
)
|
||||
self.assertFalse(caps["can_merge_prs"])
|
||||
|
||||
def test_role_filter_restricts_matching_profiles(self):
|
||||
caps = mcp_server._build_runtime_task_capabilities(
|
||||
["gitea.read"], [], CONFIG, active_role_kind="author"
|
||||
)
|
||||
review_entry = next(
|
||||
t for t in caps["task_capabilities"] if t["task"] == "review_pr"
|
||||
)
|
||||
self.assertEqual(
|
||||
review_entry["matching_configured_profiles"],
|
||||
["prgs-reviewer"],
|
||||
"role-exclusive review_pr must not list the merger profile",
|
||||
)
|
||||
merge_entry = next(
|
||||
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
|
||||
)
|
||||
self.assertEqual(
|
||||
merge_entry["matching_configured_profiles"], ["prgs-merger"]
|
||||
)
|
||||
|
||||
def test_permission_only_view_is_labeled(self):
|
||||
caps = mcp_server._build_runtime_task_capabilities(
|
||||
["gitea.pr.merge", "gitea.read"], [], CONFIG
|
||||
)
|
||||
merge_entry = next(
|
||||
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
|
||||
)
|
||||
self.assertEqual(merge_entry["capability_view"], "permission_only")
|
||||
# Legacy permission-only semantics preserved when no role supplied.
|
||||
self.assertTrue(merge_entry["allowed_in_current_session"])
|
||||
|
||||
def test_incident_722_shape_runtime_context_agrees_with_resolver(self):
|
||||
"""Post-970e68b shape: review_pr requires merger; a reviewer session
|
||||
must see review_pr denied in runtime context, matching the resolver."""
|
||||
with patch.object(
|
||||
mcp_server.task_capability_map,
|
||||
"required_role",
|
||||
side_effect=lambda t: (
|
||||
"merger"
|
||||
if t == "review_pr"
|
||||
else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
|
||||
),
|
||||
):
|
||||
caps = mcp_server._build_runtime_task_capabilities(
|
||||
REVIEWER_PROFILE["allowed_operations"],
|
||||
REVIEWER_PROFILE["forbidden_operations"],
|
||||
CONFIG,
|
||||
active_role_kind="reviewer",
|
||||
)
|
||||
review_entry = next(
|
||||
t for t in caps["task_capabilities"] if t["task"] == "review_pr"
|
||||
)
|
||||
self.assertFalse(
|
||||
review_entry["allowed_in_current_session"],
|
||||
"runtime context must not report review_pr allowed when the "
|
||||
"resolver would fail-close it (incident #722)",
|
||||
)
|
||||
self.assertFalse(caps["can_review_prs"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -111,13 +111,21 @@ class TestMcpStaleRuntime(unittest.TestCase):
|
||||
reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
|
||||
self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
|
||||
|
||||
def test_auto_restart_helper_removed_from_read_only_path(self):
|
||||
"""#685: config-touch / os._exit self-recovery is no longer on the server."""
|
||||
self.assertFalse(
|
||||
hasattr(gitea_mcp_server, "_trigger_mcp_auto_restart"),
|
||||
"_trigger_mcp_auto_restart must not remain (side-effect-free resolver)",
|
||||
)
|
||||
self.assertFalse(hasattr(gitea_mcp_server, "_restart_triggered"))
|
||||
@patch("threading.Thread")
|
||||
@patch("os.utime")
|
||||
@patch("os.path.exists")
|
||||
@patch.dict("os.environ", {"MCP_CONFIG_PATH": "/tmp/mcp_config.json"})
|
||||
def test_auto_restart_trigger_touches_and_spawns(self, mock_exists, mock_utime, mock_thread):
|
||||
mock_exists.return_value = True
|
||||
gitea_mcp_server._restart_triggered = False
|
||||
|
||||
# Ensure we are not skipped in test mode for testing purposes
|
||||
with patch("gitea_mcp_server._preflight_in_test_mode", return_value=False):
|
||||
gitea_mcp_server._trigger_mcp_auto_restart()
|
||||
|
||||
mock_utime.assert_called_once_with("/tmp/mcp_config.json", None)
|
||||
mock_thread.assert_called_once()
|
||||
self.assertTrue(gitea_mcp_server._restart_triggered)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -19,7 +19,12 @@ import unittest
|
||||
|
||||
import gitea_config
|
||||
from role_session_router import MERGER_TASKS, REVIEWER_TASKS
|
||||
from task_capability_map import required_permission, required_role
|
||||
from task_capability_map import (
|
||||
ROLE_EXCLUSIVE_TASKS,
|
||||
TASK_CAPABILITY_MAP,
|
||||
required_permission,
|
||||
required_role,
|
||||
)
|
||||
|
||||
# Canonical role-profile permission shape. Mirrors the configured
|
||||
# author/reviewer/merger/reconciler profiles (profiles.json v2 role split):
|
||||
@@ -201,5 +206,44 @@ class TestMergerBoundary(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestRoleExclusiveSetIntegrity(unittest.TestCase):
|
||||
"""#723 AC1/AC5: the shared role-exclusive set stays coherent."""
|
||||
|
||||
def test_every_role_exclusive_task_exists_in_capability_map(self):
|
||||
for task in sorted(ROLE_EXCLUSIVE_TASKS):
|
||||
with self.subTest(task=task):
|
||||
self.assertIn(
|
||||
task,
|
||||
TASK_CAPABILITY_MAP,
|
||||
f"role-exclusive task {task!r} missing from the "
|
||||
f"capability map — required_role() would raise and the "
|
||||
f"resolver would 500 instead of failing closed",
|
||||
)
|
||||
|
||||
def test_formal_review_tasks_are_role_exclusive(self):
|
||||
for task in FORMAL_REVIEW_TASKS:
|
||||
with self.subTest(task=task):
|
||||
self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
|
||||
|
||||
def test_merge_pr_is_role_exclusive_and_merger_satisfiable(self):
|
||||
"""AC1 merger equivalent: merging must stay possible for a canonical
|
||||
merger profile (permission AND role together)."""
|
||||
self.assertIn("merge_pr", ROLE_EXCLUSIVE_TASKS)
|
||||
self.assertTrue(
|
||||
_profile_satisfies("merger", "merge_pr"),
|
||||
"no canonical merger profile satisfies merge_pr — merging would "
|
||||
"be impossible for every configured profile",
|
||||
)
|
||||
|
||||
def test_reconciler_cleanup_tasks_stay_reconciler_satisfiable(self):
|
||||
for task in ("cleanup_merged_pr_branch", "reconciliation_cleanup"):
|
||||
with self.subTest(task=task):
|
||||
self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
|
||||
self.assertTrue(
|
||||
_profile_satisfies("reconciler", task),
|
||||
f"no canonical reconciler profile satisfies {task!r}",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user