"""Integration tests for FastMCP server using best practices."""
import asyncio
from datetime import datetime
from unittest.mock import patch
import pytest
from fastmcp import Client
from catabus_mcp.server import mcp
class TestMCPIntegration:
"""End-to-end MCP integration tests."""
@pytest.mark.asyncio
async def test_server_initialization(self):
"""Test that server initializes properly with mocked data."""
with patch("catabus_mcp.server.ensure_initialized") as mock_init:
mock_init.return_value = None
client = Client(mcp)
async with client:
# Server should connect without errors
tools = await client.list_tools()
tool_names = [tool.name for tool in tools]
expected_tools = [
"list_routes_tool",
"search_stops_tool",
"next_arrivals_tool",
"vehicle_positions_tool",
"trip_alerts_tool",
"health_check",
]
for tool in expected_tools:
assert tool in tool_names
@pytest.mark.asyncio
async def test_tool_schemas(self):
"""Test that all tools have proper input/output schemas."""
client = Client(mcp)
async with client:
tools = await client.list_tools()
# Check search_stops_tool has required query parameter
search_tool = next(t for t in tools if t.name == "search_stops_tool")
assert "query" in str(search_tool.inputSchema)
# Check next_arrivals_tool has stop_id parameter
arrivals_tool = next(t for t in tools if t.name == "next_arrivals_tool")
assert "stop_id" in str(arrivals_tool.inputSchema)
# Check vehicle_positions_tool has route_id parameter
vehicles_tool = next(t for t in tools if t.name == "vehicle_positions_tool")
assert "route_id" in str(vehicles_tool.inputSchema)
class TestRealWorldScenarios:
"""Test real-world usage scenarios with proper mocking."""
@pytest.fixture
def mock_cata_data(self):
"""Mock real CATA data responses."""
return {
"routes": [
{
"route_id": "BL",
"short_name": "BL",
"long_name": "Blue Loop",
"color": "#0000FF",
},
{
"route_id": "WL",
"short_name": "WL",
"long_name": "White Loop",
"color": "#FFFFFF",
},
{
"route_id": "N",
"short_name": "N",
"long_name": "Campus Loop North",
"color": "#00FF00",
},
],
"stops": [
{"stop_id": "HUB", "name": "HUB-Robeson Center", "lat": 40.7982, "lon": -77.8599},
{
"stop_id": "ATHERTON_CURTIN",
"name": "Atherton St at Curtin Rd",
"lat": 40.8012,
"lon": -77.8634,
},
],
"vehicles": [
{
"vehicle_id": "BUS_001",
"route_id": "BL",
"lat": 40.7982,
"lon": -77.8599,
"bearing": 90.0,
"speed_mps": 10.5,
}
],
}
@pytest.mark.asyncio
async def test_blue_loop_tracking_scenario(self, mock_cata_data):
"""Test complete Blue Loop tracking scenario."""
with patch("catabus_mcp.server.list_routes") as mock_list_routes:
with patch("catabus_mcp.server.vehicle_positions") as mock_vehicles:
with patch("catabus_mcp.server.initialized", True):
# Mock responses
mock_list_routes.return_value = mock_cata_data["routes"]
mock_vehicles.return_value = mock_cata_data["vehicles"]
client = Client(mcp)
async with client:
# Step 1: User asks for all routes
routes = await client.call_tool("list_routes_tool", {})
blue_loop = next((r for r in routes if r["short_name"] == "BL"), None)
assert blue_loop is not None
assert blue_loop["long_name"] == "Blue Loop"
# Step 2: User tracks Blue Loop vehicles
vehicles = await client.call_tool(
"vehicle_positions_tool", {"route_id": "BL"}
)
assert len(vehicles) == 1
assert vehicles[0]["vehicle_id"] == "BUS_001"
assert vehicles[0]["lat"] == 40.7982
@pytest.mark.asyncio
async def test_stop_search_and_arrivals_scenario(self, mock_cata_data):
"""Test searching stops and getting arrivals."""
with patch("catabus_mcp.server.search_stops") as mock_search:
with patch("catabus_mcp.server.next_arrivals") as mock_arrivals:
with patch("catabus_mcp.server.initialized", True):
# Mock responses
mock_search.return_value = mock_cata_data["stops"]
mock_arrivals.return_value = [
{
"trip_id": "BL_001",
"route_id": "BL",
"arrival_time_iso": "2024-01-01T09:15:00-05:00",
"delay_sec": 180,
}
]
client = Client(mcp)
async with client:
# Step 1: Search for HUB
stops = await client.call_tool("search_stops_tool", {"query": "HUB"})
hub_stop = next((s for s in stops if "HUB" in s["name"]), None)
assert hub_stop is not None
# Step 2: Get arrivals at HUB
arrivals = await client.call_tool(
"next_arrivals_tool",
{"stop_id": hub_stop["stop_id"], "horizon_minutes": 30},
)
assert len(arrivals) == 1
assert arrivals[0]["route_id"] == "BL"
assert arrivals[0]["delay_sec"] == 180 # 3 minutes late
class TestErrorHandlingBestPractices:
"""Test error handling following FastMCP best practices."""
@pytest.mark.asyncio
async def test_graceful_degradation(self):
"""Test server handles data loading failures gracefully."""
with patch("catabus_mcp.server.ensure_initialized") as mock_init:
# Simulate initialization failure
mock_init.side_effect = Exception("GTFS download failed")
client = Client(mcp)
async with client:
# Server should still respond but return empty data
try:
routes = await client.call_tool("list_routes_tool", {})
# Should return empty list rather than crash
assert isinstance(routes, list)
except Exception as e:
# Or handle error gracefully
assert "GTFS" in str(e)
@pytest.mark.asyncio
async def test_invalid_input_validation(self):
"""Test proper input validation."""
with patch("catabus_mcp.server.initialized", True):
client = Client(mcp)
async with client:
# Test missing required parameter
with pytest.raises(Exception):
await client.call_tool("search_stops_tool", {})
# Test invalid parameter type
with pytest.raises(Exception):
await client.call_tool(
"next_arrivals_tool",
{
"stop_id": 123, # Should be string
"horizon_minutes": "invalid", # Should be int
},
)
class TestPerformanceBestPractices:
"""Performance tests following FastMCP best practices."""
@pytest.mark.asyncio
async def test_concurrent_tool_calls(self):
"""Test server handles concurrent requests properly."""
with patch("catabus_mcp.server.initialized", True):
with patch("catabus_mcp.server.list_routes") as mock_routes:
with patch("catabus_mcp.server.search_stops") as mock_search:
with patch("catabus_mcp.server.vehicle_positions") as mock_vehicles:
# Mock quick responses
mock_routes.return_value = [
{
"route_id": "BL",
"short_name": "BL",
"long_name": "Blue Loop",
"color": "#0000FF",
}
]
mock_search.return_value = [
{
"stop_id": "HUB",
"name": "HUB-Robeson Center",
"lat": 40.7982,
"lon": -77.8599,
}
]
mock_vehicles.return_value = []
client = Client(mcp)
async with client:
# Make concurrent requests
tasks = [
client.call_tool("list_routes_tool", {}),
client.call_tool("search_stops_tool", {"query": "HUB"}),
client.call_tool("vehicle_positions_tool", {"route_id": "BL"}),
client.call_tool("health_check", {}),
]
# All should complete successfully
results = await asyncio.gather(*tasks, return_exceptions=True)
# Check no exceptions occurred
for result in results:
assert not isinstance(
result, Exception
), f"Request failed: {result}"
@pytest.mark.asyncio
async def test_response_time_reasonable(self):
"""Test that responses come back in reasonable time."""
with patch("catabus_mcp.server.initialized", True):
with patch("catabus_mcp.server.list_routes") as mock_routes:
mock_routes.return_value = [
{
"route_id": "BL",
"short_name": "BL",
"long_name": "Blue Loop",
"color": "#0000FF",
}
]
client = Client(mcp)
async with client:
start_time = datetime.now()
await client.call_tool("list_routes_tool", {})
end_time = datetime.now()
# Should complete in under 1 second for in-memory operations
duration = (end_time - start_time).total_seconds()
assert duration < 1.0, f"Response took too long: {duration}s"
class TestDataConsistency:
"""Test data consistency and integrity."""
@pytest.mark.asyncio
async def test_route_data_consistency(self):
"""Test that route data is consistent across tools."""
mock_routes = [
{"route_id": "BL", "short_name": "BL", "long_name": "Blue Loop", "color": "#0000FF"}
]
mock_vehicles = [
{
"vehicle_id": "BUS_001",
"lat": 40.7982,
"lon": -77.8599,
"bearing": 90.0,
"speed_mps": 10.5,
}
]
with patch("catabus_mcp.server.initialized", True):
with patch("catabus_mcp.server.list_routes", return_value=mock_routes):
with patch("catabus_mcp.server.vehicle_positions", return_value=mock_vehicles):
client = Client(mcp)
async with client:
# Get routes
routes = await client.call_tool("list_routes_tool", {})
bl_route = next((r for r in routes if r["route_id"] == "BL"), None)
assert bl_route is not None
# Get vehicles for same route - should be consistent
vehicles = await client.call_tool(
"vehicle_positions_tool", {"route_id": "BL"}
)
# Vehicles exist for this route
assert len(vehicles) == 1
