"""Integration tests for MCP Stdio Server.
These tests assume the MCP server is running via stdio.
The server should be started with: python main.py
"""
import pytest
import subprocess
import json
import time
import sys
from typing import Dict, Any, Optional
class MCPStdioClient:
"""Client for communicating with MCP server via stdio."""
def __init__(self, server_command: list[str] = None):
"""Initialize the stdio client.
Args:
server_command: Command to start the server
"""
self.server_command = server_command or ["python", "main.py"]
self.process: Optional[subprocess.Popen] = None
self.request_id = 1
def start(self) -> None:
"""Start the MCP server process."""
if self.process:
raise RuntimeError("Server is already running")
self.process = subprocess.Popen(
self.server_command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
time.sleep(0.5) # Give server time to initialize
def stop(self) -> None:
"""Stop the MCP server process."""
if self.process:
self.process.terminate()
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait()
self.process = None
def send_request(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""Send a JSON-RPC request."""
if not self.process:
raise RuntimeError("Server is not running")
request = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
}
if params:
request["params"] = params
self.request_id += 1
# Send request
request_json = json.dumps(request) + "\n"
try:
self.process.stdin.write(request_json)
self.process.stdin.flush()
except BrokenPipeError:
raise RuntimeError("Server process terminated unexpectedly")
# Read response
response_line = self.process.stdout.readline()
if not response_line:
raise RuntimeError("No response from server")
try:
return json.loads(response_line.strip())
except json.JSONDecodeError as e:
raise RuntimeError(f"Invalid JSON response: {response_line.strip()}")
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 initialize(self) -> Dict[str, Any]:
"""Initialize connection."""
return self.send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
})
def __enter__(self):
"""Context manager entry."""
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
self.stop()
@pytest.fixture(scope="function")
def mcp_client():
"""Create MCP stdio client fixture."""
client = MCPStdioClient()
try:
client.start()
# Test if server is responsive
try:
client.initialize()
except Exception:
pytest.skip("MCP server is not responding. Start it with: python main.py")
yield client
finally:
client.stop()
@pytest.fixture(scope="function")
def collection_id(mcp_client):
"""Get a collection ID for testing from real API."""
try:
response = mcp_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"]
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):
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="function")
def api_id(mcp_client, collection_id):
"""Get an API ID for testing from real API."""
if not collection_id:
pytest.skip("No collection_id available")
try:
response = mcp_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"]
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 TestMCPProtocol:
"""Test MCP protocol compliance."""
def test_initialize(self, mcp_client):
"""Test initialize method."""
response = mcp_client.initialize()
assert "jsonrpc" in response
assert response["jsonrpc"] == "2.0"
assert "id" in response
assert "result" in response
result = response["result"]
assert "protocolVersion" in result or "serverInfo" in result
def test_invalid_method(self, mcp_client):
"""Test invalid method handling."""
response = mcp_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
class TestMCPToolsList:
"""Test tools/list method."""
def test_list_tools(self, mcp_client):
"""Test listing available tools."""
response = mcp_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 at least 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_descriptions(self, mcp_client):
"""Test tool descriptions."""
response = mcp_client.list_tools()
tools = response["result"]["tools"]
for tool in tools:
assert "name" in tool
assert "description" in tool
assert len(tool["description"]) > 0
class TestMCPListCollections:
"""Test list_collections tool via MCP."""
def test_list_collections_default(self, mcp_client):
"""Test list_collections with default parameters."""
response = mcp_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, mcp_client):
"""Test list_collections with custom parameters."""
response = mcp_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_multiple_calls(self, mcp_client):
"""Test multiple consecutive calls."""
# First call
response1 = mcp_client.call_tool("list_collections", {"per_page": 2})
assert response1["result"]["success"] is True
# Second call
response2 = mcp_client.call_tool("list_collections", {"per_page": 2})
assert response2["result"]["success"] is True
# Should get different request IDs
assert response1["id"] != response2["id"]
class TestMCPGetCollectionAPIs:
"""Test get_collection_apis tool via MCP."""
def test_get_collection_apis_missing_id(self, mcp_client):
"""Test get_collection_apis without collection_id."""
response = mcp_client.call_tool("get_collection_apis", {})
assert "result" in response
result = response["result"]
assert result["success"] is False
assert "error" in result
def test_get_collection_apis_with_id(self, mcp_client, collection_id):
"""Test get_collection_apis with valid collection_id from real API."""
response = mcp_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):
assert any(key in data for key in ["apis", "items", "results", "data", "collection"])
def test_get_collection_apis_invalid_id(self, mcp_client):
"""Test get_collection_apis with invalid collection_id."""
response = mcp_client.call_tool("get_collection_apis", {
"collection_id": "invalid-uuid-12345",
"with_tags": True
})
assert "result" in response
# May return success=False or success=True depending on API behavior
assert isinstance(response["result"]["success"], bool)
class TestMCPGetAPIDetails:
"""Test get_api_details tool via MCP."""
def test_get_api_details_missing_id(self, mcp_client):
"""Test get_api_details without api_id."""
response = mcp_client.call_tool("get_api_details", {})
assert "result" in response
result = response["result"]
assert result["success"] is False
assert "error" in result
def test_get_api_details_with_id(self, mcp_client, api_id):
"""Test get_api_details with valid api_id from real API."""
response = mcp_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, mcp_client):
"""Test get_api_details with invalid api_id."""
response = mcp_client.call_tool("get_api_details", {
"api_id": "invalid-uuid-12345",
"branch": "main"
})
assert "result" in response
assert isinstance(response["result"]["success"], bool)
class TestMCPErrorHandling:
"""Test error handling in MCP server."""
def test_malformed_request(self, mcp_client):
"""Test handling of malformed request."""
# Send invalid JSON
try:
mcp_client.process.stdin.write("invalid json\n")
mcp_client.process.stdin.flush()
# Should handle gracefully or close connection
except Exception:
# Expected if server closes connection
pass
def test_missing_required_params(self, mcp_client):
"""Test missing required parameters."""
response = mcp_client.call_tool("get_collection_apis", {})
assert "result" in response
result = response["result"]
assert result["success"] is False
assert "error" in result
def test_concurrent_requests(self, mcp_client):
"""Test handling of concurrent requests."""
# Send multiple requests
responses = []
for i in range(3):
response = mcp_client.call_tool("list_collections", {"per_page": 1})
responses.append(response)
# All should succeed
assert len(responses) == 3
for response in responses:
assert response["result"]["success"] is True
class TestMCPConnection:
"""Test MCP connection handling."""
def test_server_restart(self, mcp_client):
"""Test that client can handle server restart."""
# Make a request
response1 = mcp_client.call_tool("list_collections")
assert response1["result"]["success"] is True
# Restart server
mcp_client.stop()
time.sleep(0.5)
mcp_client.start()
time.sleep(0.5)
# Make another request
response2 = mcp_client.call_tool("list_collections")
assert response2["result"]["success"] is True
def test_response_structure_validation(self, mcp_client):
"""Test that responses follow expected structure."""
response = mcp_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, mcp_client):
"""Test calling tools with empty arguments."""
response = mcp_client.call_tool("list_collections", {})
assert "result" in response
assert response["result"]["success"] is True
def test_extra_arguments(self, mcp_client):
"""Test calling tools with extra unexpected arguments."""
response = mcp_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
def test_request_id_uniqueness(self, mcp_client):
"""Test that request IDs are unique."""
responses = []
for _ in range(5):
response = mcp_client.call_tool("list_collections")
responses.append(response)
# Extract IDs
ids = [r["id"] for r in responses]
# All IDs should be unique
assert len(ids) == len(set(ids))
def test_large_response(self, mcp_client):
"""Test handling of large responses."""
response = mcp_client.call_tool("list_collections", {
"per_page": 100
})
assert "result" in response
result = response["result"]
assert result["success"] is True
assert "data" in result
# pytest configuration moved to conftest.py
