"""Unit tests for OpenF1 MCP Server"""
import pytest
import json
from unittest.mock import AsyncMock, patch, MagicMock
from src.server import OpenF1MCPServer
from src.openf1_client import OpenF1Client
@pytest.fixture
def server():
"""Create a test server"""
return OpenF1MCPServer()
def test_server_initialization(server):
"""Test server is initialized correctly"""
assert server.server is not None
assert server.client is None
def test_get_tools(server):
"""Test that all tools are registered"""
tools = server.get_tools()
assert len(tools) == 12
tool_names = [tool.name for tool in tools]
assert "list_drivers" in tool_names
assert "list_teams" in tool_names
assert "list_races" in tool_names
assert "list_results" in tool_names
assert "list_sessions" in tool_names
assert "list_laps" in tool_names
assert "list_stints" in tool_names
assert "list_pit_stops" in tool_names
assert "get_weather" in tool_names
assert "list_incidents" in tool_names
assert "get_car_data" in tool_names
def test_tool_descriptions(server):
"""Test that all tools have descriptions"""
tools = server.get_tools()
for tool in tools:
assert tool.description is not None
assert len(tool.description) > 0
def test_tool_input_schemas(server):
"""Test that all tools have input schemas"""
tools = server.get_tools()
for tool in tools:
assert tool.inputSchema is not None
assert "type" in tool.inputSchema
assert "properties" in tool.inputSchema
def test_list_drivers_tool_schema(server):
"""Test list_drivers tool has correct schema"""
tools = server.get_tools()
drivers_tool = next(t for t in tools if t.name == "list_drivers")
schema = drivers_tool.inputSchema
assert "season" in schema["properties"]
assert "driver_number" in schema["properties"]
def test_list_races_tool_schema(server):
"""Test list_races tool has correct schema"""
tools = server.get_tools()
races_tool = next(t for t in tools if t.name == "list_races")
schema = races_tool.inputSchema
assert "season" in schema["properties"]
assert "round_number" in schema["properties"]
def test_list_sessions_tool_schema(server):
"""Test list_sessions tool has correct schema"""
tools = server.get_tools()
sessions_tool = next(t for t in tools if t.name == "list_sessions")
schema = sessions_tool.inputSchema
assert "season" in schema["properties"]
assert "round_number" in schema["properties"]
@pytest.mark.asyncio
async def test_run_tool_list_drivers(server):
"""Test running list_drivers tool"""
mock_response = [
{"driver_number": 1, "name": "Max Verstappen"}
]
with patch.object(OpenF1Client, 'get_drivers', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_drivers", {}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_drivers_with_params(server):
"""Test running list_drivers tool with parameters"""
mock_response = [
{"driver_number": 1, "name": "Max Verstappen", "season": 2024}
]
with patch.object(OpenF1Client, 'get_drivers', new_callable=AsyncMock, return_value=mock_response) as mock_get:
result = await server._run_tool("list_drivers", {"season": 2024, "driver_number": 1}, OpenF1Client())
mock_get.assert_called_once_with(2024, 1)
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_teams(server):
"""Test running list_teams tool"""
mock_response = [
{"team_id": 1, "name": "Red Bull Racing"}
]
with patch.object(OpenF1Client, 'get_teams', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_teams", {}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_races(server):
"""Test running list_races tool"""
mock_response = [
{"round": 1, "name": "Bahrain Grand Prix"}
]
with patch.object(OpenF1Client, 'get_races', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_races", {}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_results(server):
"""Test running list_results tool"""
mock_response = [
{"position": 1, "driver_number": 1, "points": 25}
]
with patch.object(OpenF1Client, 'get_results', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_results", {}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_sessions(server):
"""Test running list_sessions tool"""
mock_response = [
{"session_key": 1, "session_type": "Practice"}
]
with patch.object(OpenF1Client, 'get_sessions', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_sessions", {}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_laps(server):
"""Test running list_laps tool"""
mock_response = [
{"lap_number": 1, "driver_number": 1, "lap_duration": 90.5}
]
with patch.object(OpenF1Client, 'get_laps', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_laps", {"session_key": 1234}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_stints(server):
"""Test running list_stints tool"""
mock_response = [
{"stint_number": 1, "driver_number": 1, "tyre_compound": "SOFT"}
]
with patch.object(OpenF1Client, 'get_stints', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_stints", {"session_key": 1234}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_pit_stops(server):
"""Test running list_pit_stops tool"""
mock_response = [
{"pit_stop_number": 1, "driver_number": 1, "duration": 2.5}
]
with patch.object(OpenF1Client, 'get_pit_stops', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_pit_stops", {"session_key": 1234}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_get_weather(server):
"""Test running get_weather tool"""
mock_response = [
{"track_temperature": 28.5, "air_temperature": 25.2}
]
with patch.object(OpenF1Client, 'get_weather', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("get_weather", {"session_key": 1234}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_incidents(server):
"""Test running list_incidents tool"""
mock_response = [
{"incident_type": "collision", "driver_number": 1, "lap": 5}
]
with patch.object(OpenF1Client, 'get_incidents', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_incidents", {"session_key": 1234}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_get_car_data(server):
"""Test running get_car_data tool"""
mock_response = [
{"throttle": 95, "brake": 0, "drs": True}
]
with patch.object(OpenF1Client, 'get_car_data', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("get_car_data", {"session_key": 1234}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_list_positions(server):
"""Test running list_positions tool"""
mock_response = [
{"position": 1, "driver_number": 1, "lap": 1}
]
with patch.object(OpenF1Client, 'get_positions', new_callable=AsyncMock, return_value=mock_response):
result = await server._run_tool("list_positions", {"session_key": 1234}, OpenF1Client())
assert result == mock_response
@pytest.mark.asyncio
async def test_run_tool_unknown_tool(server):
"""Test running unknown tool raises error"""
with pytest.raises(ValueError, match="Unknown tool"):
await server._run_tool("unknown_tool", {}, OpenF1Client())
@pytest.mark.asyncio
async def test_execute_tool(server):
"""Test execute_tool method"""
mock_response = [
{"driver_number": 1, "name": "Max Verstappen"}
]
with patch.object(server, '_run_tool', new_callable=AsyncMock, return_value=mock_response):
result = await server._execute_tool("list_drivers", {})
assert result == mock_response
@pytest.mark.asyncio
async def test_execute_tool_with_existing_client(server):
"""Test execute_tool with existing client"""
server.client = OpenF1Client()
mock_response = [
{"driver_number": 1, "name": "Max Verstappen"}
]
with patch.object(server, '_run_tool', new_callable=AsyncMock, return_value=mock_response):
result = await server._execute_tool("list_drivers", {})
assert result == mock_response
