test_integration.pyā¢7.31 kB
"""Integration tests for MCP server."""
import pytest
from fastapi.testclient import TestClient
from mcp_server.mcp_http_server import app
@pytest.fixture
def client():
"""Create a test client."""
return TestClient(app)
class TestHTTPTransportIntegration:
"""Integration tests for HTTP transport."""
def test_full_workflow_list_and_call_tool(self, client):
"""Test complete workflow: list tools then call one."""
# Step 1: List available tools
list_payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
list_response = client.post("/mcp/http", json=list_payload)
assert list_response.status_code == 200
list_data = list_response.json()
assert "result" in list_data
assert "tools" in list_data["result"]
# Step 2: Call a specific tool (info)
call_payload = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "info",
"arguments": {}
}
}
call_response = client.post("/mcp/http", json=call_payload)
assert call_response.status_code == 200
call_data = call_response.json()
assert "result" in call_data or "error" in call_data
def test_solve_n_queens_via_http(self, client):
"""Test solving N-Queens problem via HTTP transport."""
payload = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "solve_n_queens",
"arguments": {"n": 4}
}
}
response = client.post("/mcp/http", json=payload)
assert response.status_code == 200
data = response.json()
assert "result" in data or "error" in data
if "result" in data:
result = data["result"]
assert "content" in result
def test_solve_24_point_game_via_http(self, client):
"""Test solving 24-point game via HTTP transport."""
payload = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "solve_24_point_game",
"arguments": {"numbers": [3, 3, 8, 8]}
}
}
response = client.post("/mcp/http", json=payload)
assert response.status_code == 200
data = response.json()
assert "result" in data or "error" in data
def test_streaming_response_format(self, client):
"""Test that streaming responses are properly formatted."""
payload = {
"jsonrpc": "2.0",
"id": 5,
"method": "tools/list",
"params": {}
}
response = client.post(
"/mcp/http",
json=payload,
headers={"Accept": "text/event-stream"}
)
assert response.status_code == 200
# Check SSE format
content = response.text
assert "data:" in content
# Parse SSE data
lines = content.strip().split("\n")
data_lines = [line for line in lines if line.startswith("data:")]
assert len(data_lines) > 0
# Each data line should contain valid JSON
import json
for line in data_lines:
json_str = line[5:].strip() # Remove "data:" prefix
if json_str:
data = json.loads(json_str)
assert "jsonrpc" in data
class TestErrorHandling:
"""Test error handling across transports."""
def test_invalid_method(self, client):
"""Test calling invalid method."""
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "invalid/method",
"params": {}
}
response = client.post("/mcp/http", json=payload)
assert response.status_code == 200
data = response.json()
assert "error" in data
def test_invalid_tool_name(self, client):
"""Test calling non-existent tool."""
payload = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "non_existent_tool",
"arguments": {}
}
}
response = client.post("/mcp/http", json=payload)
assert response.status_code == 200
data = response.json()
# Error might be in result.content or as error field
assert "error" in data or ("result" in data and "error" in str(data["result"]))
def test_missing_required_arguments(self, client):
"""Test calling tool with missing required arguments."""
payload = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "solve_n_queens",
"arguments": {} # Missing 'n' argument
}
}
response = client.post("/mcp/http", json=payload)
assert response.status_code == 200
data = response.json()
# Should either succeed with default or return error
assert "result" in data or "error" in data
def test_invalid_argument_type(self, client):
"""Test calling tool with invalid argument type."""
payload = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "solve_n_queens",
"arguments": {"n": "not_a_number"}
}
}
response = client.post("/mcp/http", json=payload)
assert response.status_code == 200
data = response.json()
assert "error" in data or "result" in data
class TestCORS:
"""Test CORS middleware."""
@pytest.mark.skip(reason="TestClient doesn't include CORS headers in response")
def test_cors_headers_present(self, client):
"""Test that CORS headers are present (skipped - TestClient limitation)."""
# TestClient doesn't include middleware headers in the same way as real requests
pass
@pytest.mark.skip(reason="TestClient doesn't support OPTIONS the same way")
def test_options_request(self, client):
"""Test OPTIONS preflight request (skipped - TestClient limitation)."""
# TestClient handles OPTIONS differently than real HTTP clients
pass
class TestDocumentation:
"""Test API documentation endpoints."""
def test_openapi_docs_available(self, client):
"""Test that OpenAPI docs are available."""
response = client.get("/docs")
assert response.status_code == 200
def test_openapi_json_available(self, client):
"""Test that OpenAPI JSON schema is available."""
response = client.get("/openapi.json")
assert response.status_code == 200
data = response.json()
assert "openapi" in data
assert "info" in data
assert data["info"]["title"] == "Gurddy MCP HTTP Server"
if __name__ == "__main__":
pytest.main([__file__, "-v"])