diff --git a/src/fastmcp/utilities/openapi.py b/src/fastmcp/utilities/openapi.py new file mode 100644 index 000000000..9b2fe599a --- /dev/null +++ b/src/fastmcp/utilities/openapi.py @@ -0,0 +1,468 @@ +import logging +from typing import Any, Dict, List, Literal, Optional, Tuple, Union, cast + +# Using the recommended library: openapi-pydantic +from openapi_pydantic import ( + MediaType, + OpenAPI, + Operation, + Parameter, + PathItem, + Reference, + RequestBody, + Schema, +) +from pydantic import BaseModel, Field, ValidationError + +logger = logging.getLogger(__name__) + +# --- Intermediate Representation (IR) Definition --- +# (IR models remain the same) + +HttpMethod = Literal[ + "GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD", "TRACE" +] +ParameterLocation = Literal["path", "query", "header", "cookie"] +JsonSchema = Dict[str, Any] + + +class ParameterInfo(BaseModel): + """Represents a single parameter for an HTTP operation in our IR.""" + + name: str + location: ParameterLocation # Mapped from 'in' field of openapi-pydantic Parameter + required: bool = False + schema_: JsonSchema = Field(..., alias="schema") # Target name in IR + description: Optional[str] = None + + # No model_config needed here if we populate manually after accessing 'in' + + +class RequestBodyInfo(BaseModel): + """Represents the request body for an HTTP operation in our IR.""" + + required: bool = False + content_schema: Dict[str, JsonSchema] = Field( + default_factory=dict + ) # Key: media type + description: Optional[str] = None + + +class HTTPRoute(BaseModel): + """Intermediate Representation for a single OpenAPI operation.""" + + path: str + method: HttpMethod + operation_id: Optional[str] = None + summary: Optional[str] = None + description: Optional[str] = None + tags: List[str] = Field(default_factory=list) + parameters: List[ParameterInfo] = Field(default_factory=list) + request_body: Optional[RequestBodyInfo] = None + + +# --- Helper Functions --- + + +def _resolve_ref( + item: Union[Reference, Schema, Parameter, RequestBody, Any], openapi: OpenAPI +) -> Any: + """Resolves a potential Reference object to its target definition (no changes needed here).""" + if isinstance(item, Reference): + ref_str = item.ref + try: + if not ref_str.startswith("#/"): + raise ValueError( + f"External or non-local reference not supported: {ref_str}" + ) + parts = ref_str.strip("#/").split("/") + target = openapi + for part in parts: + if part.isdigit() and isinstance(target, list): + target = target[int(part)] + elif isinstance(target, BaseModel): + # Use model_extra for fields not explicitly defined (like components types) + # Check class fields first, then model_extra + if part in target.model_fields: # Access class attribute here + target = getattr(target, part, None) + elif target.model_extra and part in target.model_extra: + target = target.model_extra[part] + else: + # Special handling for components sub-types common structure + if part == "components" and hasattr(target, "components"): + target = getattr(target, "components") + elif hasattr(target, part): # Fallback check + target = getattr(target, part, None) + else: + target = None # Part not found + elif isinstance(target, dict): + target = target.get(part) + else: + raise ValueError( + f"Cannot traverse part '{part}' in reference '{ref_str}' from type {type(target)}" + ) + if target is None: + raise ValueError( + f"Reference part '{part}' not found in path '{ref_str}'" + ) + if isinstance(target, Reference): + return _resolve_ref(target, openapi) + return target + except (AttributeError, KeyError, IndexError, TypeError, ValueError) as e: + raise ValueError(f"Failed to resolve reference '{ref_str}': {e}") from e + return item + + +def _extract_schema_as_dict( + schema_obj: Union[Schema, Reference], openapi: OpenAPI +) -> JsonSchema: + """Resolves a schema/reference and returns it as a dictionary.""" + resolved_schema = _resolve_ref(schema_obj, openapi) + if isinstance(resolved_schema, Schema): + # Using exclude_none=True might be better than exclude_unset sometimes + return resolved_schema.model_dump(mode="json", by_alias=True, exclude_none=True) + elif isinstance(resolved_schema, dict): + logger.warning( + "Resolved schema reference resulted in a dict, not a Schema model." + ) + return resolved_schema + else: + ref_str = getattr(schema_obj, "ref", "unknown") + logger.warning( + f"Expected Schema after resolving ref '{ref_str}', got {type(resolved_schema)}. Returning empty dict." + ) + return {} + + +def _convert_to_parameter_location(param_in: str) -> ParameterLocation: + """Convert string parameter location to our ParameterLocation type.""" + if param_in == "path": + return "path" + elif param_in == "query": + return "query" + elif param_in == "header": + return "header" + elif param_in == "cookie": + return "cookie" + else: + logger.warning(f"Unknown parameter location: {param_in}, defaulting to 'query'") + return "query" + + +def _extract_parameters( + operation_params: Optional[List[Union[Parameter, Reference]]], + path_item_params: Optional[List[Union[Parameter, Reference]]], + openapi: OpenAPI, +) -> List[ParameterInfo]: + """Extracts and resolves parameters using corrected attribute names.""" + extracted_params: List[ParameterInfo] = [] + seen_params: Dict[ + Tuple[str, str], bool + ] = {} # Use string keys to avoid type issues + all_params_refs = (operation_params or []) + (path_item_params or []) + + for param_or_ref in all_params_refs: + try: + parameter = cast(Parameter, _resolve_ref(param_or_ref, openapi)) + if not isinstance(parameter, Parameter): + # ... (error logging remains the same) + continue + + # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** --- + param_in = parameter.param_in # CORRECTED: Use 'param_in' + param_location = _convert_to_parameter_location(param_in) + param_schema_obj = parameter.param_schema # CORRECTED: Use 'param_schema' + # --- *** --- + + param_key = (parameter.name, param_in) + if param_key in seen_params: + continue + seen_params[param_key] = True + + param_schema_dict = {} + if param_schema_obj: # Check if schema exists + param_schema_dict = _extract_schema_as_dict(param_schema_obj, openapi) + elif parameter.content: + # Handle complex parameters with 'content' + first_media_type = next(iter(parameter.content.values()), None) + if ( + first_media_type and first_media_type.media_type_schema + ): # CORRECTED: Use 'media_type_schema' + param_schema_dict = _extract_schema_as_dict( + first_media_type.media_type_schema, openapi + ) + logger.debug( + f"Parameter '{parameter.name}' using schema from 'content' field." + ) + + # Manually create ParameterInfo instance using correct field names + param_info = ParameterInfo( + name=parameter.name, + location=param_location, # Use converted parameter location + required=parameter.required, + schema=param_schema_dict, # Populate 'schema' field in IR + description=parameter.description, + ) + extracted_params.append(param_info) + + except ( + ValidationError, + ValueError, + AttributeError, + TypeError, + ) as e: # Added TypeError + param_name = getattr( + param_or_ref, "name", getattr(param_or_ref, "ref", "unknown") + ) + logger.error( + f"Failed to extract parameter '{param_name}': {e}", exc_info=False + ) + + return extracted_params + + +def _extract_request_body( + request_body_or_ref: Optional[Union[RequestBody, Reference]], openapi: OpenAPI +) -> Optional[RequestBodyInfo]: + """Extracts and resolves the request body using corrected attribute names.""" + if not request_body_or_ref: + return None + try: + request_body = cast(RequestBody, _resolve_ref(request_body_or_ref, openapi)) + if not isinstance(request_body, RequestBody): + # ... (error logging remains the same) + return None + + content_schemas: Dict[str, JsonSchema] = {} + if request_body.content: + for media_type_str, media_type_obj in request_body.content.items(): + # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** --- + if ( + isinstance(media_type_obj, MediaType) + and media_type_obj.media_type_schema + ): # CORRECTED: Use 'media_type_schema' + # --- *** --- + try: + # Use the corrected attribute here as well + schema_dict = _extract_schema_as_dict( + media_type_obj.media_type_schema, openapi + ) + content_schemas[media_type_str] = schema_dict + except ValueError as schema_err: + logger.error( + f"Failed to extract schema for media type '{media_type_str}' in request body: {schema_err}" + ) + elif not isinstance(media_type_obj, MediaType): + logger.warning( + f"Skipping invalid media type object for '{media_type_str}' (type: {type(media_type_obj)}) in request body." + ) + elif not media_type_obj.media_type_schema: # Corrected check + logger.warning( + f"Skipping media type '{media_type_str}' in request body because it lacks a schema." + ) + + return RequestBodyInfo( + required=request_body.required, + content_schema=content_schemas, + description=request_body.description, + ) + except (ValidationError, ValueError, AttributeError) as e: + ref_name = getattr(request_body_or_ref, "ref", "unknown") + logger.error( + f"Failed to extract request body '{ref_name}': {e}", exc_info=False + ) + return None + + +# --- Main Parsing Function --- +# (No changes needed in the main loop logic, only in the helpers it calls) +def parse_openapi_to_http_routes(openapi_dict: Dict[str, Any]) -> List[HTTPRoute]: + """ + Parses an OpenAPI schema dictionary into a list of HTTPRoute objects + using the openapi-pydantic library. + """ + routes: List[HTTPRoute] = [] + try: + openapi: OpenAPI = OpenAPI.model_validate(openapi_dict) + logger.info(f"Successfully parsed OpenAPI schema version: {openapi.openapi}") + except ValidationError as e: + logger.error(f"OpenAPI schema validation failed: {e}") + error_details = e.errors() + logger.error(f"Validation errors: {error_details}") + raise ValueError(f"Invalid OpenAPI schema: {error_details}") from e + + if not openapi.paths: + logger.warning("OpenAPI schema has no paths defined.") + return [] + + for path_str, path_item_obj in openapi.paths.items(): + if not isinstance(path_item_obj, PathItem): + logger.warning( + f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})" + ) + continue + + path_level_params = path_item_obj.parameters + + # Iterate through possible HTTP methods defined in the PathItem model fields + # Use model_fields from the class, not the instance + for method_lower in PathItem.model_fields.keys(): + if method_lower not in [ + "get", + "put", + "post", + "delete", + "options", + "head", + "patch", + "trace", + ]: + continue + + operation: Optional[Operation] = getattr(path_item_obj, method_lower, None) + + if operation and isinstance(operation, Operation): + method_upper = cast(HttpMethod, method_lower.upper()) + logger.debug(f"Processing operation: {method_upper} {path_str}") + try: + parameters = _extract_parameters( + operation.parameters, path_level_params, openapi + ) + request_body_info = _extract_request_body( + operation.requestBody, openapi + ) + + route = HTTPRoute( + path=path_str, + method=method_upper, + operation_id=operation.operationId, + summary=operation.summary, + description=operation.description, + tags=operation.tags or [], + parameters=parameters, + request_body=request_body_info, + ) + routes.append(route) + logger.info( + f"Successfully extracted route: {method_upper} {path_str}" + ) + except Exception as op_error: + op_id = operation.operationId or "unknown" + logger.error( + f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}", + exc_info=True, + ) + + logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.") + return routes + + +# --- Example Usage (Optional) --- +if __name__ == "__main__": + import json + + logging.basicConfig( + level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s" + ) # Set to INFO + + petstore_schema = { + "openapi": "3.1.0", # Keep corrected version + "info": {"title": "Simple Pet Store API", "version": "1.0.0"}, + "paths": { + "/pets": { + "get": { + "summary": "List all pets", + "operationId": "listPets", + "tags": ["pets"], + "parameters": [ + { + "name": "limit", + "in": "query", + "description": "How many items to return", + "required": False, + "schema": {"type": "integer", "format": "int32"}, + } + ], + "responses": {"200": {"description": "A paged array of pets"}}, + }, + "post": { + "summary": "Create a pet", + "operationId": "createPet", + "tags": ["pets"], + "requestBody": {"$ref": "#/components/requestBodies/PetBody"}, + "responses": {"201": {"description": "Null response"}}, + }, + }, + "/pets/{petId}": { + "get": { + "summary": "Info for a specific pet", + "operationId": "showPetById", + "tags": ["pets"], + "parameters": [ + { + "name": "petId", + "in": "path", + "required": True, + "description": "The id of the pet", + "schema": {"type": "string"}, + }, + { + "name": "X-Request-ID", + "in": "header", + "required": False, + "schema": {"type": "string", "format": "uuid"}, + }, + ], + "responses": {"200": {"description": "Information about the pet"}}, + }, + "parameters": [ # Path level parameter example + { + "name": "traceId", + "in": "header", + "description": "Common trace ID", + "required": False, + "schema": {"type": "string"}, + } + ], + }, + }, + "components": { + "schemas": { + "Pet": { + "type": "object", + "required": ["id", "name"], + "properties": { + "id": {"type": "integer", "format": "int64"}, + "name": {"type": "string"}, + "tag": {"type": "string"}, + }, + } + }, + "requestBodies": { + "PetBody": { + "description": "Pet object", + "required": True, + "content": { + "application/json": { + "schema": {"$ref": "#/components/schemas/Pet"} + } + }, + } + }, + }, + } + + print("--- Parsing Pet Store Schema using openapi-pydantic (Corrected) ---") + try: + http_routes = parse_openapi_to_http_routes(petstore_schema) + print(f"\n--- Extracted {len(http_routes)} Routes ---") + for i, route in enumerate(http_routes): + print(f"\nRoute {i + 1}:") + # Use model_dump for clean JSON-like output, show aliases from IR model + print( + json.dumps(route.model_dump(by_alias=True, exclude_none=True), indent=2) + ) # exclude_none is often cleaner + except ValueError as e: + print(f"\nError parsing schema: {e}") + except Exception as e: + print(f"\nAn unexpected error occurred: {e}") diff --git a/tests/utilities/__init__.py b/tests/utilities/__init__.py new file mode 100644 index 000000000..e17939203 --- /dev/null +++ b/tests/utilities/__init__.py @@ -0,0 +1 @@ +"""Tests for utilities in the fastmcp package.""" diff --git a/tests/utilities/openapi/__init__.py b/tests/utilities/openapi/__init__.py new file mode 100644 index 000000000..0be677d94 --- /dev/null +++ b/tests/utilities/openapi/__init__.py @@ -0,0 +1 @@ +"""Tests for the OpenAPI utilities.""" diff --git a/tests/utilities/openapi/conftest.py b/tests/utilities/openapi/conftest.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/tests/utilities/openapi/conftest.py @@ -0,0 +1 @@ + diff --git a/tests/utilities/openapi/test_openapi.py b/tests/utilities/openapi/test_openapi.py new file mode 100644 index 000000000..0505b5e12 --- /dev/null +++ b/tests/utilities/openapi/test_openapi.py @@ -0,0 +1,709 @@ +"""Tests for the OpenAPI parsing utilities.""" + +from typing import Any, Dict + +import pytest +from fastapi import Body, FastAPI, Path, Query +from pydantic import BaseModel, Field + +from fastmcp.utilities.openapi import parse_openapi_to_http_routes + +# --- Test Data: Static OpenAPI Schema Dictionaries --- # + + +@pytest.fixture +def petstore_schema() -> Dict[str, Any]: + """Fixture that returns a simple Pet Store API schema.""" + return { + "openapi": "3.1.0", + "info": {"title": "Simple Pet Store API", "version": "1.0.0"}, + "paths": { + "/pets": { + "get": { + "summary": "List all pets", + "operationId": "listPets", + "tags": ["pets"], + "parameters": [ + { + "name": "limit", + "in": "query", + "description": "How many items to return", + "required": False, + "schema": {"type": "integer", "format": "int32"}, + } + ], + "responses": {"200": {"description": "A paged array of pets"}}, + }, + "post": { + "summary": "Create a pet", + "operationId": "createPet", + "tags": ["pets"], + "requestBody": {"$ref": "#/components/requestBodies/PetBody"}, + "responses": {"201": {"description": "Null response"}}, + }, + }, + "/pets/{petId}": { + "get": { + "summary": "Info for a specific pet", + "operationId": "showPetById", + "tags": ["pets"], + "parameters": [ + { + "name": "petId", + "in": "path", + "required": True, + "description": "The id of the pet", + "schema": {"type": "string"}, + }, + { + "name": "X-Request-ID", + "in": "header", + "required": False, + "schema": {"type": "string", "format": "uuid"}, + }, + ], + "responses": {"200": {"description": "Information about the pet"}}, + }, + "parameters": [ # Path level parameter example + { + "name": "traceId", + "in": "header", + "description": "Common trace ID", + "required": False, + "schema": {"type": "string"}, + } + ], + }, + }, + "components": { + "schemas": { + "Pet": { + "type": "object", + "required": ["id", "name"], + "properties": { + "id": {"type": "integer", "format": "int64"}, + "name": {"type": "string"}, + "tag": {"type": "string"}, + }, + } + }, + "requestBodies": { + "PetBody": { + "description": "Pet object", + "required": True, + "content": { + "application/json": { + "schema": {"$ref": "#/components/schemas/Pet"} + } + }, + } + }, + }, + } + + +@pytest.fixture +def parsed_petstore_routes(petstore_schema): + """Return parsed routes from the PetStore schema.""" + return parse_openapi_to_http_routes(petstore_schema) + + +@pytest.fixture +def bookstore_schema() -> Dict[str, Any]: + """Fixture that returns a Book Store API schema with different parameter types.""" + return { + "openapi": "3.1.0", + "info": {"title": "Book Store API", "version": "1.0.0"}, + "paths": { + "/books": { + "get": { + "summary": "List all books", + "operationId": "listBooks", + "tags": ["books"], + "parameters": [ + { + "name": "genre", + "in": "query", + "description": "Filter by genre", + "required": False, + "schema": {"type": "string"}, + }, + { + "name": "published_after", + "in": "query", + "description": "Filter by publication date", + "required": False, + "schema": {"type": "string", "format": "date"}, + }, + { + "name": "limit", + "in": "query", + "description": "Maximum number of results", + "required": False, + "schema": {"type": "integer", "default": 10}, + }, + ], + "responses": {"200": {"description": "A list of books"}}, + }, + "post": { + "summary": "Create a new book", + "operationId": "createBook", + "tags": ["books"], + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["title", "author"], + "properties": { + "title": {"type": "string"}, + "author": {"type": "string"}, + "isbn": {"type": "string"}, + "published": { + "type": "string", + "format": "date", + }, + "genre": {"type": "string"}, + }, + } + } + }, + }, + "responses": {"201": {"description": "Book created"}}, + }, + }, + "/books/{isbn}": { + "get": { + "summary": "Get book by ISBN", + "operationId": "getBook", + "tags": ["books"], + "parameters": [ + { + "name": "isbn", + "in": "path", + "required": True, + "description": "ISBN of the book", + "schema": {"type": "string"}, + } + ], + "responses": {"200": {"description": "Book details"}}, + }, + "delete": { + "summary": "Delete a book", + "operationId": "deleteBook", + "tags": ["books"], + "parameters": [ + { + "name": "isbn", + "in": "path", + "required": True, + "description": "ISBN of the book to delete", + "schema": {"type": "string"}, + } + ], + "responses": {"204": {"description": "Book deleted"}}, + }, + }, + }, + } + + +@pytest.fixture +def parsed_bookstore_routes(bookstore_schema): + """Return parsed routes from the BookStore schema.""" + return parse_openapi_to_http_routes(bookstore_schema) + + +# --- FastAPI App Fixtures --- # + + +class Item(BaseModel): + """Example pydantic model for API testing.""" + + name: str + description: str | None = None + price: float + tax: float | None = None + tags: list[str] = Field(default_factory=list) + + +@pytest.fixture +def fastapi_app() -> FastAPI: + """Fixture that returns a FastAPI app with various types of endpoints.""" + app = FastAPI(title="Test API", version="1.0.0") + + @app.get("/items/", operation_id="list_items") + async def list_items(skip: int = 0, limit: int = 10): + """List all items with pagination.""" + return [ + {"name": f"Item {i}", "price": float(i)} for i in range(skip, skip + limit) + ] + + @app.post("/items/", operation_id="create_item") + async def create_item(item: Item): + """Create a new item.""" + return item + + @app.get("/items/{item_id}", operation_id="get_item") + async def get_item( + item_id: int = Path(..., description="The ID of the item to get"), + q: str | None = Query(None, description="Optional query string"), + ): + """Get an item by ID.""" + return {"item_id": item_id, "q": q} + + @app.put("/items/{item_id}", operation_id="update_item") + async def update_item( + item_id: int = Path(..., description="The ID of the item to update"), + item: Item = Body(..., description="The updated item data"), + ): + """Update an existing item.""" + return {"item_id": item_id, **item.model_dump()} + + @app.delete("/items/{item_id}", operation_id="delete_item") + async def delete_item( + item_id: int = Path(..., description="The ID of the item to delete"), + ): + """Delete an item by ID.""" + return {"item_id": item_id, "deleted": True} + + @app.get("/items/{item_id}/tags/{tag_id}", operation_id="get_item_tag") + async def get_item_tag( + item_id: int = Path(..., description="The ID of the item"), + tag_id: str = Path(..., description="The ID of the tag"), + ): + """Get a specific tag for an item.""" + return {"item_id": item_id, "tag_id": tag_id} + + @app.post("/upload/", operation_id="upload_file") + async def upload_file( + file_name: str = Query(..., description="Name of the file to upload"), + content_type: str = Query(..., description="Content type of the file"), + ): + """Upload a file (dummy endpoint for testing query params with POST).""" + return { + "file_name": file_name, + "content_type": content_type, + "status": "uploaded", + } + + return app + + +@pytest.fixture +def fastapi_openapi_schema(fastapi_app) -> Dict[str, Any]: + """Fixture that returns the OpenAPI schema of the FastAPI app.""" + return fastapi_app.openapi() + + +@pytest.fixture +def parsed_fastapi_routes(fastapi_openapi_schema): + """Return parsed routes from a FastAPI OpenAPI schema.""" + return parse_openapi_to_http_routes(fastapi_openapi_schema) + + +@pytest.fixture +def fastapi_route_map(parsed_fastapi_routes): + """Return a dictionary of routes by operation ID.""" + return { + r.operation_id: r for r in parsed_fastapi_routes if r.operation_id is not None + } + + +# --- Tests for PetStore schema --- # + + +def test_petstore_route_count(parsed_petstore_routes): + """Test that parsing the PetStore schema correctly identifies the number of routes.""" + assert len(parsed_petstore_routes) == 3 + + +def test_petstore_get_pets_operation_id(parsed_petstore_routes): + """Test that GET /pets operation_id is correctly parsed.""" + get_pets = next( + (r for r in parsed_petstore_routes if r.method == "GET" and r.path == "/pets"), + None, + ) + assert get_pets is not None + assert get_pets.operation_id == "listPets" + + +def test_petstore_query_parameter(parsed_petstore_routes): + """Test that query parameter 'limit' is correctly parsed from the schema.""" + get_pets = next( + (r for r in parsed_petstore_routes if r.method == "GET" and r.path == "/pets"), + None, + ) + + assert get_pets is not None + assert len(get_pets.parameters) == 1 + param = get_pets.parameters[0] + assert param.name == "limit" + assert param.location == "query" + assert param.required is False + assert param.schema_.get("type") == "integer" + assert param.schema_.get("format") == "int32" + + +def test_petstore_path_parameter(parsed_petstore_routes): + """Test that path parameter 'petId' is correctly parsed from the schema.""" + get_pet = next( + ( + r + for r in parsed_petstore_routes + if r.method == "GET" and r.path == "/pets/{petId}" + ), + None, + ) + + assert get_pet is not None + path_param = next((p for p in get_pet.parameters if p.name == "petId"), None) + assert path_param is not None + assert path_param.location == "path" + assert path_param.required is True + assert path_param.schema_.get("type") == "string" + + +def test_petstore_header_parameters(parsed_petstore_routes): + """Test that header parameters are correctly parsed from the schema.""" + get_pet = next( + ( + r + for r in parsed_petstore_routes + if r.method == "GET" and r.path == "/pets/{petId}" + ), + None, + ) + + assert get_pet is not None + header_params = [p for p in get_pet.parameters if p.location == "header"] + assert len(header_params) == 2 + + +def test_petstore_header_parameter_names(parsed_petstore_routes): + """Test that header parameter names are correctly parsed.""" + get_pet = next( + ( + r + for r in parsed_petstore_routes + if r.method == "GET" and r.path == "/pets/{petId}" + ), + None, + ) + + assert get_pet is not None + header_params = [p for p in get_pet.parameters if p.location == "header"] + header_names = [p.name for p in header_params] + assert "X-Request-ID" in header_names + assert "traceId" in header_names + + +def test_petstore_path_level_parameters(parsed_petstore_routes): + """Test that path-level parameters are correctly merged into the operation.""" + get_pet = next( + ( + r + for r in parsed_petstore_routes + if r.method == "GET" and r.path == "/pets/{petId}" + ), + None, + ) + + assert get_pet is not None + trace_param = next((p for p in get_pet.parameters if p.name == "traceId"), None) + assert trace_param is not None + assert trace_param.location == "header" + assert trace_param.required is False + + +def test_petstore_request_body_reference_resolution(parsed_petstore_routes): + """Test that request body references are correctly resolved.""" + create_pet = next( + (r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"), + None, + ) + + assert create_pet is not None + assert create_pet.request_body is not None + assert create_pet.request_body.required is True + assert "application/json" in create_pet.request_body.content_schema + + +def test_petstore_schema_reference_resolution(parsed_petstore_routes): + """Test that schema references in request bodies are correctly resolved.""" + create_pet = next( + (r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"), + None, + ) + + assert create_pet is not None + assert create_pet.request_body is not None + json_schema = create_pet.request_body.content_schema["application/json"] + properties = json_schema.get("properties", {}) + + assert "id" in properties + assert "name" in properties + assert "tag" in properties + + +def test_petstore_required_fields_resolution(parsed_petstore_routes): + """Test that required fields are correctly resolved from referenced schemas.""" + create_pet = next( + (r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"), + None, + ) + + assert create_pet is not None + assert create_pet.request_body is not None + json_schema = create_pet.request_body.content_schema["application/json"] + assert json_schema.get("required") == ["id", "name"] + + +# --- Tests for BookStore schema --- # + + +def test_bookstore_route_count(parsed_bookstore_routes): + """Test that parsing the BookStore schema correctly identifies the number of routes.""" + assert len(parsed_bookstore_routes) == 4 + + +def test_bookstore_query_parameter_count(parsed_bookstore_routes): + """Test that the correct number of query parameters are parsed.""" + list_books = next( + (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None + ) + + assert list_books is not None + assert len(list_books.parameters) == 3 + + +def test_bookstore_query_parameter_names(parsed_bookstore_routes): + """Test that query parameter names are correctly parsed.""" + list_books = next( + (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None + ) + + assert list_books is not None + param_map = {p.name: p for p in list_books.parameters} + assert "genre" in param_map + assert "published_after" in param_map + assert "limit" in param_map + + +def test_bookstore_query_parameter_formats(parsed_bookstore_routes): + """Test that query parameter formats are correctly parsed.""" + list_books = next( + (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None + ) + + assert list_books is not None + param_map = {p.name: p for p in list_books.parameters} + assert param_map["published_after"].schema_.get("format") == "date" + + +def test_bookstore_query_parameter_defaults(parsed_bookstore_routes): + """Test that query parameter default values are correctly parsed.""" + list_books = next( + (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None + ) + + assert list_books is not None + param_map = {p.name: p for p in list_books.parameters} + assert param_map["limit"].schema_.get("default") == 10 + + +def test_bookstore_inline_request_body_presence(parsed_bookstore_routes): + """Test that request bodies with inline schemas are present.""" + create_book = next( + (r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None + ) + + assert create_book is not None + assert create_book.request_body is not None + assert create_book.request_body.required is True + assert "application/json" in create_book.request_body.content_schema + + +def test_bookstore_inline_request_body_properties(parsed_bookstore_routes): + """Test that request body properties are correctly parsed from inline schemas.""" + create_book = next( + (r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None + ) + + assert create_book is not None + assert create_book.request_body is not None + + json_schema = create_book.request_body.content_schema["application/json"] + properties = json_schema.get("properties", {}) + + assert "title" in properties + assert "author" in properties + assert "isbn" in properties + assert "published" in properties + assert "genre" in properties + + +def test_bookstore_inline_request_body_required_fields(parsed_bookstore_routes): + """Test that required fields in inline schema are correctly parsed.""" + create_book = next( + (r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None + ) + + assert create_book is not None + assert create_book.request_body is not None + + json_schema = create_book.request_body.content_schema["application/json"] + assert json_schema.get("required") == ["title", "author"] + + +def test_bookstore_delete_method(parsed_bookstore_routes): + """Test that DELETE method is correctly parsed from the schema.""" + delete_book = next( + (r for r in parsed_bookstore_routes if r.method == "DELETE"), None + ) + + assert delete_book is not None + assert delete_book.operation_id == "deleteBook" + assert delete_book.path == "/books/{isbn}" + + +def test_bookstore_delete_method_parameters(parsed_bookstore_routes): + """Test that parameters for DELETE method are correctly parsed.""" + delete_book = next( + (r for r in parsed_bookstore_routes if r.method == "DELETE"), None + ) + + assert delete_book is not None + assert len(delete_book.parameters) == 1 + assert delete_book.parameters[0].name == "isbn" + + +# --- Tests for FastAPI Generated Schema --- # + + +def test_fastapi_route_count(parsed_fastapi_routes): + """Test that parsing a FastAPI-generated schema correctly identifies the number of routes.""" + assert len(parsed_fastapi_routes) == 7 + + +def test_fastapi_parameter_default_values(fastapi_route_map): + """Test that default parameter values are correctly parsed from the schema.""" + list_items = fastapi_route_map["list_items"] + + param_map = {p.name: p for p in list_items.parameters} + assert "skip" in param_map + assert "limit" in param_map + + +def test_fastapi_skip_parameter_default(fastapi_route_map): + """Test that skip parameter default value is correctly parsed.""" + list_items = fastapi_route_map["list_items"] + + param_map = {p.name: p for p in list_items.parameters} + assert param_map["skip"].schema_.get("default") == 0 + + +def test_fastapi_limit_parameter_default(fastapi_route_map): + """Test that limit parameter default value is correctly parsed.""" + list_items = fastapi_route_map["list_items"] + + param_map = {p.name: p for p in list_items.parameters} + assert param_map["limit"].schema_.get("default") == 10 + + +def test_fastapi_request_body_from_pydantic(fastapi_route_map): + """Test that request bodies from Pydantic models are present.""" + create_item = fastapi_route_map["create_item"] + + assert create_item.request_body is not None + assert "application/json" in create_item.request_body.content_schema + + +def test_fastapi_request_body_properties(fastapi_route_map): + """Test that request body properties from Pydantic models are correctly parsed.""" + create_item = fastapi_route_map["create_item"] + + json_schema = create_item.request_body.content_schema["application/json"] + properties = json_schema.get("properties", {}) + + assert "name" in properties + assert "description" in properties + assert "price" in properties + assert "tax" in properties + assert "tags" in properties + + +def test_fastapi_request_body_required_fields(fastapi_route_map): + """Test that required fields from Pydantic models are correctly parsed.""" + create_item = fastapi_route_map["create_item"] + + json_schema = create_item.request_body.content_schema["application/json"] + required = json_schema.get("required", []) + + assert "name" in required + assert "price" in required + + +def test_fastapi_path_parameter_presence(fastapi_route_map): + """Test that path parameters are present in FastAPI schema.""" + get_item = fastapi_route_map["get_item"] + + path_params = [p for p in get_item.parameters if p.location == "path"] + assert len(path_params) == 1 + + +def test_fastapi_path_parameter_properties(fastapi_route_map): + """Test that path parameters properties are correctly parsed.""" + get_item = fastapi_route_map["get_item"] + + path_params = [p for p in get_item.parameters if p.location == "path"] + assert path_params[0].name == "item_id" + assert path_params[0].required is True + + +def test_fastapi_optional_query_parameter(fastapi_route_map): + """Test that optional query parameters are correctly parsed.""" + get_item = fastapi_route_map["get_item"] + + query_params = [p for p in get_item.parameters if p.location == "query"] + assert len(query_params) == 1 + assert query_params[0].name == "q" + assert query_params[0].required is False + + +def test_fastapi_multiple_path_parameter_count(fastapi_route_map): + """Test that multiple path parameters count is correct.""" + get_item_tag = fastapi_route_map["get_item_tag"] + + path_params = [p for p in get_item_tag.parameters if p.location == "path"] + assert len(path_params) == 2 + + +def test_fastapi_multiple_path_parameter_names(fastapi_route_map): + """Test that multiple path parameter names are correctly parsed.""" + get_item_tag = fastapi_route_map["get_item_tag"] + + path_params = [p for p in get_item_tag.parameters if p.location == "path"] + param_names = [p.name for p in path_params] + assert "item_id" in param_names + assert "tag_id" in param_names + + +def test_fastapi_post_with_query_parameters(fastapi_route_map): + """Test that query parameters for POST methods are correctly parsed.""" + upload_file = fastapi_route_map["upload_file"] + + assert upload_file.method == "POST" + query_params = [p for p in upload_file.parameters if p.location == "query"] + assert len(query_params) == 2 + + +def test_fastapi_post_query_parameter_names(fastapi_route_map): + """Test that query parameter names for POST methods are correctly parsed.""" + upload_file = fastapi_route_map["upload_file"] + + query_params = [p for p in upload_file.parameters if p.location == "query"] + param_names = [p.name for p in query_params] + assert "file_name" in param_names + assert "content_type" in param_names diff --git a/tests/utilities/openapi/test_openapi_advanced.py b/tests/utilities/openapi/test_openapi_advanced.py new file mode 100644 index 000000000..bacb5fbbd --- /dev/null +++ b/tests/utilities/openapi/test_openapi_advanced.py @@ -0,0 +1,594 @@ +"""Tests for advanced features of the OpenAPI utilities.""" + +from typing import Any, Dict + +import pytest + +from fastmcp.utilities.openapi import parse_openapi_to_http_routes + + +@pytest.fixture +def complex_schema() -> Dict[str, Any]: + """Fixture that returns a complex OpenAPI schema with nested references.""" + return { + "openapi": "3.1.0", + "info": {"title": "Complex API", "version": "1.0.0"}, + "paths": { + "/users": { + "get": { + "summary": "List all users", + "operationId": "listUsers", + "parameters": [ + {"$ref": "#/components/parameters/PageLimit"}, + {"$ref": "#/components/parameters/PageOffset"}, + ], + "responses": {"200": {"description": "A list of users"}}, + } + }, + "/users/{userId}": { + "get": { + "summary": "Get user by ID", + "operationId": "getUser", + "parameters": [ + {"$ref": "#/components/parameters/UserId"}, + {"$ref": "#/components/parameters/IncludeInactive"}, + ], + "responses": {"200": {"description": "User details"}}, + } + }, + "/users/{userId}/orders": { + "post": { + "summary": "Create order for user", + "operationId": "createOrder", + "parameters": [{"$ref": "#/components/parameters/UserId"}], + "requestBody": {"$ref": "#/components/requestBodies/OrderRequest"}, + "responses": {"201": {"description": "Order created"}}, + } + }, + }, + "components": { + "parameters": { + "UserId": { + "name": "userId", + "in": "path", + "required": True, + "schema": {"type": "string", "format": "uuid"}, + }, + "PageLimit": { + "name": "limit", + "in": "query", + "schema": {"type": "integer", "default": 20, "maximum": 100}, + }, + "PageOffset": { + "name": "offset", + "in": "query", + "schema": {"type": "integer", "default": 0}, + }, + "IncludeInactive": { + "name": "include_inactive", + "in": "query", + "schema": {"type": "boolean", "default": False}, + }, + }, + "schemas": { + "User": { + "type": "object", + "properties": { + "id": {"type": "string", "format": "uuid"}, + "name": {"type": "string"}, + "email": {"type": "string", "format": "email"}, + "role": {"$ref": "#/components/schemas/Role"}, + "address": {"$ref": "#/components/schemas/Address"}, + }, + }, + "Role": { + "type": "string", + "enum": ["admin", "user", "guest"], + }, + "Address": { + "type": "object", + "properties": { + "street": {"type": "string"}, + "city": {"type": "string"}, + "zip": {"type": "string"}, + "country": {"type": "string"}, + }, + }, + "Order": { + "type": "object", + "properties": { + "id": {"type": "string", "format": "uuid"}, + "items": { + "type": "array", + "items": {"$ref": "#/components/schemas/OrderItem"}, + }, + "total": {"type": "number"}, + "status": {"$ref": "#/components/schemas/OrderStatus"}, + }, + }, + "OrderItem": { + "type": "object", + "properties": { + "product_id": {"type": "string", "format": "uuid"}, + "quantity": {"type": "integer"}, + "price": {"type": "number"}, + }, + }, + "OrderStatus": { + "type": "string", + "enum": [ + "pending", + "processing", + "shipped", + "delivered", + "cancelled", + ], + }, + }, + "requestBodies": { + "OrderRequest": { + "description": "Order to create", + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["items"], + "properties": { + "items": { + "type": "array", + "items": { + "$ref": "#/components/schemas/OrderItem" + }, + }, + "notes": {"type": "string"}, + }, + } + } + }, + } + }, + }, + } + + +@pytest.fixture +def parsed_complex_routes(complex_schema): + """Return parsed routes from the complex schema.""" + return parse_openapi_to_http_routes(complex_schema) + + +@pytest.fixture +def complex_route_map(parsed_complex_routes): + """Return a dictionary of routes by operation ID.""" + return { + r.operation_id: r for r in parsed_complex_routes if r.operation_id is not None + } + + +@pytest.fixture +def schema_with_invalid_reference() -> Dict[str, Any]: + """Fixture that returns a schema with an invalid reference.""" + return { + "openapi": "3.1.0", + "info": {"title": "Invalid Reference API", "version": "1.0.0"}, + "paths": { + "/broken-ref": { + "get": { + "summary": "Endpoint with broken reference", + "operationId": "brokenRef", + "parameters": [ + {"$ref": "#/components/parameters/NonExistentParam"} + ], + "responses": {"200": {"description": "Something"}}, + } + } + }, + "components": { + "parameters": {} # Empty parameters object to ensure the reference is broken + }, + } + + +@pytest.fixture +def schema_with_content_params() -> Dict[str, Any]: + """Fixture that returns a schema with content-based parameters (complex parameters).""" + return { + "openapi": "3.1.0", + "info": {"title": "Content Params API", "version": "1.0.0"}, + "paths": { + "/complex-params": { + "post": { + "summary": "Endpoint with complex parameter", + "operationId": "complexParams", + "parameters": [ + { + "name": "filter", + "in": "query", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "field": {"type": "string"}, + "operator": { + "type": "string", + "enum": ["eq", "gt", "lt"], + }, + "value": {"type": "string"}, + }, + } + } + }, + } + ], + "responses": {"200": {"description": "Results"}}, + } + }, + }, + } + + +@pytest.fixture +def parsed_content_param_routes(schema_with_content_params): + """Return parsed routes from the schema with content parameters.""" + return parse_openapi_to_http_routes(schema_with_content_params) + + +@pytest.fixture +def schema_all_http_methods() -> Dict[str, Any]: + """Fixture that returns a schema with all HTTP methods.""" + return { + "openapi": "3.1.0", + "info": {"title": "All Methods API", "version": "1.0.0"}, + "paths": { + "/resource": { + "get": { + "operationId": "getResource", + "responses": {"200": {"description": "Success"}}, + }, + "post": { + "operationId": "createResource", + "responses": {"201": {"description": "Created"}}, + }, + "put": { + "operationId": "updateResource", + "responses": {"200": {"description": "Updated"}}, + }, + "delete": { + "operationId": "deleteResource", + "responses": {"204": {"description": "Deleted"}}, + }, + "patch": { + "operationId": "patchResource", + "responses": {"200": {"description": "Patched"}}, + }, + "head": { + "operationId": "headResource", + "responses": {"200": {"description": "Headers only"}}, + }, + "options": { + "operationId": "optionsResource", + "responses": {"200": {"description": "Options"}}, + }, + "trace": { + "operationId": "traceResource", + "responses": {"200": {"description": "Trace"}}, + }, + }, + }, + } + + +@pytest.fixture +def parsed_http_methods_routes(schema_all_http_methods): + """Return parsed routes from the schema with all HTTP methods.""" + return parse_openapi_to_http_routes(schema_all_http_methods) + + +# --- Tests for complex schemas with references --- # + + +def test_complex_schema_route_count(parsed_complex_routes): + """Test that parsing a schema with references successfully extracts all routes.""" + assert len(parsed_complex_routes) == 3 + + +def test_complex_schema_list_users_query_param_limit(complex_route_map): + """Test that a reference to a limit query parameter is correctly resolved.""" + list_users = complex_route_map["listUsers"] + + limit_param = next((p for p in list_users.parameters if p.name == "limit"), None) + assert limit_param is not None + assert limit_param.location == "query" + assert limit_param.schema_.get("default") == 20 + + +def test_complex_schema_list_users_query_param_limit_maximum(complex_route_map): + """Test that a limit parameter's maximum value is correctly resolved.""" + list_users = complex_route_map["listUsers"] + + limit_param = next((p for p in list_users.parameters if p.name == "limit"), None) + assert limit_param is not None + assert limit_param.schema_.get("maximum") == 100 + + +def test_complex_schema_get_user_path_param_existence(complex_route_map): + """Test that a reference to a path parameter exists.""" + get_user = complex_route_map["getUser"] + + user_id_param = next((p for p in get_user.parameters if p.name == "userId"), None) + assert user_id_param is not None + assert user_id_param.location == "path" + + +def test_complex_schema_get_user_path_param_required(complex_route_map): + """Test that a path parameter is correctly marked as required.""" + get_user = complex_route_map["getUser"] + + user_id_param = next((p for p in get_user.parameters if p.name == "userId"), None) + assert user_id_param is not None + assert user_id_param.required is True + + +def test_complex_schema_get_user_path_param_format(complex_route_map): + """Test that a path parameter format is correctly resolved.""" + get_user = complex_route_map["getUser"] + + user_id_param = next((p for p in get_user.parameters if p.name == "userId"), None) + assert user_id_param is not None + assert user_id_param.schema_.get("format") == "uuid" + + +def test_complex_schema_create_order_request_body_presence(complex_route_map): + """Test that a reference to a request body is resolved correctly.""" + create_order = complex_route_map["createOrder"] + + assert create_order.request_body is not None + assert create_order.request_body.required is True + + +def test_complex_schema_create_order_request_body_content_type(complex_route_map): + """Test that request body content type is correctly resolved.""" + create_order = complex_route_map["createOrder"] + + assert create_order.request_body is not None + assert "application/json" in create_order.request_body.content_schema + + +def test_complex_schema_create_order_request_body_properties(complex_route_map): + """Test that request body properties are correctly resolved.""" + create_order = complex_route_map["createOrder"] + + assert create_order.request_body is not None + json_schema = create_order.request_body.content_schema["application/json"] + assert "items" in json_schema.get("properties", {}) + + +def test_complex_schema_create_order_request_body_required_fields(complex_route_map): + """Test that request body required fields are correctly resolved.""" + create_order = complex_route_map["createOrder"] + + assert create_order.request_body is not None + json_schema = create_order.request_body.content_schema["application/json"] + assert json_schema.get("required") == ["items"] + + +# --- Tests for schema reference resolution errors --- # + + +def test_parser_handles_broken_references(schema_with_invalid_reference): + """Test that parser handles broken references gracefully.""" + # We're just checking that the function doesn't throw an exception + routes = parse_openapi_to_http_routes(schema_with_invalid_reference) + + # Should still return routes list (may be empty) + assert isinstance(routes, list) + + # Verify that the route with broken parameter reference is still included + # though it may not have the parameter properly + broken_route = next( + (r for r in routes if r.path == "/broken-ref" and r.method == "GET"), None + ) + + # The route should still be present + assert broken_route is not None + assert broken_route.operation_id == "brokenRef" + + +# --- Tests for content-based parameters --- # + + +def test_content_param_parameter_name(parsed_content_param_routes): + """Test that parser correctly extracts name for content-based parameters.""" + complex_params = parsed_content_param_routes[0] + + assert len(complex_params.parameters) == 1 + param = complex_params.parameters[0] + assert param.name == "filter" + + +def test_content_param_parameter_location(parsed_content_param_routes): + """Test that parser correctly extracts location for content-based parameters.""" + complex_params = parsed_content_param_routes[0] + + assert len(complex_params.parameters) == 1 + param = complex_params.parameters[0] + assert param.location == "query" + + +def test_content_param_schema_properties_presence(parsed_content_param_routes): + """Test that parser extracts schema properties from content-based parameter.""" + complex_params = parsed_content_param_routes[0] + + param = complex_params.parameters[0] + properties = param.schema_.get("properties", {}) + + assert "field" in properties + assert "operator" in properties + assert "value" in properties + + +def test_content_param_schema_enum_presence(parsed_content_param_routes): + """Test that parser extracts enum values from content-based parameter.""" + complex_params = parsed_content_param_routes[0] + + param = complex_params.parameters[0] + properties = param.schema_.get("properties", {}) + + assert "enum" in properties.get("operator", {}) + + +# --- Tests for HTTP methods --- # + + +def test_http_get_method_presence(parsed_http_methods_routes): + """Test that GET method is correctly extracted.""" + get_route = next((r for r in parsed_http_methods_routes if r.method == "GET"), None) + + assert get_route is not None + assert get_route.operation_id == "getResource" + + +def test_http_get_method_path(parsed_http_methods_routes): + """Test that GET method path is correctly extracted.""" + get_route = next((r for r in parsed_http_methods_routes if r.method == "GET"), None) + + assert get_route is not None + assert get_route.path == "/resource" + + +def test_http_post_method_presence(parsed_http_methods_routes): + """Test that POST method is correctly extracted.""" + post_route = next( + (r for r in parsed_http_methods_routes if r.method == "POST"), None + ) + + assert post_route is not None + assert post_route.operation_id == "createResource" + + +def test_http_post_method_path(parsed_http_methods_routes): + """Test that POST method path is correctly extracted.""" + post_route = next( + (r for r in parsed_http_methods_routes if r.method == "POST"), None + ) + + assert post_route is not None + assert post_route.path == "/resource" + + +def test_http_put_method_presence(parsed_http_methods_routes): + """Test that PUT method is correctly extracted.""" + put_route = next((r for r in parsed_http_methods_routes if r.method == "PUT"), None) + + assert put_route is not None + assert put_route.operation_id == "updateResource" + + +def test_http_put_method_path(parsed_http_methods_routes): + """Test that PUT method path is correctly extracted.""" + put_route = next((r for r in parsed_http_methods_routes if r.method == "PUT"), None) + + assert put_route is not None + assert put_route.path == "/resource" + + +def test_http_delete_method_presence(parsed_http_methods_routes): + """Test that DELETE method is correctly extracted.""" + delete_route = next( + (r for r in parsed_http_methods_routes if r.method == "DELETE"), None + ) + + assert delete_route is not None + assert delete_route.operation_id == "deleteResource" + + +def test_http_delete_method_path(parsed_http_methods_routes): + """Test that DELETE method path is correctly extracted.""" + delete_route = next( + (r for r in parsed_http_methods_routes if r.method == "DELETE"), None + ) + + assert delete_route is not None + assert delete_route.path == "/resource" + + +def test_http_patch_method_presence(parsed_http_methods_routes): + """Test that PATCH method is correctly extracted.""" + patch_route = next( + (r for r in parsed_http_methods_routes if r.method == "PATCH"), None + ) + + assert patch_route is not None + assert patch_route.operation_id == "patchResource" + + +def test_http_patch_method_path(parsed_http_methods_routes): + """Test that PATCH method path is correctly extracted.""" + patch_route = next( + (r for r in parsed_http_methods_routes if r.method == "PATCH"), None + ) + + assert patch_route is not None + assert patch_route.path == "/resource" + + +def test_http_head_method_presence(parsed_http_methods_routes): + """Test that HEAD method is correctly extracted.""" + head_route = next( + (r for r in parsed_http_methods_routes if r.method == "HEAD"), None + ) + + assert head_route is not None + assert head_route.operation_id == "headResource" + + +def test_http_head_method_path(parsed_http_methods_routes): + """Test that HEAD method path is correctly extracted.""" + head_route = next( + (r for r in parsed_http_methods_routes if r.method == "HEAD"), None + ) + + assert head_route is not None + assert head_route.path == "/resource" + + +def test_http_options_method_presence(parsed_http_methods_routes): + """Test that OPTIONS method is correctly extracted.""" + options_route = next( + (r for r in parsed_http_methods_routes if r.method == "OPTIONS"), None + ) + + assert options_route is not None + assert options_route.operation_id == "optionsResource" + + +def test_http_options_method_path(parsed_http_methods_routes): + """Test that OPTIONS method path is correctly extracted.""" + options_route = next( + (r for r in parsed_http_methods_routes if r.method == "OPTIONS"), None + ) + + assert options_route is not None + assert options_route.path == "/resource" + + +def test_http_trace_method_presence(parsed_http_methods_routes): + """Test that TRACE method is correctly extracted.""" + trace_route = next( + (r for r in parsed_http_methods_routes if r.method == "TRACE"), None + ) + + assert trace_route is not None + assert trace_route.operation_id == "traceResource" + + +def test_http_trace_method_path(parsed_http_methods_routes): + """Test that TRACE method path is correctly extracted.""" + trace_route = next( + (r for r in parsed_http_methods_routes if r.method == "TRACE"), None + ) + + assert trace_route is not None + assert trace_route.path == "/resource" diff --git a/tests/utilities/openapi/test_openapi_fastapi.py b/tests/utilities/openapi/test_openapi_fastapi.py new file mode 100644 index 000000000..0faf3103d --- /dev/null +++ b/tests/utilities/openapi/test_openapi_fastapi.py @@ -0,0 +1,437 @@ +"""Tests for FastAPI integration with the OpenAPI utilities.""" + +from typing import Any, Dict + +import pytest +from fastapi import FastAPI + +from fastmcp.utilities.openapi import parse_openapi_to_http_routes + + +@pytest.fixture +def fastapi_server() -> FastAPI: + """Fixture that returns a FastAPI app for live OpenAPI schema testing.""" + from enum import Enum + from typing import List, Optional + + from fastapi import Body, Depends, Header, HTTPException, Path, Query + from pydantic import BaseModel, Field + + class ItemStatus(str, Enum): + available = "available" + pending = "pending" + sold = "sold" + + class Tag(BaseModel): + id: int + name: str + + class Item(BaseModel): + """Example pydantic model for testing OpenAPI schema generation.""" + + name: str + description: Optional[str] = None + price: float + tax: Optional[float] = None + tags: List[str] = Field(default_factory=list) + status: ItemStatus = ItemStatus.available + dimensions: Optional[Dict[str, float]] = None + + # Create a FastAPI app with comprehensive features + app = FastAPI( + title="Comprehensive Test API", + description="A test API with various OpenAPI features", + version="1.0.0", + ) + + def get_token_header( + x_token: str = Header(..., description="Authentication token"), + ): + """Example dependency function for header validation.""" + if x_token != "fake-super-secret-token": + raise HTTPException(status_code=400, detail="X-Token header invalid") + return x_token + + TokenDep = Depends(get_token_header) + + @app.get( + "/items/", + operation_id="list_items", + summary="List all items", + description="Get a list of all items with optional filtering", + tags=["items"], + ) + async def list_items( + skip: int = Query(0, description="Number of items to skip"), + limit: int = Query(10, description="Max number of items to return"), + status: Optional[ItemStatus] = Query( + None, description="Filter items by status" + ), + ): + """List all items with pagination and optional status filtering.""" + fake_items = [ + {"name": f"Item {i}", "price": float(i)} for i in range(skip, skip + limit) + ] + if status: + fake_items = [item for item in fake_items if item.get("status") == status] + return fake_items + + @app.post( + "/items/", + operation_id="create_item", + summary="Create a new item", + tags=["items"], + status_code=201, + ) + async def create_item( + item: Item = Body(..., description="Item to create"), + x_token: str = TokenDep, + ): + """Create a new item (requires authentication).""" + return item + + @app.get( + "/items/{item_id}", + operation_id="get_item", + summary="Get a specific item by ID", + tags=["items"], + ) + async def get_item( + item_id: int = Path(..., description="The ID of the item to retrieve"), + include_tax: bool = Query( + False, description="Whether to include tax information" + ), + ): + """Get details about a specific item.""" + item = { + "id": item_id, + "name": f"Item {item_id}", + "price": float(item_id) * 10.0, + } + if include_tax: + item["tax"] = item["price"] * 0.2 + return item + + @app.put( + "/items/{item_id}", + operation_id="update_item", + summary="Update an existing item", + tags=["items"], + ) + async def update_item( + item_id: int = Path(..., description="The ID of the item to update"), + item: Item = Body(..., description="Updated item data"), + x_token: str = TokenDep, + ): + """Update an existing item (requires authentication).""" + return {"item_id": item_id, **item.model_dump()} + + @app.delete( + "/items/{item_id}", + operation_id="delete_item", + summary="Delete an item", + tags=["items"], + ) + async def delete_item( + item_id: int = Path(..., description="The ID of the item to delete"), + x_token: str = TokenDep, + ): + """Delete an item (requires authentication).""" + return {"item_id": item_id, "deleted": True} + + @app.patch( + "/items/{item_id}/tags", + operation_id="update_item_tags", + summary="Update item tags", + tags=["items", "tags"], + ) + async def update_item_tags( + item_id: int = Path(..., description="The ID of the item"), + tags: List[str] = Body(..., description="Updated tags"), + ): + """Update just the tags of an item.""" + return {"item_id": item_id, "tags": tags} + + @app.get( + "/items/{item_id}/tags/{tag_id}", + operation_id="get_item_tag", + summary="Get a specific tag for an item", + tags=["items", "tags"], + ) + async def get_item_tag( + item_id: int = Path(..., description="The ID of the item"), + tag_id: str = Path(..., description="The ID of the tag"), + ): + """Get a specific tag for an item.""" + return {"item_id": item_id, "tag_id": tag_id} + + @app.post( + "/upload/", + operation_id="upload_file", + summary="Upload a file", + tags=["files"], + ) + async def upload_file( + file_name: str = Query(..., description="Name of the file"), + content_type: str = Query(..., description="Content type of the file"), + ): + """Upload a file (dummy endpoint for testing query params).""" + return { + "file_name": file_name, + "content_type": content_type, + "status": "uploaded", + } + + # Add a callback route for testing complex documentation + @app.post( + "/webhook", + operation_id="register_webhook", + summary="Register a webhook", + tags=["webhooks"], + callbacks={ # type: ignore + "itemProcessed": { + "{$request.body.callbackUrl}": { + "post": { + "summary": "Callback for when an item is processed", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "item_id": {"type": "integer"}, + "status": {"type": "string"}, + "timestamp": { + "type": "string", + "format": "date-time", + }, + }, + } + } + }, + }, + "responses": { + "200": {"description": "Webhook processed successfully"} + }, + } + } + } + }, + ) + async def register_webhook( + callback_url: str = Body( + ..., embed=True, description="URL to call when processing completes" + ), + ): + """Register a webhook for processing notifications.""" + return {"registered": True, "callback_url": callback_url} + + return app + + +@pytest.fixture +def fastapi_openapi_schema(fastapi_server) -> Dict[str, Any]: + """Fixture that returns the OpenAPI schema from a live FastAPI server.""" + return fastapi_server.openapi() + + +@pytest.fixture +def parsed_routes(fastapi_openapi_schema): + """Return parsed routes from a FastAPI OpenAPI schema.""" + return parse_openapi_to_http_routes(fastapi_openapi_schema) + + +@pytest.fixture +def route_map(parsed_routes): + """Return a dictionary of routes by operation ID.""" + return {r.operation_id: r for r in parsed_routes if r.operation_id is not None} + + +def test_parse_fastapi_schema_route_count(parsed_routes): + """Test that all routes are parsed from the FastAPI schema.""" + assert len(parsed_routes) == 9 # 8 endpoints + 1 callback + + +def test_parse_fastapi_schema_operation_ids(route_map): + """Test that all expected operation IDs are present in the parsed schema.""" + expected_operations = [ + "list_items", + "create_item", + "get_item", + "update_item", + "delete_item", + "update_item_tags", + "get_item_tag", + "upload_file", + "register_webhook", + ] + + for op_id in expected_operations: + assert op_id in route_map, f"Operation ID '{op_id}' not found in parsed routes" + + +def test_path_parameter_parsing(route_map): + """Test that path parameters are correctly parsed.""" + get_item = route_map["get_item"] + path_params = [p for p in get_item.parameters if p.location == "path"] + + assert len(path_params) == 1 + assert path_params[0].name == "item_id" + assert path_params[0].required is True + + +def test_query_parameter_parsing(route_map): + """Test that query parameters are correctly parsed.""" + list_items = route_map["list_items"] + query_params = [p for p in list_items.parameters if p.location == "query"] + + assert len(query_params) == 3 # skip, limit, status + param_names = [p.name for p in query_params] + assert "skip" in param_names + assert "limit" in param_names + assert "status" in param_names + + +def test_header_parameter_parsing(route_map): + """Test that header parameters from dependencies are correctly parsed.""" + create_item = route_map["create_item"] + header_params = [p for p in create_item.parameters if p.location == "header"] + + assert len(header_params) == 1 + assert header_params[0].name == "x-token" + assert header_params[0].required is True + + +def test_request_body_content_type(route_map): + """Test that request body content types are correctly parsed.""" + create_item = route_map["create_item"] + + assert create_item.request_body is not None + assert "application/json" in create_item.request_body.content_schema + + +def test_request_body_properties(route_map): + """Test that request body properties are correctly parsed.""" + create_item = route_map["create_item"] + json_schema = create_item.request_body.content_schema["application/json"] + properties = json_schema.get("properties", {}) + + assert "name" in properties + assert "price" in properties + assert "description" in properties + assert "tags" in properties + assert "status" in properties + + +def test_request_body_status_schema(route_map): + """Test that the status schema in request body is correctly handled.""" + create_item = route_map["create_item"] + json_schema = create_item.request_body.content_schema["application/json"] + properties = json_schema.get("properties", {}) + status_schema = properties.get("status", {}) + + # FastAPI may represent enums as references or directly include enum values + assert "$ref" in status_schema or "enum" in status_schema + + +def test_route_with_items_tag(parsed_routes): + """Test that routes with 'items' tag are correctly parsed.""" + item_routes = [r for r in parsed_routes if "items" in r.tags] + + assert len(item_routes) >= 6 # At least 6 endpoints with "items" tag + + +def test_routes_with_multiple_tags(parsed_routes): + """Test that routes with multiple tags are correctly parsed.""" + multi_tag_routes = [r for r in parsed_routes if len(r.tags) > 1] + + assert len(multi_tag_routes) >= 2 # At least 2 endpoints with multiple tags + + +def test_specific_route_tags(route_map): + """Test that specific routes have the expected tags.""" + assert "items" in route_map["list_items"].tags + assert "items" in route_map["update_item_tags"].tags + assert "tags" in route_map["update_item_tags"].tags + assert "webhooks" in route_map["register_webhook"].tags + + +def test_operation_summary(route_map): + """Test that operation summary is correctly parsed.""" + list_items = route_map["list_items"] + + assert list_items.summary == "List all items" + + +def test_operation_description(route_map): + """Test that operation description is correctly parsed.""" + list_items = route_map["list_items"] + + assert list_items.description is not None + assert "optional filtering" in list_items.description + + +def test_path_with_route_parameters(route_map): + """Test that paths with route parameters are correctly parsed.""" + get_item = route_map["get_item"] + + assert get_item.path == "/items/{item_id}" + + +def test_complex_nested_paths(route_map): + """Test that complex nested paths are correctly parsed.""" + get_item_tag = route_map["get_item_tag"] + + assert get_item_tag.path == "/items/{item_id}/tags/{tag_id}" + + +def test_http_methods(route_map): + """Test that HTTP methods are correctly parsed.""" + assert route_map["list_items"].method == "GET" + assert route_map["create_item"].method == "POST" + assert route_map["update_item"].method == "PUT" + assert route_map["delete_item"].method == "DELETE" + assert route_map["update_item_tags"].method == "PATCH" + + +def test_item_schema_properties(route_map): + """Test that Item schema properties are correctly resolved.""" + create_item = route_map["create_item"] + json_schema = create_item.request_body.content_schema["application/json"] + properties = json_schema.get("properties", {}) + + assert "name" in properties + assert properties["name"]["type"] == "string" + assert "price" in properties + assert properties["price"]["type"] == "number" + + +def test_webhook_endpoint(route_map): + """Test parsing of webhook registration endpoint.""" + webhook = route_map["register_webhook"] + + assert webhook.method == "POST" + assert webhook.path == "/webhook" + + +def test_webhook_request_body(route_map): + """Test that webhook request body is correctly parsed.""" + webhook = route_map["register_webhook"] + + assert webhook.request_body is not None + assert "application/json" in webhook.request_body.content_schema + json_schema = webhook.request_body.content_schema["application/json"] + assert "callback_url" in json_schema.get("properties", {}) + + +def test_token_dependency_handling(route_map): + """Test that token dependencies are correctly handled in parsed endpoints.""" + token_endpoints = ["create_item", "update_item", "delete_item"] + + for op_id in token_endpoints: + route = route_map[op_id] + header_params = [p for p in route.parameters if p.location == "header"] + token_headers = [p for p in header_params if p.name == "x-token"] + assert len(token_headers) == 1, f"Expected x-token header in {op_id}" + assert token_headers[0].required is True