"""Comprehensive tests for OpenAPI new implementation.""" import json from unittest.mock import AsyncMock, Mock import httpx import pytest from httpx import Response from fastmcp.client import Client from fastmcp.experimental.server.openapi import FastMCPOpenAPI class TestOpenAPIComprehensive: """Comprehensive tests ensuring no functionality is lost.""" @pytest.fixture def comprehensive_openapi_spec(self): """Comprehensive OpenAPI spec covering all major features.""" return { "openapi": "3.0.0", "info": {"title": "Comprehensive API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "components": { "schemas": { "User": { "type": "object", "properties": { "id": {"type": "integer"}, "name": {"type": "string"}, "email": {"type": "string", "format": "email"}, "age": {"type": "integer", "minimum": 0}, }, "required": ["name", "email"], }, "Error": { "type": "object", "properties": { "code": {"type": "integer"}, "message": {"type": "string"}, }, }, }, "parameters": { "UserId": { "name": "id", "in": "path", "required": True, "schema": {"type": "integer"}, "description": "User identifier", } }, }, "paths": { # Basic CRUD operations "/users": { "get": { "operationId": "list_users", "summary": "List all users", "parameters": [ { "name": "limit", "in": "query", "schema": { "type": "integer", "default": 10, "minimum": 1, "maximum": 100, }, "description": "Number of users to return", }, { "name": "offset", "in": "query", "schema": { "type": "integer", "default": 0, "minimum": 0, }, "description": "Number of users to skip", }, { "name": "sort", "in": "query", "schema": { "type": "string", "enum": ["name", "email", "age"], }, "description": "Sort field", }, ], "responses": { "200": { "description": "List of users", "content": { "application/json": { "schema": { "type": "array", "items": { "$ref": "#/components/schemas/User" }, } } }, } }, }, "post": { "operationId": "create_user", "summary": "Create a new user", "requestBody": { "required": True, "content": { "application/json": { "schema": {"$ref": "#/components/schemas/User"} } }, }, "responses": { "201": { "description": "User created", "content": { "application/json": { "schema": {"$ref": "#/components/schemas/User"} } }, }, "400": { "description": "Invalid input", "content": { "application/json": { "schema": {"$ref": "#/components/schemas/Error"} } }, }, }, }, }, "/users/{id}": { "parameters": [{"$ref": "#/components/parameters/UserId"}], "get": { "operationId": "get_user", "summary": "Get user by ID", "responses": { "200": { "description": "User details", "content": { "application/json": { "schema": {"$ref": "#/components/schemas/User"} } }, }, "404": { "description": "User not found", "content": { "application/json": { "schema": {"$ref": "#/components/schemas/Error"} } }, }, }, }, "put": { "operationId": "update_user", "summary": "Update user", "requestBody": { "required": True, "content": { "application/json": { "schema": {"$ref": "#/components/schemas/User"} } }, }, "responses": { "200": { "description": "User updated", "content": { "application/json": { "schema": {"$ref": "#/components/schemas/User"} } }, }, }, }, "delete": { "operationId": "delete_user", "summary": "Delete user", "responses": { "204": {"description": "User deleted"}, "404": { "description": "User not found", "content": { "application/json": { "schema": {"$ref": "#/components/schemas/Error"} } }, }, }, }, }, # Complex parameter scenarios "/search": { "get": { "operationId": "search_users", "summary": "Search users with complex filters", "parameters": [ { "name": "q", "in": "query", "required": True, "schema": {"type": "string"}, "description": "Search query", }, { "name": "filter", "in": "query", "style": "deepObject", "explode": True, "schema": { "type": "object", "properties": { "age": { "type": "object", "properties": { "min": {"type": "integer"}, "max": {"type": "integer"}, }, }, "name": {"type": "string"}, "active": {"type": "boolean"}, }, }, }, { "name": "X-Request-ID", "in": "header", "schema": {"type": "string"}, "description": "Request identifier for tracing", }, ], "responses": { "200": { "description": "Search results", "content": { "application/json": { "schema": { "type": "object", "properties": { "results": { "type": "array", "items": { "$ref": "#/components/schemas/User" }, }, "total": {"type": "integer"}, "page": {"type": "integer"}, }, } } }, } }, } }, # Parameter collision scenario "/collision/{id}": { "patch": { "operationId": "collision_test", "summary": "Test parameter collision handling", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "string"}, "description": "Resource ID", }, { "name": "version", "in": "query", "schema": {"type": "integer", "default": 1}, }, { "name": "version", "in": "header", "schema": {"type": "string"}, }, ], "requestBody": { "required": True, "content": { "application/json": { "schema": { "type": "object", "properties": { "id": { "type": "integer", "description": "Internal ID", }, "version": { "type": "string", "description": "Data version", }, "data": {"type": "object"}, }, } } }, }, "responses": {"200": {"description": "Updated"}}, } }, }, } @pytest.fixture def openapi_31_spec(self): """OpenAPI 3.1 spec to test compatibility.""" return { "openapi": "3.1.0", "info": {"title": "OpenAPI 3.1 Test", "version": "1.0.0"}, "paths": { "/items/{id}": { "get": { "operationId": "get_item_31", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "string"}, } ], "responses": { "200": { "description": "Item details", "content": { "application/json": { "schema": { "type": "object", "properties": { "id": {"type": "string"}, "name": {"type": "string"}, }, } } }, } }, } } }, } async def test_comprehensive_server_initialization( self, comprehensive_openapi_spec ): """Test server initialization with comprehensive spec.""" async with httpx.AsyncClient(base_url="https://api.example.com") as client: server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=client, name="Comprehensive Test Server", ) # Should initialize successfully assert server.name == "Comprehensive Test Server" assert hasattr(server, "_director") assert hasattr(server, "_spec") # Test with in-memory client async with Client(server) as mcp_client: tools = await mcp_client.list_tools() # Should have created tools for all operations tool_names = {tool.name for tool in tools} expected_operations = { "list_users", "create_user", "get_user", "update_user", "delete_user", "search_users", "collision_test", } assert tool_names == expected_operations async def test_openapi_31_compatibility(self, openapi_31_spec): """Test that OpenAPI 3.1 specs work correctly.""" async with httpx.AsyncClient(base_url="https://api.example.com") as client: server = FastMCPOpenAPI( openapi_spec=openapi_31_spec, client=client, name="OpenAPI 3.1 Test", ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() assert len(tools) == 1 tool = tools[0] assert tool.name == "get_item_31" async def test_parameter_collision_handling(self, comprehensive_openapi_spec): """Test that parameter collisions are handled correctly.""" async with httpx.AsyncClient(base_url="https://api.example.com") as client: server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=client, ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() collision_tool = next( tool for tool in tools if tool.name == "collision_test" ) schema = collision_tool.inputSchema properties = schema["properties"] # Should have unique parameter names for colliding parameters param_names = list(properties.keys()) # Should have some form of id parameters (path and body) id_params = [name for name in param_names if "id" in name] assert len(id_params) >= 2 # Should have some form of version parameters (query, header, body) version_params = [name for name in param_names if "version" in name] assert len(version_params) >= 3 # Should have other parameters assert "data" in param_names async def test_deep_object_parameters(self, comprehensive_openapi_spec): """Test deepObject parameter handling.""" async with httpx.AsyncClient(base_url="https://api.example.com") as client: server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=client, ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() search_tool = next( tool for tool in tools if tool.name == "search_users" ) schema = search_tool.inputSchema properties = schema["properties"] # Should have flattened deepObject parameters # The exact flattening depends on implementation assert "q" in properties # Regular query parameter # Should have some form of filter parameters filter_params = [name for name in properties.keys() if "filter" in name] assert len(filter_params) > 0 async def test_request_building_and_execution(self, comprehensive_openapi_spec): """Test that requests are built and executed correctly.""" # Create a mock client that tracks requests mock_client = Mock(spec=httpx.AsyncClient) mock_client.base_url = "https://api.example.com" mock_client.headers = None # Mock successful response mock_response = Mock(spec=Response) mock_response.status_code = 200 mock_response.json.return_value = { "id": 123, "name": "Test User", "email": "test@example.com", } mock_response.text = json.dumps( {"id": 123, "name": "Test User", "email": "test@example.com"} ) mock_response.raise_for_status = Mock() mock_client.send = AsyncMock(return_value=mock_response) server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=mock_client, ) async with Client(server) as mcp_client: # Test GET request with path parameter await mcp_client.call_tool("get_user", {"id": 123}) # Should have made a request mock_client.send.assert_called_once() request = mock_client.send.call_args[0][0] # Verify request details assert request.method == "GET" assert "123" in str(request.url) assert "users/123" in str(request.url) async def test_complex_request_with_body_and_parameters( self, comprehensive_openapi_spec ): """Test complex request with both parameters and body.""" mock_client = Mock(spec=httpx.AsyncClient) mock_client.base_url = "https://api.example.com" mock_client.headers = None mock_response = Mock(spec=Response) mock_response.status_code = 201 mock_response.json.return_value = { "id": 456, "name": "New User", "email": "new@example.com", } mock_response.raise_for_status = Mock() mock_client.send = AsyncMock(return_value=mock_response) server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=mock_client, ) async with Client(server) as mcp_client: # Test POST request with body await mcp_client.call_tool( "create_user", { "name": "New User", "email": "new@example.com", "age": 25, }, ) # Should have made a request mock_client.send.assert_called_once() request = mock_client.send.call_args[0][0] # Verify request details assert request.method == "POST" assert "users" in str(request.url) # Should have JSON body assert request.content is not None body_data = json.loads(request.content) assert body_data["name"] == "New User" assert body_data["email"] == "new@example.com" assert body_data["age"] == 25 async def test_query_parameters(self, comprehensive_openapi_spec): """Test query parameter handling.""" mock_client = Mock(spec=httpx.AsyncClient) mock_client.base_url = "https://api.example.com" mock_client.headers = None mock_response = Mock(spec=Response) mock_response.status_code = 200 mock_response.json.return_value = [] mock_response.raise_for_status = Mock() mock_client.send = AsyncMock(return_value=mock_response) server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=mock_client, ) async with Client(server) as mcp_client: # Test GET request with query parameters await mcp_client.call_tool( "list_users", { "limit": 20, "offset": 10, "sort": "name", }, ) mock_client.send.assert_called_once() request = mock_client.send.call_args[0][0] # Verify query parameters in URL url_str = str(request.url) assert "limit=20" in url_str assert "offset=10" in url_str assert "sort=name" in url_str async def test_error_handling(self, comprehensive_openapi_spec): """Test error handling for HTTP errors.""" mock_client = Mock(spec=httpx.AsyncClient) mock_client.base_url = "https://api.example.com" mock_client.headers = None # Mock HTTP error response mock_response = Mock(spec=Response) mock_response.status_code = 404 mock_response.reason_phrase = "Not Found" mock_response.json.return_value = {"code": 404, "message": "User not found"} mock_response.text = json.dumps({"code": 404, "message": "User not found"}) # Configure raise_for_status to raise HTTPStatusError def raise_for_status(): raise httpx.HTTPStatusError( "404 Not Found", request=Mock(), response=mock_response ) mock_response.raise_for_status = raise_for_status mock_client.send = AsyncMock(return_value=mock_response) server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=mock_client, ) async with Client(server) as mcp_client: # Should handle HTTP errors gracefully with pytest.raises(Exception) as exc_info: await mcp_client.call_tool("get_user", {"id": 999}) # Error should be wrapped appropriately error_message = str(exc_info.value) assert "404" in error_message async def test_schema_refs_resolution(self, comprehensive_openapi_spec): """Test that schema references are resolved correctly.""" async with httpx.AsyncClient(base_url="https://api.example.com") as client: server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=client, ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() # Find create_user tool which uses schema refs create_tool = next(tool for tool in tools if tool.name == "create_user") schema = create_tool.inputSchema properties = schema["properties"] # Should have resolved User schema properties assert "name" in properties assert "email" in properties # May also have id and age depending on implementation async def test_optional_vs_required_parameters(self, comprehensive_openapi_spec): """Test handling of optional vs required parameters.""" async with httpx.AsyncClient(base_url="https://api.example.com") as client: server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=client, ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() # Check list_users tool - has optional query parameters list_tool = next(tool for tool in tools if tool.name == "list_users") schema = list_tool.inputSchema # Query parameters should be optional # (may not appear in required list) # This test just ensures the schema is well-formed assert "properties" in schema # Check search_users tool - has required query parameter search_tool = next( tool for tool in tools if tool.name == "search_users" ) search_schema = search_tool.inputSchema # Should have some required parameters assert len(search_schema["properties"]) > 0 async def test_server_performance_no_latency(self, comprehensive_openapi_spec): """Test that server initialization is fast (no code generation latency).""" import time # Time the server creation start_time = time.time() async with httpx.AsyncClient(base_url="https://api.example.com") as client: server = FastMCPOpenAPI( openapi_spec=comprehensive_openapi_spec, client=client, ) end_time = time.time() # Should be very fast (no code generation) initialization_time = end_time - start_time assert initialization_time < 0.1 # Should be under 100ms # Verify server was created correctly assert server is not None assert hasattr(server, "_director") assert hasattr(server, "_spec")