test_mcp_server.py•5.31 kB
"""Test MCP OpenAPI server functionality."""
import pytest
from tests.tool_operations import (
MINIMUM_EXPECTED_TOOLS,
build_tool_arguments,
extract_tool_names,
find_read_operation_tool,
)
JSONRPC_VERSION = "2.0"
EXPECTED_SCHEMA_TYPE = "object"
SAMPLE_TOOLS_COUNT = 5
class TestMCPProtocol:
"""Test MCP protocol compliance."""
def test_tools_list(self, mcp_client, mcp_url):
"""Test listing available MCP tools."""
response = mcp_client.request("tools/list")
self._validate_jsonrpc_response(response)
tools = self._extract_tools(response)
assert len(tools) > 0, "No tools returned"
print(f"\n✓ Found {len(tools)} tools")
print(f"✓ Endpoint: {mcp_url}")
def test_tool_structure(self, mcp_client):
"""Test that tools have the correct structure."""
response = mcp_client.request("tools/list")
tools = response["result"]["tools"]
for tool in tools[:3]:
self._validate_tool_structure(tool)
self._validate_input_schema(tool["inputSchema"])
def _validate_jsonrpc_response(self, response: dict) -> None:
"""Validate basic JSON-RPC response structure."""
assert "jsonrpc" in response, "Missing 'jsonrpc' field"
assert response["jsonrpc"] == JSONRPC_VERSION, "Invalid JSON-RPC version"
assert "id" in response, "Missing 'id' field"
assert "error" not in response, f"Response contains error: {response.get('error')}"
assert "result" in response, "Missing 'result' field"
def _extract_tools(self, response: dict) -> list:
"""Extract and validate tools list from response."""
result = response["result"]
assert "tools" in result, f"Missing 'tools' in result: {result}"
tools = result["tools"]
assert isinstance(tools, list), f"'tools' is not a list: {type(tools)}"
return tools
def _validate_tool_structure(self, tool: dict) -> None:
"""Validate tool has required fields."""
assert "name" in tool, f"Tool missing 'name': {tool}"
assert "description" in tool, f"Tool missing 'description': {tool}"
assert "inputSchema" in tool, f"Tool missing 'inputSchema': {tool}"
def _validate_input_schema(self, schema: dict) -> None:
"""Validate input schema structure."""
assert "type" in schema, f"Input schema missing 'type': {schema}"
assert schema["type"] == EXPECTED_SCHEMA_TYPE, \
f"Input schema type should be '{EXPECTED_SCHEMA_TYPE}': {schema}"
class TestAPIIntegration:
"""Test API integration through MCP tools."""
@pytest.fixture
def available_tools(self, mcp_client):
"""Get list of available tools."""
response = mcp_client.request("tools/list")
return response["result"]["tools"]
def test_tool_call_with_get_operation(self, mcp_client, available_tools):
"""Test calling a GET operation through MCP."""
tool = find_read_operation_tool(available_tools)
assert tool is not None, "No GET operation found"
tool_name = tool["name"]
print(f"\n✓ Testing tool: {tool_name}")
arguments = build_tool_arguments(tool)
response = self._call_tool(mcp_client, tool_name, arguments)
self._validate_successful_call(response)
print("✓ Tool call completed successfully")
def test_multiple_tools_available(self, available_tools):
"""Test that multiple API operations are exposed as tools."""
assert len(available_tools) > MINIMUM_EXPECTED_TOOLS, \
f"Expected more than {MINIMUM_EXPECTED_TOOLS} tools, got {len(available_tools)}"
tool_names = extract_tool_names(available_tools)
print(f"\n✓ Total tools: {len(tool_names)}")
print(f"✓ Sample tools: {', '.join(tool_names[:SAMPLE_TOOLS_COUNT])}")
def _call_tool(self, mcp_client, tool_name: str, arguments: dict) -> dict:
"""Execute tool call via MCP."""
return mcp_client.request(
"tools/call",
params={"name": tool_name, "arguments": arguments},
request_id=2
)
def _validate_successful_call(self, response: dict) -> None:
"""Validate tool call was successful."""
assert "error" not in response, \
f"Tool call returned error: {response.get('error')}"
assert "result" in response, "Missing result in tool call response"
class TestErrorHandling:
"""Test error handling."""
def test_invalid_method(self, mcp_client):
"""Test calling an invalid MCP method."""
response = mcp_client.request("invalid/method")
assert "error" in response, "Expected error for invalid method"
assert "result" not in response, "Should not have result for invalid method"
def test_invalid_tool_name(self, mcp_client):
"""Test calling a non-existent tool."""
response = mcp_client.request(
"tools/call",
params={"name": "nonExistentTool", "arguments": {}},
request_id=3
)
assert "result" in response, "Expected result field"
result = response["result"]
assert result.get("isError") is True, \
"Expected isError flag for non-existent tool"