test(mcp): drive the openapi health check tests through real spec files

The added tests replaced the manager's own lookup and client factory with mocks, which the
repo guidance calls an anti-pattern, and patched litellm internals to stand in for the spec
load, which the test quality gate counts against its budget

They now seed config_mcp_servers and point spec_path at a real file on disk, so nothing is
faked. url is a closed port, so a regression that opens an MCP session instead reports
unhealthy and the tests fail
This commit is contained in:
ruoyu-lu 2026-09-07 23:57:13 +10:00
parent 6d1c9b1f7b
commit 229f1fb6d9

View file

@ -4494,86 +4494,73 @@ class TestMCPServerManager:
assert "OAuth discovery unavailable" in (result.health_check_error or "")
@pytest.mark.asyncio
async def test_health_check_server_openapi_spec_is_probed_instead_of_mcp_session(self):
async def test_health_check_server_openapi_spec_is_probed_instead_of_mcp_session(self, tmp_path):
"""An OpenAPI server is healthy when its spec loads, and is never asked to speak MCP.
Regression for #40079: these servers store the REST base url in ``url`` and only look like
http transport, so opening an MCP session against them always failed with "Session
terminated" even though every tool had loaded from the spec.
terminated" even though every tool had loaded from the spec. ``url`` is a closed port, so
an MCP handshake would report unhealthy.
"""
spec = tmp_path / "openapi.json"
spec.write_text(json.dumps({"openapi": "3.0.0", "paths": {}}))
manager = MCPServerManager()
server = MCPServer(
server_id="openapi-server",
name="openapi-server",
transport=MCPTransport.http,
auth_type=MCPAuth.none,
url="http://127.0.0.1:1",
spec_path=str(spec),
)
manager.config_mcp_servers[server.server_id] = server
result = await manager.health_check_server("openapi-server")
assert result.status == "healthy"
assert result.health_check_error is None
@pytest.mark.asyncio
async def test_health_check_server_openapi_spec_failure_is_unhealthy(self, tmp_path):
"""A spec that stopped loading is unhealthy, and the upstream error is what gets reported."""
missing = tmp_path / "openapi.json"
manager = MCPServerManager()
server = MCPServer(
server_id="openapi-server",
name="openapi-server",
transport=MCPTransport.http,
auth_type=MCPAuth.none,
url="http://127.0.0.1:1",
spec_path=str(missing),
)
manager.config_mcp_servers[server.server_id] = server
result = await manager.health_check_server("openapi-server")
assert result.status == "unhealthy"
assert result.health_check_error == f"OpenAPI spec not found at {missing}"
@pytest.mark.asyncio
async def test_health_check_server_openapi_spec_keeps_per_user_auth_skip(self, tmp_path):
"""Per-user auth still wins: fetching the spec userless could 401 and report a false failure.
The spec file is absent, so probing it would have reported unhealthy rather than unknown.
"""
manager = MCPServerManager()
server = MCPServer(
server_id="openapi-server",
name="openapi-server",
transport=MCPTransport.http,
auth_type=MCPAuth.none,
url="http://rest.example.com",
spec_path="http://rest.example.com/openapi.json",
)
manager.get_mcp_server_by_id = MagicMock(return_value=server)
manager._create_mcp_client = AsyncMock()
with patch(
"litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.load_openapi_spec_async",
new=AsyncMock(return_value={"openapi": "3.0.0", "paths": {}}),
) as mock_load:
result = await manager.health_check_server("openapi-server")
assert result.status == "healthy"
assert result.health_check_error is None
mock_load.assert_awaited_once_with("http://rest.example.com/openapi.json")
manager._create_mcp_client.assert_not_called()
@pytest.mark.asyncio
async def test_health_check_server_openapi_spec_failure_is_unhealthy(self):
"""A spec that stopped loading is unhealthy, and the upstream error is what gets reported."""
manager = MCPServerManager()
server = MCPServer(
server_id="openapi-server",
name="openapi-server",
transport=MCPTransport.http,
auth_type=MCPAuth.none,
url="http://rest.example.com",
spec_path="http://rest.example.com/openapi.json",
)
manager.get_mcp_server_by_id = MagicMock(return_value=server)
manager._create_mcp_client = AsyncMock()
with patch(
"litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.load_openapi_spec_async",
new=AsyncMock(side_effect=httpx.ConnectError("Cannot connect to host rest.example.com")),
):
result = await manager.health_check_server("openapi-server")
assert result.status == "unhealthy"
assert result.health_check_error == "Cannot connect to host rest.example.com"
manager._create_mcp_client.assert_not_called()
@pytest.mark.asyncio
async def test_health_check_server_openapi_spec_keeps_per_user_auth_skip(self):
"""Per-user auth still wins: fetching the spec userless could 401 and report a false failure."""
manager = MCPServerManager()
server = MCPServer(
server_id="openapi-server",
name="openapi-server",
transport=MCPTransport.http,
auth_type=MCPAuth.bearer_token,
url="http://rest.example.com",
spec_path="http://rest.example.com/openapi.json",
url="http://127.0.0.1:1",
spec_path=str(tmp_path / "openapi.json"),
)
manager.get_mcp_server_by_id = MagicMock(return_value=server)
manager._create_mcp_client = AsyncMock()
manager.config_mcp_servers[server.server_id] = server
with patch(
"litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.load_openapi_spec_async",
new=AsyncMock(return_value={"openapi": "3.0.0", "paths": {}}),
) as mock_load:
result = await manager.health_check_server("openapi-server")
result = await manager.health_check_server("openapi-server")
assert result.status == "unknown"
assert result.health_check_error is None
mock_load.assert_not_called()
manager._create_mcp_client.assert_not_called()
@pytest.mark.asyncio
async def test_health_check_server_not_found(self):