"""Unit tests for RequestDirector.""" from urllib.parse import unquote import pytest from jsonschema_path import SchemaPath from fastmcp.utilities.openapi.director import RequestDirector from fastmcp.utilities.openapi.models import ( HTTPRoute, ParameterInfo, RequestBodyInfo, ) from fastmcp.utilities.openapi.parser import parse_openapi_to_http_routes class TestRequestDirector: """Test RequestDirector request building functionality.""" @pytest.fixture 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"}}, ) @pytest.fixture 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"}, }, ) @pytest.fixture 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"}, }, ) @pytest.fixture 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") class TestPathTraversalPrevention: """Test that path parameter values are URL-encoded to prevent SSRF/path traversal.""" @pytest.fixture def director(self, basic_openapi_30_spec): spec = SchemaPath.from_dict(basic_openapi_30_spec) return RequestDirector(spec) @pytest.fixture def path_route(self): return HTTPRoute( path="/api/v1/users/{id}/profile", method="GET", operation_id="get_user_profile", parameters=[ ParameterInfo( name="id", location="path", required=True, schema={"type": "string"}, ) ], flat_param_schema={ "type": "object", "properties": {"id": {"type": "string"}}, "required": ["id"], }, parameter_map={"id": {"location": "path", "openapi_name": "id"}}, ) @pytest.mark.parametrize( "malicious_id", [ "../../../admin/delete-all?", "../../secret", "../../../etc/passwd", "foo/../../../admin", "..%2F..%2Fadmin", "..%2f..%2fadmin", ], ) def test_path_traversal_encoded(self, director, path_route, malicious_id: str): request = director.build( path_route, {"id": malicious_id}, "https://api.example.com" ) url = str(request.url) assert "/admin" not in url assert "/secret" not in url assert "/etc/passwd" not in url assert url.startswith("https://api.example.com/api/v1/users/") def test_slash_in_param_is_encoded(self, director, path_route): request = director.build(path_route, {"id": "a/b"}, "https://api.example.com") url = str(request.url) assert "/a/b/" not in url assert "a%2Fb" in url def test_dot_dot_slash_is_encoded(self, director, path_route): request = director.build( path_route, {"id": "../admin"}, "https://api.example.com" ) url = str(request.url) assert "%2E%2E%2Fadmin" in url or "%2e%2e%2fadmin" in url assert url.startswith("https://api.example.com/api/v1/users/") def test_question_mark_encoded(self, director, path_route): request = director.build( path_route, {"id": "foo?bar=baz"}, "https://api.example.com" ) url = str(request.url) assert "foo%3Fbar%3Dbaz" in url or "foo%3fbar%3dbaz" in url def test_hash_encoded(self, director, path_route): request = director.build( path_route, {"id": "foo#fragment"}, "https://api.example.com" ) url = str(request.url) assert "foo%23fragment" in url def test_normal_values_still_work(self, director, path_route): request = director.build( path_route, {"id": "user-123"}, "https://api.example.com" ) assert ( str(request.url) == "https://api.example.com/api/v1/users/user-123/profile" ) def test_dotted_values_encode_dots(self, director, path_route): """Dots are encoded to prevent path normalization by urljoin.""" request = director.build( path_route, {"id": "v1.2.3"}, "https://api.example.com" ) url = str(request.url) assert "v1%2E2%2E3" in url assert url.startswith("https://api.example.com/api/v1/users/") def test_numeric_values_still_work(self, director, path_route): request = director.build(path_route, {"id": 42}, "https://api.example.com") assert str(request.url) == "https://api.example.com/api/v1/users/42/profile" def test_bare_single_dot_encoded(self, director, path_route): """Bare '.' must be encoded so urljoin doesn't normalize it away.""" request = director.build(path_route, {"id": "."}, "https://api.example.com") url = str(request.url) assert "%2E" in url assert url.startswith("https://api.example.com/api/v1/users/") def test_bare_dotdot_encoded(self, director, path_route): """Bare '..' must be encoded so urljoin doesn't resolve it as traversal.""" request = director.build(path_route, {"id": ".."}, "https://api.example.com") url = str(request.url) assert "%2E%2E" in url or "%2e%2e" in url assert url.startswith("https://api.example.com/api/v1/users/") def test_double_encoded_traversal(self, director, path_route): request = director.build( path_route, {"id": "..%2F..%2Fadmin"}, "https://api.example.com", ) url = str(request.url) decoded = unquote(unquote(url)) # Verify traversal didn't escape the users/ prefix assert decoded.startswith("https://api.example.com/api/v1/users/") assert url.startswith("https://api.example.com/api/v1/users/")