"""Integration tests for HTTP MCP Server.
These tests assume the HTTP server is running at http://localhost:8000.
Start the server with: python http_main.py
"""
import pytest
import requests
import json
from typing import Dict, Any, Optional
class HTTPMCPClient:
"""Simple HTTP client for testing."""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url.rstrip('/')
self.jsonrpc_url = f"{self.base_url}/jsonrpc"
self.request_id = 1
def send_request(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""Send a JSON-RPC request."""
request = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
}
if params:
request["params"] = params
self.request_id += 1
response = requests.post(
self.jsonrpc_url,
json=request,
headers={"Content-Type": "application/json"},
timeout=30
)
response.raise_for_status()
return response.json()
def call_tool(self, name: str, arguments: Dict[str, Any] = None) -> Dict[str, Any]:
"""Call a tool."""
return self.send_request("tools/call", {
"name": name,
"arguments": arguments or {}
})
def list_tools(self) -> Dict[str, Any]:
"""List available tools."""
return self.send_request("tools/list")
def ping(self) -> Dict[str, Any]:
"""Send ping."""
return self.send_request("ping")
def health_check(self) -> Dict[str, Any]:
"""Check server health."""
response = requests.get(f"{self.base_url}/health", timeout=5)
response.raise_for_status()
return response.json()
@pytest.fixture(scope="module")
def http_client():
"""Create HTTP client fixture."""
client = HTTPMCPClient()
# Check if server is running
try:
client.health_check()
except requests.exceptions.ConnectionError:
pytest.skip("HTTP server is not running. Start it with: python http_main.py")
return client
@pytest.fixture(scope="module")
def collection_id(http_client):
"""Get a collection ID for testing from real API."""
try:
response = http_client.call_tool("list_collections", {"per_page": 10})
if "result" in response:
result = response["result"]
if result.get("success") and "data" in result:
data = result["data"]
# Try to extract collection ID (structure may vary)
if "collections" in data and len(data["collections"]) > 0:
collection = data["collections"][0]
return collection.get("id") or collection.get("collection_id")
elif isinstance(data, list) and len(data) > 0:
collection = data[0]
return collection.get("id") or collection.get("collection_id")
elif isinstance(data, dict):
# Check if data itself contains collections
for key in ["items", "results", "data"]:
if key in data and isinstance(data[key], list) and len(data[key]) > 0:
collection = data[key][0]
return collection.get("id") or collection.get("collection_id")
except Exception as e:
pytest.fail(f"Failed to get collection ID from real API: {e}")
pytest.fail("No collection ID found in API response")
@pytest.fixture(scope="module")
def api_id(http_client, collection_id):
"""Get an API ID for testing from real API."""
if not collection_id:
pytest.skip("No collection_id available")
try:
response = http_client.call_tool("get_collection_apis", {
"collection_id": collection_id,
"with_tags": True
})
if "result" in response:
result = response["result"]
if result.get("success") and "data" in result:
data = result["data"]
# Try to extract API ID
if "apis" in data and len(data["apis"]) > 0:
api = data["apis"][0]
return api.get("id") or api.get("api_id")
elif isinstance(data, list) and len(data) > 0:
api = data[0]
return api.get("id") or api.get("api_id")
elif isinstance(data, dict):
for key in ["items", "results", "data"]:
if key in data and isinstance(data[key], list) and len(data[key]) > 0:
api = data[key][0]
return api.get("id") or api.get("api_id")
except Exception as e:
pytest.fail(f"Failed to get API ID from real API: {e}")
pytest.skip("No API ID found in collection")
class TestHTTPHealthCheck:
"""Test health check endpoint."""
def test_health_endpoint(self, http_client):
"""Test health check endpoint."""
health = http_client.health_check()
assert health["status"] == "healthy"
assert "service" in health
def test_root_endpoint(self, http_client):
"""Test root endpoint."""
response = requests.get(f"{http_client.base_url}/", timeout=5)
assert response.status_code == 200
data = response.json()
assert "service" in data
assert data["service"] == "42crunch MCP Server"
def test_tools_endpoint(self, http_client):
"""Test tools listing endpoint."""
response = requests.get(f"{http_client.base_url}/tools", timeout=5)
assert response.status_code == 200
data = response.json()
assert "tools" in data
assert len(data["tools"]) > 0
class TestJSONRPCProtocol:
"""Test JSON-RPC 2.0 protocol compliance."""
def test_initialize(self, http_client):
"""Test initialize method."""
response = http_client.send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
})
assert "jsonrpc" in response
assert response["jsonrpc"] == "2.0"
assert "id" in response
assert "result" in response
result = response["result"]
assert "protocolVersion" in result
assert "serverInfo" in result
def test_ping(self, http_client):
"""Test ping method."""
response = http_client.ping()
assert "jsonrpc" in response
assert response["jsonrpc"] == "2.0"
assert "result" in response
assert response["result"]["status"] == "pong"
def test_invalid_method(self, http_client):
"""Test invalid method handling."""
response = http_client.send_request("invalid_method")
assert "jsonrpc" in response
assert response["jsonrpc"] == "2.0"
assert "error" in response
error = response["error"]
assert error["code"] == -32601 # Method not found
assert "Method not found" in error["message"]
def test_invalid_jsonrpc_version(self, http_client):
"""Test invalid JSON-RPC version handling."""
request = {
"jsonrpc": "1.0", # Invalid version
"id": 1,
"method": "ping"
}
response = requests.post(
http_client.jsonrpc_url,
json=request,
headers={"Content-Type": "application/json"},
timeout=5
)
response.raise_for_status()
result = response.json()
assert "error" in result
assert result["error"]["code"] == -32600 # Invalid Request
class TestToolsList:
"""Test tools/list method."""
def test_list_tools(self, http_client):
"""Test listing available tools."""
response = http_client.list_tools()
assert "jsonrpc" in response
assert response["jsonrpc"] == "2.0"
assert "result" in response
result = response["result"]
assert "tools" in result
tools = result["tools"]
assert len(tools) == 3 # Should have 3 tools
tool_names = [tool["name"] for tool in tools]
assert "list_collections" in tool_names
assert "get_collection_apis" in tool_names
assert "get_api_details" in tool_names
def test_tool_schemas(self, http_client):
"""Test tool input schemas."""
response = http_client.list_tools()
tools = response["result"]["tools"]
for tool in tools:
assert "name" in tool
assert "description" in tool
assert "inputSchema" in tool
assert "type" in tool["inputSchema"]
assert tool["inputSchema"]["type"] == "object"
class TestListCollections:
"""Test list_collections tool."""
def test_list_collections_default(self, http_client):
"""Test list_collections with default parameters."""
response = http_client.call_tool("list_collections")
assert "jsonrpc" in response
assert "result" in response
result = response["result"]
assert result["success"] is True
assert "data" in result
def test_list_collections_with_params(self, http_client):
"""Test list_collections with custom parameters."""
response = http_client.call_tool("list_collections", {
"page": 1,
"per_page": 5,
"order": "default",
"sort": "default"
})
assert "jsonrpc" in response
assert "result" in response
result = response["result"]
assert result["success"] is True
assert "data" in result
def test_list_collections_pagination(self, http_client):
"""Test list_collections pagination."""
# Test first page
response1 = http_client.call_tool("list_collections", {
"page": 1,
"per_page": 2
})
assert response1["result"]["success"] is True
# Test second page
response2 = http_client.call_tool("list_collections", {
"page": 2,
"per_page": 2
})
assert response2["result"]["success"] is True
class TestGetCollectionAPIs:
"""Test get_collection_apis tool."""
def test_get_collection_apis_missing_id(self, http_client):
"""Test get_collection_apis without collection_id."""
response = http_client.call_tool("get_collection_apis", {})
assert "error" in response or "result" in response
if "error" in response:
# Should return JSON-RPC error
assert response["error"]["code"] == -32602 # Invalid params
elif "result" in response:
# Or tool-level error
assert response["result"]["success"] is False
def test_get_collection_apis_with_id(self, http_client, collection_id):
"""Test get_collection_apis with valid collection_id from real API."""
response = http_client.call_tool("get_collection_apis", {
"collection_id": collection_id,
"with_tags": True
})
assert "jsonrpc" in response
assert "result" in response
result = response["result"]
assert result["success"] is True
assert "data" in result
# Validate real API response structure
data = result["data"]
assert isinstance(data, (dict, list))
# If it's a dict, check for common API response keys
if isinstance(data, dict):
# Should have some indication of APIs
assert any(key in data for key in ["apis", "items", "results", "data", "collection"])
def test_get_collection_apis_invalid_id(self, http_client):
"""Test get_collection_apis with invalid collection_id."""
response = http_client.call_tool("get_collection_apis", {
"collection_id": "invalid-uuid-12345",
"with_tags": True
})
# Should either return error or success=False
assert "result" in response or "error" in response
if "result" in response:
# Tool may return success=False for invalid ID
assert response["result"]["success"] is False or response["result"]["success"] is True
class TestGetAPIDetails:
"""Test get_api_details tool."""
def test_get_api_details_missing_id(self, http_client):
"""Test get_api_details without api_id."""
response = http_client.call_tool("get_api_details", {})
assert "error" in response or "result" in response
if "error" in response:
assert response["error"]["code"] == -32602 # Invalid params
elif "result" in response:
assert response["result"]["success"] is False
def test_get_api_details_with_id(self, http_client, api_id):
"""Test get_api_details with valid api_id from real API."""
response = http_client.call_tool("get_api_details", {
"api_id": api_id,
"branch": "main",
"include_definition": True,
"include_assessment": True,
"include_scan": True
})
assert "jsonrpc" in response
assert "result" in response
result = response["result"]
assert result["success"] is True
assert "data" in result
# Validate real API response structure
data = result["data"]
assert isinstance(data, dict)
# Should have API information
assert len(data) > 0
def test_get_api_details_invalid_id(self, http_client):
"""Test get_api_details with invalid api_id."""
response = http_client.call_tool("get_api_details", {
"api_id": "invalid-uuid-12345",
"branch": "main"
})
# Should return error or success=False
assert "result" in response or "error" in response
class TestErrorHandling:
"""Test error handling."""
def test_malformed_json(self, http_client):
"""Test handling of malformed JSON."""
response = requests.post(
http_client.jsonrpc_url,
data="not json",
headers={"Content-Type": "application/json"},
timeout=5
)
# Should return 422 or 400
assert response.status_code in [400, 422]
def test_missing_method(self, http_client):
"""Test request without method."""
request = {
"jsonrpc": "2.0",
"id": 1
}
response = requests.post(
http_client.jsonrpc_url,
json=request,
headers={"Content-Type": "application/json"},
timeout=5
)
# Should handle gracefully
assert response.status_code in [200, 400, 422]
def test_tool_error_propagation(self, http_client):
"""Test that tool errors are properly propagated."""
# Use invalid parameters to trigger an error
response = http_client.call_tool("list_collections", {
"page": -1 # Invalid page number
})
# Should return a response (either error or result with success=False)
assert "result" in response or "error" in response
def test_timeout_handling(self, http_client):
"""Test timeout handling for long-running requests."""
# This test verifies the server handles timeouts properly
# Note: Actual timeout testing would require a slow endpoint
response = http_client.call_tool("list_collections")
assert "result" in response or "error" in response
def test_large_response(self, http_client):
"""Test handling of large responses."""
# Request a large number of items
response = http_client.call_tool("list_collections", {
"per_page": 100
})
assert "result" in response
result = response["result"]
assert result["success"] is True
assert "data" in result
def test_response_structure_validation(self, http_client):
"""Test that responses follow expected structure."""
response = http_client.call_tool("list_collections")
# Validate JSON-RPC structure
assert "jsonrpc" in response
assert response["jsonrpc"] == "2.0"
assert "id" in response
# Validate result structure
if "result" in response:
result = response["result"]
assert "success" in result
assert isinstance(result["success"], bool)
if result["success"]:
assert "data" in result
else:
assert "error" in result
def test_empty_arguments(self, http_client):
"""Test calling tools with empty arguments."""
response = http_client.call_tool("list_collections", {})
assert "result" in response
assert response["result"]["success"] is True
def test_extra_arguments(self, http_client):
"""Test calling tools with extra unexpected arguments."""
response = http_client.call_tool("list_collections", {
"page": 1,
"extra_param": "should_be_ignored"
})
# Should still work, ignoring extra params
assert "result" in response
assert response["result"]["success"] is True
# pytest configuration moved to conftest.py
