207 lines
7.8 KiB
Python
207 lines
7.8 KiB
Python
"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
|
|
import sys
|
|
import unittest
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import native_mcp_preference as nmp # noqa: E402
|
|
|
|
|
|
class TestShellHealthCircuitBreaker(unittest.TestCase):
|
|
def setUp(self):
|
|
nmp.clear_shell_health_for_tests()
|
|
|
|
def test_two_spawn_failures_hard_stop(self):
|
|
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
|
status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
|
self.assertTrue(status["hard_stopped"])
|
|
self.assertFalse(status["shell_use_allowed"])
|
|
self.assertEqual(status["consecutive_spawn_failures"], 2)
|
|
|
|
def test_success_resets_breaker(self):
|
|
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
|
status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="")
|
|
self.assertFalse(status["hard_stopped"])
|
|
self.assertEqual(status["consecutive_spawn_failures"], 0)
|
|
|
|
|
|
class TestTerminalDiagnostics(unittest.TestCase):
|
|
def test_diagnose_missing_cwd(self):
|
|
res = nmp.diagnose_terminal_failure("/nonexistent_directory_for_test", ["git", "status"])
|
|
self.assertFalse(res["cwd_exists"])
|
|
self.assertEqual(res["error_type"], "missing cwd")
|
|
self.assertIn("does not exist", res["error_msg"])
|
|
|
|
def test_diagnose_cwd_is_not_dir(self):
|
|
import tempfile
|
|
import os
|
|
with tempfile.NamedTemporaryFile() as tmp:
|
|
res = nmp.diagnose_terminal_failure(tmp.name, ["git", "status"])
|
|
self.assertFalse(res["cwd_is_dir"])
|
|
self.assertEqual(res["error_type"], "cwd is not a directory")
|
|
|
|
def test_diagnose_missing_executable(self):
|
|
res = nmp.diagnose_terminal_failure(None, ["nonexistent_exec_for_test"])
|
|
self.assertFalse(res["executable_exists"])
|
|
self.assertEqual(res["error_type"], "missing executable")
|
|
|
|
def test_diagnose_missing_runtime_wrapper(self):
|
|
res = nmp.diagnose_terminal_failure("/tmp", ["venv/bin/pytest"])
|
|
self.assertFalse(res["executable_exists"])
|
|
self.assertEqual(res["error_type"], "missing runtime wrapper")
|
|
|
|
def test_diagnose_relative_executable_from_cwd(self):
|
|
import os
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
script = Path(tmp) / "tool"
|
|
script.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
os.chmod(script, 0o755)
|
|
|
|
res = nmp.diagnose_terminal_failure(tmp, ["./tool"])
|
|
self.assertTrue(res["executable_exists"])
|
|
self.assertEqual(res["resolved_executable"], str(script))
|
|
|
|
def test_probe_terminal_spawn_success(self):
|
|
res = nmp.probe_terminal_spawn()
|
|
self.assertTrue(res["healthy"])
|
|
self.assertIn("command", res)
|
|
self.assertIn("diagnostics", res)
|
|
|
|
def test_probe_terminal_spawn_missing_cwd(self):
|
|
res = nmp.probe_terminal_spawn("/nonexistent_directory_for_test")
|
|
self.assertFalse(res["healthy"])
|
|
self.assertEqual(res["error_type"], "missing cwd")
|
|
|
|
def test_probe_terminal_spawn_honors_custom_command(self):
|
|
res = nmp.probe_terminal_spawn(
|
|
command=[sys.executable, "-c", "raise SystemExit(3)"]
|
|
)
|
|
self.assertTrue(res["healthy"])
|
|
self.assertEqual(res["command"][0], sys.executable)
|
|
self.assertEqual(res["exit_code"], 3)
|
|
|
|
@unittest.skipUnless(shutil.which("git"), "git is required for git -C probe")
|
|
def test_probe_terminal_spawn_supports_explicit_git_c_worktree(self):
|
|
import subprocess
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
subprocess.run(
|
|
["git", "init", tmp],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
res = nmp.probe_terminal_spawn(
|
|
cwd="/",
|
|
command=["git", "-C", tmp, "status", "--short"],
|
|
)
|
|
|
|
self.assertTrue(res["healthy"])
|
|
self.assertEqual(res["command"][1], "-C")
|
|
|
|
def test_probe_terminal_spawn_blocks_missing_custom_command(self):
|
|
res = nmp.probe_terminal_spawn(command=["nonexistent_exec_for_test"])
|
|
self.assertFalse(res["healthy"])
|
|
self.assertEqual(res["error_type"], "missing executable")
|
|
self.assertEqual(res["command"], ["nonexistent_exec_for_test"])
|
|
|
|
|
|
class TestNativeMcpPreferenceGate(unittest.TestCase):
|
|
def setUp(self):
|
|
nmp.clear_shell_health_for_tests()
|
|
|
|
def test_mcp_available_blocks_local_script(self):
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="create_pr",
|
|
command_or_detail="python create_pr.py --remote prgs --title T --head h",
|
|
mcp_available=True,
|
|
mcp_tool_visible=True,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["path_kind"], "shell_script")
|
|
|
|
def test_mcp_native_path_allowed(self):
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="comment_issue",
|
|
path_kind="mcp_native",
|
|
mcp_available=True,
|
|
mcp_tool_visible=True,
|
|
)
|
|
self.assertFalse(result["block"])
|
|
self.assertTrue(result["allowed"])
|
|
|
|
def test_mcp_unavailable_requires_terminal_report(self):
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="merge_pr",
|
|
path_kind="shell_script",
|
|
mcp_available=False,
|
|
mcp_tool_visible=False,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertTrue(
|
|
any("terminal recovery report" in r for r in result["reasons"])
|
|
)
|
|
self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"])
|
|
|
|
def test_recovery_fallback_allowed_with_proof(self):
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="merge_pr",
|
|
path_kind="shell_script",
|
|
command_or_detail="Recovery mode: MCP tools unavailable in this session.",
|
|
mcp_available=False,
|
|
recovery_mode=True,
|
|
recovery_proof_complete=True,
|
|
)
|
|
self.assertFalse(result["block"])
|
|
|
|
def test_cli_auth_divergence_blocked(self):
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="create_pr",
|
|
command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py",
|
|
mcp_available=True,
|
|
active_profile="prgs-author",
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"]))
|
|
|
|
def test_unsafe_helper_fallback_denied(self):
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="commit_files",
|
|
command_or_detail="python _encode_payload.py",
|
|
mcp_available=True,
|
|
mcp_tool_visible=True,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["path_kind"], "unsafe_helper")
|
|
|
|
def test_reviewer_mcp_server_touch_blocked(self):
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="review_pr",
|
|
command_or_detail="pkill -f gitea_mcp_server.py",
|
|
mcp_available=True,
|
|
role="reviewer",
|
|
active_profile="prgs-reviewer",
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["path_kind"], "mcp_server_touch")
|
|
|
|
def test_hard_stopped_shell_blocks_non_mcp_path(self):
|
|
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
|
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
|
result = nmp.assess_gitea_operation_path(
|
|
task="lock_issue",
|
|
path_kind="shell_script",
|
|
mcp_available=True,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertTrue(any("circuit breaker" in r for r in result["reasons"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|