Spaces:
Running
Running
| """Unit tests for RequestDirector.""" | |
| import pytest | |
| from jsonschema_path import SchemaPath | |
| from fastmcp.experimental.utilities.openapi.director import RequestDirector | |
| from fastmcp.experimental.utilities.openapi.models import ( | |
| HTTPRoute, | |
| ParameterInfo, | |
| RequestBodyInfo, | |
| ) | |
| from fastmcp.experimental.utilities.openapi.parser import parse_openapi_to_http_routes | |
| class TestRequestDirector: | |
| """Test RequestDirector request building functionality.""" | |
| def basic_route(self): | |
| """Create a basic HTTPRoute for testing.""" | |
| return HTTPRoute( | |
| path="/users/{id}", | |
| method="GET", | |
| operation_id="get_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| ) | |
| ], | |
| flat_param_schema={ | |
| "type": "object", | |
| "properties": {"id": {"type": "integer"}}, | |
| "required": ["id"], | |
| }, | |
| parameter_map={"id": {"location": "path", "openapi_name": "id"}}, | |
| ) | |
| def complex_route(self): | |
| """Create a complex HTTPRoute 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"}, | |
| }, | |
| "required": ["title"], | |
| } | |
| }, | |
| ), | |
| flat_param_schema={ | |
| "type": "object", | |
| "properties": { | |
| "id": {"type": "string"}, | |
| "version": {"type": "integer", "default": 1}, | |
| "X-Client-Version": {"type": "string"}, | |
| "title": {"type": "string"}, | |
| "description": {"type": "string"}, | |
| }, | |
| "required": ["id", "title"], | |
| }, | |
| parameter_map={ | |
| "id": {"location": "path", "openapi_name": "id"}, | |
| "version": {"location": "query", "openapi_name": "version"}, | |
| "X-Client-Version": { | |
| "location": "header", | |
| "openapi_name": "X-Client-Version", | |
| }, | |
| "title": {"location": "body", "openapi_name": "title"}, | |
| "description": {"location": "body", "openapi_name": "description"}, | |
| }, | |
| ) | |
| 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"}, | |
| ) | |
| ], | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "id": {"type": "integer"}, | |
| "name": {"type": "string"}, | |
| }, | |
| "required": ["name"], | |
| } | |
| }, | |
| ), | |
| flat_param_schema={ | |
| "type": "object", | |
| "properties": { | |
| "id__path": {"type": "integer"}, | |
| "id": {"type": "integer"}, | |
| "name": {"type": "string"}, | |
| }, | |
| "required": ["id__path", "name"], | |
| }, | |
| parameter_map={ | |
| "id__path": {"location": "path", "openapi_name": "id"}, | |
| "id": {"location": "body", "openapi_name": "id"}, | |
| "name": {"location": "body", "openapi_name": "name"}, | |
| }, | |
| ) | |
| def director(self, basic_openapi_30_spec): | |
| """Create a RequestDirector instance.""" | |
| spec = SchemaPath.from_dict(basic_openapi_30_spec) | |
| return RequestDirector(spec) | |
| def test_director_initialization(self, basic_openapi_30_spec): | |
| """Test RequestDirector initialization.""" | |
| spec = SchemaPath.from_dict(basic_openapi_30_spec) | |
| director = RequestDirector(spec) | |
| assert director._spec is not None | |
| assert director._spec == spec | |
| def test_build_basic_request(self, director, basic_route): | |
| """Test building a basic GET request with path parameter.""" | |
| flat_args = {"id": 123} | |
| request = director.build(basic_route, flat_args, "https://api.example.com") | |
| assert request.method == "GET" | |
| assert request.url == "https://api.example.com/users/123" | |
| assert ( | |
| request.content == b"" | |
| ) # httpx.Request sets content to empty bytes for GET | |
| def test_build_complex_request(self, director, complex_route): | |
| """Test building a complex request with multiple parameter types.""" | |
| flat_args = { | |
| "id": "item123", | |
| "version": 2, | |
| "X-Client-Version": "1.0.0", | |
| "title": "Updated Title", | |
| "description": "Updated description", | |
| } | |
| request = director.build(complex_route, flat_args, "https://api.example.com") | |
| assert request.method == "PATCH" | |
| assert "item123" in str(request.url) | |
| assert "version=2" in str(request.url) | |
| # Check headers | |
| headers = dict(request.headers) if request.headers else {} | |
| assert ( | |
| headers.get("x-client-version") == "1.0.0" | |
| ) # httpx normalizes headers to lowercase | |
| # Check body | |
| import json | |
| assert request.content is not None | |
| body_data = json.loads(request.content) | |
| assert body_data["title"] == "Updated Title" | |
| assert body_data["description"] == "Updated description" | |
| def test_build_request_with_collisions(self, director, collision_route): | |
| """Test building request with parameter name collisions.""" | |
| flat_args = { | |
| "id__path": 123, # Path parameter | |
| "id": 456, # Body parameter | |
| "name": "John Doe", | |
| } | |
| request = director.build(collision_route, flat_args, "https://api.example.com") | |
| assert request.method == "PUT" | |
| assert "123" in str(request.url) # Path ID should be 123 | |
| # Check body | |
| import json | |
| body_data = json.loads(request.content) | |
| assert body_data["id"] == 456 # Body ID should be 456 | |
| assert body_data["name"] == "John Doe" | |
| def test_build_request_with_none_values(self, director, complex_route): | |
| """Test that None values are skipped for optional parameters.""" | |
| flat_args = { | |
| "id": "item123", | |
| "version": None, # Optional, should be skipped | |
| "X-Client-Version": None, # Optional, should be skipped | |
| "title": "Required Title", | |
| "description": None, # Optional body param, should be skipped | |
| } | |
| request = director.build(complex_route, flat_args, "https://api.example.com") | |
| assert request.method == "PATCH" | |
| assert "item123" in str(request.url) | |
| assert "version" not in str(request.url) # Should not include None version | |
| headers = dict(request.headers) if request.headers else {} | |
| assert "X-Client-Version" not in headers | |
| import json | |
| body_data = json.loads(request.content) | |
| assert body_data["title"] == "Required Title" | |
| assert "description" not in body_data # Should not include None description | |
| def test_build_request_fallback_mapping(self, director): | |
| """Test fallback parameter mapping when parameter_map is not available.""" | |
| # Create route without parameter_map | |
| route_without_map = HTTPRoute( | |
| path="/users/{id}", | |
| method="GET", | |
| operation_id="get_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| ) | |
| ], | |
| # No parameter_map provided | |
| ) | |
| flat_args = {"id": 123} | |
| request = director.build( | |
| route_without_map, flat_args, "https://api.example.com" | |
| ) | |
| assert request.method == "GET" | |
| assert "123" in str(request.url) | |
| def test_build_request_suffixed_parameters(self, director): | |
| """Test handling of suffixed parameters in fallback mode.""" | |
| route = HTTPRoute( | |
| path="/users/{id}", | |
| method="POST", | |
| operation_id="create_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| ) | |
| ], | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": {"name": {"type": "string"}}, | |
| } | |
| }, | |
| ), | |
| ) | |
| # Use suffixed parameter names | |
| flat_args = { | |
| "id__path": 123, | |
| "name": "John Doe", | |
| } | |
| request = director.build(route, flat_args, "https://api.example.com") | |
| assert request.method == "POST" | |
| assert "123" in str(request.url) | |
| import json | |
| body_data = json.loads(request.content) | |
| assert body_data["name"] == "John Doe" | |
| def test_url_building(self, director, basic_route): | |
| """Test URL building with different base URLs.""" | |
| flat_args = {"id": 123} | |
| # Test with trailing slash | |
| request1 = director.build(basic_route, flat_args, "https://api.example.com/") | |
| assert request1.url == "https://api.example.com/users/123" | |
| # Test without trailing slash | |
| request2 = director.build(basic_route, flat_args, "https://api.example.com") | |
| assert request2.url == "https://api.example.com/users/123" | |
| # Test with path in base URL | |
| request3 = director.build(basic_route, flat_args, "https://api.example.com/v1") | |
| assert request3.url == "https://api.example.com/v1/users/123" | |
| def test_body_construction_single_value(self, director): | |
| """Test body construction when body schema is not an object.""" | |
| route = HTTPRoute( | |
| path="/upload", | |
| method="POST", | |
| operation_id="upload_file", | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={"text/plain": {"type": "string"}}, | |
| ), | |
| parameter_map={ | |
| "content": {"location": "body", "openapi_name": "content"}, | |
| }, | |
| ) | |
| flat_args = {"content": "Hello, World!"} | |
| request = director.build(route, flat_args, "https://api.example.com") | |
| assert request.method == "POST" | |
| # For non-JSON content, httpx uses 'content' parameter which becomes bytes | |
| assert request.content == b"Hello, World!" | |
| def test_body_construction_multiple_properties_non_object_schema(self, director): | |
| """Test body construction with multiple properties but non-object schema.""" | |
| route = HTTPRoute( | |
| path="/complex", | |
| method="POST", | |
| operation_id="complex_op", | |
| request_body=RequestBodyInfo( | |
| required=True, | |
| content_schema={ | |
| "application/json": {"type": "string"} # Non-object schema | |
| }, | |
| ), | |
| parameter_map={ | |
| "prop1": {"location": "body", "openapi_name": "prop1"}, | |
| "prop2": {"location": "body", "openapi_name": "prop2"}, | |
| }, | |
| ) | |
| flat_args = {"prop1": "value1", "prop2": "value2"} | |
| request = director.build(route, flat_args, "https://api.example.com") | |
| assert request.method == "POST" | |
| # Should wrap in object when multiple properties but schema is not object | |
| import json | |
| body_data = json.loads(request.content) | |
| assert body_data == {"prop1": "value1", "prop2": "value2"} | |
| class TestRequestDirectorIntegration: | |
| """Test RequestDirector with real parsed routes.""" | |
| def test_with_parsed_routes(self, basic_openapi_30_spec): | |
| """Test RequestDirector with routes parsed from real spec.""" | |
| routes = parse_openapi_to_http_routes(basic_openapi_30_spec) | |
| assert len(routes) == 1 | |
| route = routes[0] | |
| spec = SchemaPath.from_dict(basic_openapi_30_spec) | |
| director = RequestDirector(spec) | |
| flat_args = {"id": 42} | |
| request = director.build(route, flat_args, "https://api.example.com") | |
| assert request.method == "GET" | |
| assert request.url == "https://api.example.com/users/42" | |
| def test_with_collision_spec(self, collision_spec): | |
| """Test RequestDirector with collision spec.""" | |
| routes = parse_openapi_to_http_routes(collision_spec) | |
| assert len(routes) == 1 | |
| route = routes[0] | |
| spec = SchemaPath.from_dict(collision_spec) | |
| director = RequestDirector(spec) | |
| # Use the parameter names from the actual parameter map | |
| param_map = route.parameter_map | |
| path_param_name = None | |
| body_param_names = [] | |
| for param_name, mapping in param_map.items(): | |
| if mapping["location"] == "path" and mapping["openapi_name"] == "id": | |
| path_param_name = param_name | |
| elif mapping["location"] == "body": | |
| body_param_names.append(param_name) | |
| assert path_param_name is not None | |
| flat_args = {path_param_name: 123, "name": "John Doe"} | |
| # Add body id if it exists in the parameter map | |
| for param_name in body_param_names: | |
| if "id" in param_name: | |
| flat_args[param_name] = 456 | |
| request = director.build(route, flat_args, "https://api.example.com") | |
| assert request.method == "PUT" | |
| assert "123" in str(request.url) | |
| def test_with_deepobject_spec(self, deepobject_spec): | |
| """Test RequestDirector with deepObject parameters.""" | |
| routes = parse_openapi_to_http_routes(deepobject_spec) | |
| assert len(routes) == 1 | |
| route = routes[0] | |
| spec = SchemaPath.from_dict(deepobject_spec) | |
| director = RequestDirector(spec) | |
| # DeepObject parameters should be flattened in the parameter map | |
| flat_args = {} | |
| for param_name in route.parameter_map.keys(): | |
| if "filter" in param_name: | |
| # Set some test values based on parameter name | |
| if "category" in param_name: | |
| flat_args[param_name] = "electronics" | |
| elif "min" in param_name: | |
| flat_args[param_name] = 10.0 | |
| elif "max" in param_name: | |
| flat_args[param_name] = 100.0 | |
| if flat_args: # Only test if we have parameters to test with | |
| request = director.build(route, flat_args, "https://api.example.com") | |
| assert request.method == "GET" | |
| assert str(request.url).startswith("https://api.example.com/search") | |