"""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 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 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"}) 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"} ) 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)