6367e1c4c0
Bug fixes: - agent/redact.py: catastrophic regex backtracking in _ENV_ASSIGN_RE — removed re.IGNORECASE and changed [A-Z_]* to [A-Z0-9_]* to restrict matching to actual env var name chars. Without this, the pattern backtracks exponentially on large strings (e.g. 100K tool output), causing test_file_read_guards to time out. - tools/file_operations.py: over-escaped newline in find -printf format string produced literal backslash-n instead of a real newline, breaking file search result parsing (total_count always 1, paths concatenated). Test fixes: - Remove stale pytestmark.skip from 4 test modules that were blanket-skipped as 'Hangs in non-interactive environments' but actually run fine: - test_413_compression.py (12 tests, 25s) - test_file_tools_live.py (71 tests, 24s) - test_code_execution.py (61 tests, 99s) - test_agent_loop_tool_calling.py (has proper OPENROUTER_API_KEY skip already) - test_413_compression.py: fix threshold values in 2 preflight compression tests where context_length was too small for the compressed output to fit in one pass. - test_mcp_probe.py: add missing _MCP_AVAILABLE mock so tests work without MCP SDK. - test_mcp_tool_issue_948.py: inject MCP symbols (StdioServerParameters etc.) when SDK is not installed so patch() targets exist. - test_approve_deny_commands.py: replace time.sleep(0.3) with deterministic polling of _gateway_queues — fixes race condition where resolve fires before threads register their approval entries, causing the test to hang indefinitely. Net effect: +256 tests recovered from skip, 8 real failures fixed.
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
import asyncio
|
|
import os
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from tools.mcp_tool import MCPServerTask, _format_connect_error, _resolve_stdio_command, _MCP_AVAILABLE
|
|
|
|
# Ensure the mcp module symbols exist for patching even when the SDK isn't installed
|
|
if not _MCP_AVAILABLE:
|
|
import tools.mcp_tool as _mcp_mod
|
|
if not hasattr(_mcp_mod, "StdioServerParameters"):
|
|
_mcp_mod.StdioServerParameters = MagicMock
|
|
if not hasattr(_mcp_mod, "stdio_client"):
|
|
_mcp_mod.stdio_client = MagicMock
|
|
if not hasattr(_mcp_mod, "ClientSession"):
|
|
_mcp_mod.ClientSession = MagicMock
|
|
|
|
|
|
def test_resolve_stdio_command_falls_back_to_hermes_node_bin(tmp_path):
|
|
node_bin = tmp_path / "node" / "bin"
|
|
node_bin.mkdir(parents=True)
|
|
npx_path = node_bin / "npx"
|
|
npx_path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
npx_path.chmod(0o755)
|
|
|
|
with patch("tools.mcp_tool.shutil.which", return_value=None), \
|
|
patch.dict("os.environ", {"HERMES_HOME": str(tmp_path)}, clear=False):
|
|
command, env = _resolve_stdio_command("npx", {"PATH": "/usr/bin"})
|
|
|
|
assert command == str(npx_path)
|
|
assert env["PATH"].split(os.pathsep)[0] == str(node_bin)
|
|
|
|
|
|
def test_resolve_stdio_command_respects_explicit_empty_path():
|
|
seen_paths = []
|
|
|
|
def _fake_which(_cmd, path=None):
|
|
seen_paths.append(path)
|
|
return None
|
|
|
|
with patch("tools.mcp_tool.shutil.which", side_effect=_fake_which):
|
|
command, env = _resolve_stdio_command("python", {"PATH": ""})
|
|
|
|
assert command == "python"
|
|
assert env["PATH"] == ""
|
|
assert seen_paths == [""]
|
|
|
|
|
|
def test_format_connect_error_unwraps_exception_group():
|
|
error = ExceptionGroup(
|
|
"unhandled errors in a TaskGroup",
|
|
[FileNotFoundError(2, "No such file or directory", "node")],
|
|
)
|
|
|
|
message = _format_connect_error(error)
|
|
|
|
assert "missing executable 'node'" in message
|
|
|
|
|
|
def test_run_stdio_uses_resolved_command_and_prepended_path(tmp_path):
|
|
node_bin = tmp_path / "node" / "bin"
|
|
node_bin.mkdir(parents=True)
|
|
npx_path = node_bin / "npx"
|
|
npx_path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
npx_path.chmod(0o755)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.initialize = AsyncMock()
|
|
mock_session.list_tools = AsyncMock(return_value=SimpleNamespace(tools=[]))
|
|
|
|
mock_stdio_cm = MagicMock()
|
|
mock_stdio_cm.__aenter__ = AsyncMock(return_value=(object(), object()))
|
|
mock_stdio_cm.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
mock_session_cm = MagicMock()
|
|
mock_session_cm.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session_cm.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
async def _test():
|
|
with patch("tools.mcp_tool.shutil.which", return_value=None), \
|
|
patch.dict("os.environ", {"HERMES_HOME": str(tmp_path), "PATH": "/usr/bin", "HOME": str(tmp_path)}, clear=False), \
|
|
patch("tools.mcp_tool.StdioServerParameters") as mock_params, \
|
|
patch("tools.mcp_tool.stdio_client", return_value=mock_stdio_cm), \
|
|
patch("tools.mcp_tool.ClientSession", return_value=mock_session_cm):
|
|
server = MCPServerTask("srv")
|
|
await server.start({"command": "npx", "args": ["-y", "pkg"], "env": {"PATH": "/usr/bin"}})
|
|
|
|
call_kwargs = mock_params.call_args.kwargs
|
|
assert call_kwargs["command"] == str(npx_path)
|
|
assert call_kwargs["env"]["PATH"].split(os.pathsep)[0] == str(node_bin)
|
|
|
|
await server.shutdown()
|
|
|
|
asyncio.run(_test())
|