b646440ca0
Salvaged from PR #977 onto current main. Preserves the MCP stdio command resolution and improved error diagnostics, with deterministic regression tests for the npx/node PATH cases. Co-authored-by: kshitij <82637225+kshitijk4poor@users.noreply.github.com>
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
import asyncio
|
|
import os
|
|
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
|
|
|
|
|
|
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())
|