test_server.py•10.5 kB
"""Tests for SpaceX MCP server.py module."""
import pytest
import json
import sys
from unittest.mock import Mock, MagicMock, patch
from server import SpaceXMCPServer
@pytest.fixture
def server():
"""Create a SpaceXMCPServer instance for testing."""
return SpaceXMCPServer()
class TestServerInitialization:
"""Test server initialization."""
def test_server_initialization(self, server):
"""Verify server initializes with correct tools."""
assert len(server.tools) == 3
tool_names = [tool["name"] for tool in server.tools]
assert "get_latest_launch" in tool_names
assert "get_upcoming_launches" in tool_names
assert "get_company_info" in tool_names
for tool in server.tools:
assert "name" in tool
assert "description" in tool
assert "inputSchema" in tool
class TestHandleInitialize:
"""Test initialize request handling."""
def test_handle_initialize_request(self, server):
"""Verify initialize handshake works."""
request = {
"jsonrpc": "2.0",
"method": "initialize",
"id": 1,
"params": {}
}
response = server.handle_initialize(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 1
assert response["result"]["protocolVersion"] == "2024-11-05"
assert response["result"]["serverInfo"]["name"] == "spacex-mcp"
assert response["result"]["serverInfo"]["version"] == "1.0.0"
assert "capabilities" in response["result"]
assert "tools" in response["result"]["capabilities"]
class TestHandleInitialized:
"""Test initialized notification handling."""
def test_handle_initialized_notification(self, server):
"""Verify initialized notification returns None (no response)."""
request = {
"jsonrpc": "2.0",
"method": "initialized"
}
response = server.handle_request(request)
assert response is None
class TestHandleToolsList:
"""Test tools/list request handling."""
def test_handle_tools_list(self, server):
"""Verify tools/list returns all tools."""
request = {
"jsonrpc": "2.0",
"method": "tools/list",
"id": 2
}
response = server.handle_tools_list(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 2
assert "result" in response
assert "tools" in response["result"]
assert len(response["result"]["tools"]) == 3
assert response["result"]["tools"][0]["name"] == "get_latest_launch"
class TestHandleToolsCallGetLatestLaunch:
"""Test tools/call for get_latest_launch."""
def test_handle_tools_call_get_latest_launch_success(self, server, mocker):
"""Verify tools/call executes get_latest_launch."""
test_data = {
"flight_number": 100,
"name": "Test Mission",
"date_utc": "2025-01-15",
"success": True,
"details": "Test launch"
}
# Mock spacex_api.get_latest_launch
mock_get_latest = mocker.patch('server.spacex_api.get_latest_launch', return_value=test_data)
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 3,
"params": {
"name": "get_latest_launch",
"arguments": {}
}
}
response = server.handle_tools_call(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 3
assert "result" in response
assert "content" in response["result"]
assert len(response["result"]["content"]) == 1
assert response["result"]["content"][0]["type"] == "text"
assert "🚀" in response["result"]["content"][0]["text"]
mock_get_latest.assert_called_once()
def test_handle_tools_call_get_latest_launch_failure(self, server, mocker):
"""Verify error handling when API call fails."""
mock_get_latest = mocker.patch('server.spacex_api.get_latest_launch', return_value=None)
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 4,
"params": {
"name": "get_latest_launch",
"arguments": {}
}
}
response = server.handle_tools_call(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 4
assert "error" in response
assert response["error"]["code"] == -32000
assert response["error"]["message"] == "En son fırlatma verisi alınamadı"
class TestHandleToolsCallGetUpcomingLaunches:
"""Test tools/call for get_upcoming_launches."""
def test_handle_tools_call_get_upcoming_launches_with_limit(self, server, mocker):
"""Verify get_upcoming_launches with custom limit."""
test_data = [
{"flight_number": 101, "name": "Mission 1"},
{"flight_number": 102, "name": "Mission 2"}
]
mock_get_upcoming = mocker.patch(
'server.spacex_api.get_upcoming_launches',
return_value=test_data
)
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 5,
"params": {
"name": "get_upcoming_launches",
"arguments": {"limit": 10}
}
}
response = server.handle_tools_call(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 5
assert "result" in response
assert response["result"]["content"][0]["type"] == "text"
mock_get_upcoming.assert_called_once_with(10)
def test_handle_tools_call_get_upcoming_launches_default_limit(self, server, mocker):
"""Verify default limit of 5 is used."""
test_data = []
mock_get_upcoming = mocker.patch(
'server.spacex_api.get_upcoming_launches',
return_value=test_data
)
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 6,
"params": {
"name": "get_upcoming_launches",
"arguments": {}
}
}
response = server.handle_tools_call(request)
mock_get_upcoming.assert_called_once_with(5)
class TestHandleToolsCallGetCompanyInfo:
"""Test tools/call for get_company_info."""
def test_handle_tools_call_get_company_info_success(self, server, mocker):
"""Verify get_company_info returns data."""
test_data = {
"name": "SpaceX",
"founder": "Elon Musk",
"founded": 2002
}
mock_get_company = mocker.patch(
'server.spacex_api.get_company_info',
return_value=test_data
)
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 7,
"params": {
"name": "get_company_info",
"arguments": {}
}
}
response = server.handle_tools_call(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 7
assert "result" in response
assert response["result"]["content"][0]["type"] == "text"
assert "SpaceX" in response["result"]["content"][0]["text"]
mock_get_company.assert_called_once()
class TestHandleToolsCallErrors:
"""Test error handling in tools/call."""
def test_handle_tools_call_unknown_tool(self, server):
"""Verify error for unknown tool name."""
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 8,
"params": {
"name": "unknown_tool",
"arguments": {}
}
}
response = server.handle_tools_call(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 8
assert "error" in response
assert response["error"]["code"] == -32601
assert "Bilinmeyen tool: unknown_tool" in response["error"]["message"]
def test_handle_tools_call_exception(self, server, mocker):
"""Verify exception handling during tool execution."""
mock_get_latest = mocker.patch(
'server.spacex_api.get_latest_launch',
side_effect=Exception("Test error")
)
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 9,
"params": {
"name": "get_latest_launch",
"arguments": {}
}
}
response = server.handle_tools_call(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 9
assert "error" in response
assert response["error"]["code"] == -32603
assert "Tool execution error" in response["error"]["message"]
class TestHandleUnknownMethod:
"""Test handling of unknown methods."""
def test_handle_unknown_method(self, server):
"""Verify error for unknown MCP method."""
request = {
"jsonrpc": "2.0",
"method": "unknown_method",
"id": 10
}
response = server.handle_request(request)
assert response["jsonrpc"] == "2.0"
assert response["id"] == 10
assert "error" in response
assert response["error"]["code"] == -32601
assert "Bilinmeyen method" in response["error"]["message"]
class TestSendResponse:
"""Test send_response method."""
def test_send_response(self, server, mocker):
"""Verify send_response writes JSON to stdout."""
mock_stdout = mocker.Mock()
mocker.patch('sys.stdout', mock_stdout)
test_response = {
"jsonrpc": "2.0",
"result": {"test": "data"},
"id": 1
}
server.send_response(test_response)
# Verify json.dump was called (indirectly through stdout.write)
assert mock_stdout.write.called
assert mock_stdout.flush.called