Spaces:
Running
Running
| """Unit tests for schema processing and parameter mapping.""" | |
| import pytest | |
| from fastmcp.experimental.utilities.openapi.models import ( | |
| HTTPRoute, | |
| ParameterInfo, | |
| RequestBodyInfo, | |
| ) | |
| from fastmcp.experimental.utilities.openapi.schemas import ( | |
| _combine_schemas, | |
| _combine_schemas_and_map_params, | |
| _replace_ref_with_defs, | |
| ) | |
| from fastmcp.utilities.json_schema import compress_schema | |
| class TestSchemaProcessing: | |
| """Test schema processing utilities.""" | |
| def simple_route(self): | |
| """Create a simple route for testing.""" | |
| return HTTPRoute( | |
| path="/users/{id}", | |
| method="GET", | |
| operation_id="get_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| ) | |
| ], | |
| ) | |
| def collision_route(self): | |
| """Create a route with parameter name collisions.""" | |
| return HTTPRoute( | |
| path="/users/{id}", | |
| method="PUT", | |
| operation_id="update_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| description="User ID in path", | |
| ) | |
| ], | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "id": {"type": "integer", "description": "User ID in body"}, | |
| "name": {"type": "string"}, | |
| "email": {"type": "string"}, | |
| }, | |
| "required": ["name"], | |
| } | |
| }, | |
| ), | |
| ) | |
| def complex_route(self): | |
| """Create a complex route with multiple parameter types.""" | |
| return HTTPRoute( | |
| path="/items/{id}", | |
| method="PATCH", | |
| operation_id="update_item", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "string"}, | |
| ), | |
| ParameterInfo( | |
| name="version", | |
| location="query", | |
| required=False, | |
| schema={"type": "integer", "default": 1}, | |
| ), | |
| ParameterInfo( | |
| name="X-Client-Version", | |
| location="header", | |
| required=False, | |
| schema={"type": "string"}, | |
| ), | |
| ], | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "title": {"type": "string"}, | |
| "description": {"type": "string"}, | |
| "tags": { | |
| "type": "array", | |
| "items": {"type": "string"}, | |
| }, | |
| }, | |
| "required": ["title"], | |
| } | |
| }, | |
| ), | |
| ) | |
| def test_combine_schemas_simple(self, simple_route): | |
| """Test combining schemas for a simple route.""" | |
| combined_schema = _combine_schemas(simple_route) | |
| assert combined_schema["type"] == "object" | |
| assert "properties" in combined_schema | |
| properties = combined_schema["properties"] | |
| assert "id" in properties | |
| assert properties["id"]["type"] == "integer" | |
| required = combined_schema.get("required", []) | |
| assert "id" in required | |
| def test_combine_schemas_with_collisions(self, collision_route): | |
| """Test combining schemas with parameter name collisions.""" | |
| combined_schema = _combine_schemas(collision_route) | |
| assert combined_schema["type"] == "object" | |
| properties = combined_schema["properties"] | |
| # Should handle collision by suffixing | |
| id_params = [key for key in properties.keys() if "id" in key] | |
| assert len(id_params) >= 2 # Should have both path and body id | |
| # Should have other body parameters | |
| assert "name" in properties | |
| assert "email" in properties | |
| def test_combine_schemas_complex(self, complex_route): | |
| """Test combining schemas for complex route.""" | |
| combined_schema = _combine_schemas(complex_route) | |
| properties = combined_schema["properties"] | |
| # Should have path parameter | |
| assert "id" in properties | |
| # Should have query parameter | |
| assert "version" in properties | |
| assert properties["version"].get("default") == 1 | |
| # Should have header parameter | |
| assert "X-Client-Version" in properties | |
| # Should have body parameters | |
| assert "title" in properties | |
| assert "description" in properties | |
| assert "tags" in properties | |
| # Check required fields | |
| required = combined_schema.get("required", []) | |
| assert "id" in required # Path parameters are required | |
| assert "title" in required # Required body parameter | |
| def test_combine_schemas_and_map_params_simple(self, simple_route): | |
| """Test combining schemas and creating parameter map.""" | |
| combined_schema, param_map = _combine_schemas_and_map_params(simple_route) | |
| # Check schema | |
| assert combined_schema["type"] == "object" | |
| assert "id" in combined_schema["properties"] | |
| # Check parameter map | |
| assert len(param_map) == 1 | |
| assert "id" in param_map | |
| assert param_map["id"]["location"] == "path" | |
| assert param_map["id"]["openapi_name"] == "id" | |
| def test_combine_schemas_and_map_params_with_collisions(self, collision_route): | |
| """Test parameter mapping with collisions.""" | |
| combined_schema, param_map = _combine_schemas_and_map_params(collision_route) | |
| # Check that we have entries for both conflicting parameters | |
| path_id_key = None | |
| body_id_key = None | |
| for key, mapping in param_map.items(): | |
| if mapping["location"] == "path" and mapping["openapi_name"] == "id": | |
| path_id_key = key | |
| elif mapping["location"] == "body" and mapping["openapi_name"] == "id": | |
| body_id_key = key | |
| assert path_id_key is not None | |
| assert body_id_key is not None | |
| assert path_id_key != body_id_key # Should be different keys | |
| # Both should exist in schema | |
| assert path_id_key in combined_schema["properties"] | |
| assert body_id_key in combined_schema["properties"] | |
| # Should also have non-conflicting parameters | |
| assert "name" in param_map | |
| assert "email" in param_map | |
| def test_combine_schemas_and_map_params_complex(self, complex_route): | |
| """Test parameter mapping for complex route.""" | |
| combined_schema, param_map = _combine_schemas_and_map_params(complex_route) | |
| # Should have all parameters mapped | |
| actual_locations = {mapping["location"] for mapping in param_map.values()} | |
| # Should have representatives from each location | |
| assert "path" in actual_locations | |
| assert "body" in actual_locations | |
| # May or may not have query/header depending on whether they're included | |
| # Check specific mappings | |
| id_mapping = param_map["id"] | |
| assert id_mapping["location"] == "path" | |
| assert id_mapping["openapi_name"] == "id" | |
| title_mapping = param_map["title"] | |
| assert title_mapping["location"] == "body" | |
| assert title_mapping["openapi_name"] == "title" | |
| def test_replace_ref_with_defs(self): | |
| """Test replacing $ref with $defs for JSON Schema compatibility.""" | |
| schema_with_ref = { | |
| "type": "object", | |
| "properties": { | |
| "user": {"$ref": "#/components/schemas/User"}, | |
| "items": { | |
| "type": "array", | |
| "items": {"$ref": "#/components/schemas/Item"}, | |
| }, | |
| }, | |
| } | |
| # Use our recursive replacement approach | |
| result = _replace_ref_with_defs(schema_with_ref) | |
| assert result["properties"]["user"]["$ref"] == "#/$defs/User" | |
| assert result["properties"]["items"]["items"]["$ref"] == "#/$defs/Item" | |
| def test_replace_ref_with_defs_nested(self): | |
| """Test replacing $ref in deeply nested structures.""" | |
| nested_schema = { | |
| "type": "object", | |
| "properties": { | |
| "data": { | |
| "type": "object", | |
| "properties": { | |
| "nested": {"$ref": "#/components/schemas/Nested"}, | |
| }, | |
| }, | |
| "items": { | |
| "type": "array", | |
| "items": { | |
| "type": "object", | |
| "properties": { | |
| "ref_prop": {"$ref": "#/components/schemas/RefProp"}, | |
| }, | |
| }, | |
| }, | |
| }, | |
| } | |
| # Use our recursive replacement approach | |
| result = _replace_ref_with_defs(nested_schema) | |
| # Check nested object property | |
| nested_prop = result["properties"]["data"]["properties"]["nested"] | |
| assert nested_prop["$ref"] == "#/$defs/Nested" | |
| # Check array item property | |
| array_item_prop = result["properties"]["items"]["items"]["properties"][ | |
| "ref_prop" | |
| ] | |
| assert array_item_prop["$ref"] == "#/$defs/RefProp" | |
| def test_parameter_collision_suffixing_logic(self): | |
| """Test the specific logic for parameter collision suffixing.""" | |
| # Create a route that would definitely cause collisions | |
| route = HTTPRoute( | |
| path="/test/{id}", | |
| method="POST", | |
| operation_id="test_collision", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", location="path", required=True, schema={"type": "string"} | |
| ), | |
| ParameterInfo( | |
| name="name", | |
| location="query", | |
| required=False, | |
| schema={"type": "string"}, | |
| ), | |
| ParameterInfo( | |
| name="name", | |
| location="header", | |
| required=False, | |
| schema={"type": "string"}, | |
| ), | |
| ], | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "id": {"type": "integer"}, | |
| "name": {"type": "string"}, | |
| "description": {"type": "string"}, | |
| }, | |
| } | |
| }, | |
| ), | |
| ) | |
| combined_schema, param_map = _combine_schemas_and_map_params(route) | |
| # Check that all parameters are included with unique keys | |
| param_keys = list(param_map.keys()) | |
| assert len(param_keys) == len(set(param_keys)) # All keys should be unique | |
| # Should have some form of id and name parameters | |
| id_keys = [key for key in param_keys if "id" in key] | |
| name_keys = [key for key in param_keys if "name" in key] | |
| assert len(id_keys) >= 2 # Path id and body id | |
| assert len(name_keys) >= 3 # Query name, header name, and body name | |
| # Check that locations are correctly mapped | |
| path_params = [ | |
| key for key, mapping in param_map.items() if mapping["location"] == "path" | |
| ] | |
| query_params = [ | |
| key for key, mapping in param_map.items() if mapping["location"] == "query" | |
| ] | |
| header_params = [ | |
| key for key, mapping in param_map.items() if mapping["location"] == "header" | |
| ] | |
| body_params = [ | |
| key for key, mapping in param_map.items() if mapping["location"] == "body" | |
| ] | |
| assert len(path_params) == 1 | |
| assert len(query_params) == 1 | |
| assert len(header_params) == 1 | |
| assert len(body_params) >= 3 # id, name, description from body | |
| class TestEdgeCases: | |
| """Test edge cases in schema processing.""" | |
| def test_empty_route(self): | |
| """Test schema processing with empty route.""" | |
| empty_route = HTTPRoute( | |
| path="/empty", | |
| method="GET", | |
| operation_id="empty_op", | |
| parameters=[], | |
| ) | |
| combined_schema = _combine_schemas(empty_route) | |
| assert combined_schema["type"] == "object" | |
| assert combined_schema["properties"] == {} | |
| assert combined_schema.get("required", []) == [] | |
| def test_route_without_request_body(self): | |
| """Test route with only parameters, no request body.""" | |
| route = HTTPRoute( | |
| path="/test/{id}", | |
| method="GET", | |
| operation_id="test_get", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", location="path", required=True, schema={"type": "string"} | |
| ), | |
| ParameterInfo( | |
| name="filter", | |
| location="query", | |
| required=False, | |
| schema={"type": "string"}, | |
| ), | |
| ], | |
| ) | |
| combined_schema, param_map = _combine_schemas_and_map_params(route) | |
| assert "id" in combined_schema["properties"] | |
| assert "filter" in combined_schema["properties"] | |
| assert len(param_map) == 2 | |
| def test_route_with_only_request_body(self): | |
| """Test route with only request body, no parameters.""" | |
| route = HTTPRoute( | |
| path="/create", | |
| method="POST", | |
| operation_id="create_item", | |
| parameters=[], | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "name": {"type": "string"}, | |
| "description": {"type": "string"}, | |
| }, | |
| "required": ["name"], | |
| } | |
| }, | |
| ), | |
| ) | |
| combined_schema, param_map = _combine_schemas_and_map_params(route) | |
| assert "name" in combined_schema["properties"] | |
| assert "description" in combined_schema["properties"] | |
| assert "name" in combined_schema["required"] | |
| assert len(param_map) == 2 | |
| def test_parameter_without_schema(self): | |
| """Test handling parameters without schema.""" | |
| # Use minimal schema to avoid validation error | |
| route = HTTPRoute( | |
| path="/test", | |
| method="GET", | |
| operation_id="test_no_schema", | |
| parameters=[ | |
| ParameterInfo( | |
| name="param1", location="query", required=False, schema={} | |
| ), # Empty schema | |
| ], | |
| ) | |
| combined_schema, param_map = _combine_schemas_and_map_params(route) | |
| # Should handle gracefully | |
| assert combined_schema["type"] == "object" | |
| assert isinstance(param_map, dict) | |
| def test_request_body_multiple_content_types(self): | |
| """Test request body with multiple content types.""" | |
| route = HTTPRoute( | |
| path="/upload", | |
| method="POST", | |
| operation_id="upload_file", | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": {"metadata": {"type": "string"}}, | |
| }, | |
| "multipart/form-data": { | |
| "type": "object", | |
| "properties": {"file": {"type": "string", "format": "binary"}}, | |
| }, | |
| }, | |
| ), | |
| ) | |
| combined_schema, param_map = _combine_schemas_and_map_params(route) | |
| # Should use the first content type found | |
| properties = combined_schema["properties"] | |
| assert ( | |
| len(properties) > 0 | |
| ) # Should have some properties from one of the content types | |
| def test_oneof_reference_preserved(self): | |
| """Test that schemas referenced in oneOf are preserved.""" | |
| schema = { | |
| "type": "object", | |
| "properties": {"data": {"oneOf": [{"$ref": "#/$defs/TestSchema"}]}}, | |
| "$defs": { | |
| "TestSchema": {"type": "string"}, | |
| "UnusedSchema": {"type": "number"}, | |
| }, | |
| } | |
| result = compress_schema(schema) | |
| # TestSchema should be preserved (referenced in oneOf) | |
| assert "TestSchema" in result["$defs"] | |
| # UnusedSchema should be removed | |
| assert "UnusedSchema" not in result["$defs"] | |
| def test_anyof_reference_preserved(self): | |
| """Test that schemas referenced in anyOf are preserved.""" | |
| schema = { | |
| "type": "object", | |
| "properties": {"data": {"anyOf": [{"$ref": "#/$defs/TestSchema"}]}}, | |
| "$defs": { | |
| "TestSchema": {"type": "string"}, | |
| "UnusedSchema": {"type": "number"}, | |
| }, | |
| } | |
| result = compress_schema(schema) | |
| assert "TestSchema" in result["$defs"] | |
| assert "UnusedSchema" not in result["$defs"] | |
| def test_allof_reference_preserved(self): | |
| """Test that schemas referenced in allOf are preserved.""" | |
| schema = { | |
| "type": "object", | |
| "properties": {"data": {"allOf": [{"$ref": "#/$defs/TestSchema"}]}}, | |
| "$defs": { | |
| "TestSchema": {"type": "string"}, | |
| "UnusedSchema": {"type": "number"}, | |
| }, | |
| } | |
| result = compress_schema(schema) | |
| assert "TestSchema" in result["$defs"] | |
| assert "UnusedSchema" not in result["$defs"] | |