"""Unit tests for navi/mcp/ infrastructure."""
from unittest.mock import AsyncMock
import pytest
from navi.mcp.client import McpClient
from navi.mcp.config import McpServerConfig, load_mcp_servers
from navi.mcp.manager import McpManager, summarize_instructions
from navi.mcp.tools import McpTool
class TestMcpServerConfig:
def test_stdio_config(self):
cfg = McpServerConfig(
transport="stdio",
command="python",
args=["-m", "app.mcp_server"],
env={"FOO": "bar"},
)
assert cfg.is_stdio
assert not cfg.is_sse
assert cfg.command == "python"
def test_sse_config(self):
cfg = McpServerConfig(
transport="sse",
url="http://localhost:3001/sse",
headers={"Authorization": "Bearer token"},
)
assert cfg.is_sse
assert not cfg.is_stdio
assert cfg.url == "http://localhost:3001/sse"
def test_default_transport_is_stdio(self):
cfg = McpServerConfig(command="npx")
assert cfg.transport == "stdio"
class TestLoadMcpServers:
def test_missing_file_returns_empty(self, tmp_path):
result = load_mcp_servers(tmp_path / "nonexistent.json")
assert result == {}
def test_loads_valid_json(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text(
'{"book": {"transport": "stdio", "command": "python", "args": ["server.py"]}}'
)
result = load_mcp_servers(path)
assert "book" in result
assert result["book"].command == "python"
class TestMcpManager:
async def test_load_all_with_empty_config(self):
manager = McpManager()
await manager.load_all({})
assert manager.clients == {}
async def test_disconnect_all_when_empty(self):
manager = McpManager()
await manager.disconnect_all() # should not raise
async def test_get_all_tools_skips_broken_server(self):
manager = McpManager()
client = AsyncMock(spec=McpClient)
client.connected = True
client.list_tools.side_effect = RuntimeError("boom")
manager._clients = {"bad": client}
tools = await manager.get_all_tools()
assert tools == []
def test_get_instructions_returns_selected_config_overlay_when_disconnected(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text(
'{"gnexus-book": {"transport": "sse", "url": "http://example/sse", "instructions": "Use book."}}'
)
manager = McpManager(config_path=path)
instructions = manager.get_instructions({"gnexus-book"})
assert instructions == {"gnexus-book": "Use book."}
def test_get_instructions_filters_to_selected_servers(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text(
'{"gnexus-book": {"instructions": "Use book."}, "other": {"instructions": "Use other."}}'
)
manager = McpManager(config_path=path)
instructions = manager.get_instructions({"gnexus-book"})
assert instructions == {"gnexus-book": "Use book."}
class TestServerSummaries:
"""The one line the system prompt keeps per server.
Instructions used to be inlined there in full — ~8 KB of prose on every request
of a heavy profile. Only the opening claim stays; the body is one
tool_manual("<server>") call away.
"""
def test_takes_the_first_sentence(self):
assert summarize_instructions("Use book. Then read it.") == "Use book."
def test_keeps_a_single_sentence_whole(self):
assert summarize_instructions("MCP tools for gntodo.") == "MCP tools for gntodo."
def test_flattens_newlines_so_it_stays_one_line(self):
assert summarize_instructions("Use book.\n\nThen\nread it.") == "Use book."
def test_caps_a_runaway_first_sentence(self):
summary = summarize_instructions("word " * 100)
assert len(summary) <= 201
assert summary.endswith("…")
def test_empty_text_has_no_summary(self):
assert summarize_instructions("") == ""
assert summarize_instructions(" \n ") == ""
def test_summaries_fall_back_to_the_instructions(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text('{"book": {"instructions": "Use book. Details follow."}}')
manager = McpManager(config_path=path)
assert manager.get_summaries({"book"}) == {"book": "Use book."}
def test_an_explicit_summary_wins(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text(
'{"book": {"summary": "The book.", "instructions": "MANDATORY: use book now."}}'
)
manager = McpManager(config_path=path)
assert manager.get_summaries({"book"}) == {"book": "The book."}
def test_a_server_with_nothing_to_say_is_not_listed(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text('{"book": {"transport": "stdio", "command": "python"}}')
manager = McpManager(config_path=path)
assert manager.get_summaries({"book"}) == {}
def test_summaries_are_ordered(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text(
'{"zeta": {"instructions": "Z."}, "alpha": {"instructions": "A."}}'
)
manager = McpManager(config_path=path)
assert list(manager.get_summaries({"zeta", "alpha"})) == ["alpha", "zeta"]
def test_configured_servers_lists_the_config_not_the_pool(self, tmp_path):
"""A disconnected server is exactly when its instructions are worth reading."""
path = tmp_path / "mcp_servers.json"
path.write_text('{"book": {"instructions": "Use book."}, "todo": {}}')
manager = McpManager(config_path=path)
assert manager.clients == {}
assert manager.configured_servers() == ["book", "todo"]
def test_server_tool_names_dedupes_across_groups(self, tmp_path):
path = tmp_path / "mcp_servers.json"
path.write_text(
'{"book": {"groups": {"read": ["search", "get"], "write": ["get", "set"]}}}'
)
manager = McpManager(config_path=path)
assert manager.server_tool_names("book") == ["get", "search", "set"]
assert manager.server_tool_names("nope") == []
class TestMcpTool:
def test_name_prefix(self):
mock_manager = AsyncMock(spec=McpManager)
tool = McpTool(
server_name="gnexus-book",
tool_name="search_docs",
description="Search docs",
parameters={"type": "object", "properties": {}},
manager=mock_manager,
)
assert tool.name == "mcp__gnexus-book__search_docs"
async def test_execute_success(self):
mock_manager = AsyncMock(spec=McpManager)
mock_manager.call_tool.return_value = ("found 3 results", False)
tool = McpTool(
server_name="book",
tool_name="search",
description="",
parameters={},
manager=mock_manager,
)
result = await tool.execute({"query": "foo"})
assert result.success
assert result.output == "found 3 results"
mock_manager.call_tool.assert_awaited_once_with(
"book", "search", {"query": "foo"}, user_id=None
)
async def test_execute_mcp_error(self):
mock_manager = AsyncMock(spec=McpManager)
mock_manager.call_tool.return_value = ("validation failed", True)
tool = McpTool(
server_name="book",
tool_name="apply",
description="",
parameters={},
manager=mock_manager,
)
result = await tool.execute({})
assert not result.success
assert result.output == "validation failed"
assert "MCP tool reported an error" in result.error
async def test_execute_failure(self):
mock_manager = AsyncMock(spec=McpManager)
mock_manager.call_tool.side_effect = RuntimeError("server down")
tool = McpTool(
server_name="book",
tool_name="search",
description="",
parameters={},
manager=mock_manager,
)
result = await tool.execute({})
assert not result.success
assert "server down" in result.error
async def test_execute_injects_session_id(self):
from navi.tools._internal.base import current_session_id
mock_manager = AsyncMock(spec=McpManager)
mock_manager.call_tool.return_value = ("ok", False)
tool = McpTool(
server_name="book",
tool_name="search",
description="",
parameters={},
manager=mock_manager,
)
token = current_session_id.set("real-session-id")
try:
result = await tool.execute({"query": "foo"})
assert result.success
mock_manager.call_tool.assert_awaited_once_with(
"book", "search", {"query": "foo", "session_id": "real-session-id"},
user_id=None,
)
finally:
current_session_id.reset(token)
async def test_execute_overwrites_hallucinated_session_id(self):
from navi.tools._internal.base import current_session_id
mock_manager = AsyncMock(spec=McpManager)
mock_manager.call_tool.return_value = ("ok", False)
tool = McpTool(
server_name="book",
tool_name="search",
description="",
parameters={},
manager=mock_manager,
)
token = current_session_id.set("real-session-id")
try:
result = await tool.execute({"session_id": "fake-id", "query": "foo"})
assert result.success
args = mock_manager.call_tool.call_args[0][2]
assert args["session_id"] == "real-session-id"
finally:
current_session_id.reset(token)
async def test_execute_normalizes_navi3d_paths(self):
from navi.tools._internal.base import current_session_id
mock_manager = AsyncMock(spec=McpManager)
mock_manager.call_tool.return_value = ("ok", False)
tool = McpTool(
server_name="navi-3d",
tool_name="compile_scad",
description="",
parameters={},
manager=mock_manager,
)
token = current_session_id.set("real-session-id")
try:
result = await tool.execute({
"source_path": "session_files/real-session-id/falcon9_rocket.scad",
"output_path": "session_files/real-session-id/falcon9_rocket.stl",
})
assert result.success
args = mock_manager.call_tool.call_args[0][2]
assert args["source_path"] == "falcon9_rocket.scad"
assert args["output_path"] == "falcon9_rocket.stl"
assert args["session_id"] == "real-session-id"
finally:
current_session_id.reset(token)
async def test_execute_does_not_modify_paths_for_non_navi3d(self):
from navi.tools._internal.base import current_session_id
mock_manager = AsyncMock(spec=McpManager)
mock_manager.call_tool.return_value = ("ok", False)
tool = McpTool(
server_name="other-server",
tool_name="do_stuff",
description="",
parameters={},
manager=mock_manager,
)
token = current_session_id.set("real-session-id")
try:
result = await tool.execute({"path": "some/nested/path.txt"})
assert result.success
args = mock_manager.call_tool.call_args[0][2]
assert args["path"] == "some/nested/path.txt"
assert args["session_id"] == "real-session-id"
finally:
current_session_id.reset(token)
class TestLoadAllTolerance:
async def test_load_all_survives_dead_streamable_http_endpoint(self):
"""Regression: a streamable_http transport connecting to a dead
endpoint raises CancelledError from inside the MCP SDK's anyio
cancel scopes — without any task.cancel(). The runner used to treat
that as "runner is being cancelled", cancelled the caller's future,
and McpManager.load_all (which only catches Exception) blew up the
whole server startup. Now the connect failure surfaces as a normal
exception: the server starts, the server is retried by the
health-check loop."""
manager = McpManager()
cfg = McpServerConfig(
transport="streamable_http",
url="http://127.0.0.1:9/mcp", # port 9 (discard) — nothing listens
)
# Before the fix this raised CancelledError out of load_all.
await manager.load_all({"dead": cfg})
assert "dead" in manager.clients
assert manager._connected_status["dead"] is False
await manager.disconnect_all()
async def test_runner_death_fails_pending_callers(self):
"""If the runner task dies unexpectedly, callers waiting in _send
get an error instead of hanging forever."""
import asyncio
client = McpClient(
"dead",
McpServerConfig(transport="streamable_http", url="http://127.0.0.1:9/mcp"),
)
# Poison the open path so the runner exits through the CancelledError
# branch with a real task cancellation: connect once (fails normally),
# then cancel the runner mid-connect and re-issue a connect that
# would outlive it.
with pytest.raises(Exception):
await client.connect()
# The first connect failed; the runner is still alive (it keeps the
# client for health-check retries). Kill it hard and make sure a
# subsequent command surfaces an error, not a hang.
runner = client._runner_task
assert runner is not None
runner.cancel()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(asyncio.shield(runner), timeout=5)
with pytest.raises(RuntimeError):
await asyncio.wait_for(client.connect(), timeout=5)
