Spaces:
Running
Running
| """Tests for handling parameter name collisions between different OpenAPI parameter locations.""" | |
| from unittest.mock import AsyncMock, MagicMock | |
| import httpx | |
| import pytest | |
| from fastmcp.server.openapi import OpenAPITool | |
| from fastmcp.utilities.openapi import HTTPRoute, ParameterInfo, RequestBodyInfo | |
| def mock_client(): | |
| """Create a mock httpx.AsyncClient.""" | |
| client = AsyncMock(spec=httpx.AsyncClient) | |
| mock_response = MagicMock() | |
| mock_response.json.return_value = {"result": "success"} | |
| mock_response.raise_for_status.return_value = None | |
| client.request.return_value = mock_response | |
| return client | |
| class TestParameterCollisions: | |
| """Test parameter name collisions between path/query/header and body parameters.""" | |
| async def test_path_body_collision_current_broken_behavior(self, mock_client): | |
| """ | |
| Demonstrates the current broken behavior when a parameter exists in both path and body. | |
| This test should FAIL with the current implementation. | |
| """ | |
| # Create route with collision: id in both path and body | |
| route = HTTPRoute( | |
| path="/users/{id}", | |
| method="PUT", | |
| operation_id="update_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| ) | |
| ], | |
| request_body=RequestBodyInfo( | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "id": {"type": "integer", "description": "User ID"}, | |
| "name": {"type": "string", "description": "User name"}, | |
| "email": {"type": "string", "description": "User email"}, | |
| }, | |
| "required": ["id", "name"], | |
| } | |
| } | |
| ), | |
| ) | |
| # Create tool with current implementation | |
| tool = OpenAPITool( | |
| client=mock_client, | |
| route=route, | |
| name="update_user", | |
| description="Update user", | |
| parameters={}, # Schema would be generated by _combine_schemas | |
| ) | |
| # This call should work but currently fails because body 'id' is excluded | |
| arguments = {"id": 123, "name": "John Doe", "email": "john@example.com"} | |
| await tool.run(arguments) | |
| # Check what was actually sent | |
| call_args = mock_client.request.call_args | |
| assert call_args is not None | |
| # Current broken behavior: id goes to path but is excluded from body | |
| # This means the body is missing the required 'id' field | |
| assert call_args[1]["url"] == "/users/123" # Path parameter works | |
| # This assertion will FAIL with current implementation because 'id' is excluded from body | |
| expected_body = {"id": 123, "name": "John Doe", "email": "john@example.com"} | |
| assert call_args[1]["json"] == expected_body, ( | |
| "Body should include 'id' parameter" | |
| ) | |
| async def test_path_body_collision_with_suffixing(self, mock_client): | |
| """ | |
| Test the desired behavior with parameter suffixing. | |
| This test should PASS after implementing the fix. | |
| """ | |
| # Create route with collision: id in both path and body | |
| route = HTTPRoute( | |
| path="/users/{id}", | |
| method="PUT", | |
| operation_id="update_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| ) | |
| ], | |
| request_body=RequestBodyInfo( | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "id": {"type": "integer", "description": "User ID"}, | |
| "name": {"type": "string", "description": "User name"}, | |
| "email": {"type": "string", "description": "User email"}, | |
| }, | |
| "required": ["id", "name"], | |
| } | |
| } | |
| ), | |
| ) | |
| # Tool should be created with suffixed schema | |
| tool = OpenAPITool( | |
| client=mock_client, | |
| route=route, | |
| name="update_user", | |
| description="Update user", | |
| parameters={}, # Schema would include id__path and id | |
| ) | |
| # LLM would call with suffixed parameters | |
| arguments = { | |
| "id__path": 123, # Goes to path parameter | |
| "id": 123, # Goes to body parameter | |
| "name": "John Doe", | |
| "email": "john@example.com", | |
| } | |
| await tool.run(arguments) | |
| # Verify correct request was made | |
| call_args = mock_client.request.call_args | |
| assert call_args is not None | |
| # Path parameter should be populated from id__path | |
| assert call_args[1]["url"] == "/users/123" | |
| # Body should include id (from unsuffixed parameter) | |
| expected_body = {"id": 123, "name": "John Doe", "email": "john@example.com"} | |
| assert call_args[1]["json"] == expected_body | |
| async def test_query_body_collision_with_suffixing(self, mock_client): | |
| """Test parameter collision between query and body parameters.""" | |
| route = HTTPRoute( | |
| path="/search", | |
| method="POST", | |
| operation_id="search_users", | |
| parameters=[ | |
| ParameterInfo( | |
| name="limit", | |
| location="query", | |
| required=False, | |
| schema={"type": "integer", "default": 10}, | |
| ) | |
| ], | |
| request_body=RequestBodyInfo( | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "limit": { | |
| "type": "integer", | |
| "description": "Max results in response", | |
| }, | |
| "query": {"type": "string", "description": "Search query"}, | |
| }, | |
| "required": ["query"], | |
| } | |
| } | |
| ), | |
| ) | |
| tool = OpenAPITool( | |
| client=mock_client, | |
| route=route, | |
| name="search_users", | |
| description="Search users", | |
| parameters={}, | |
| ) | |
| # LLM call with suffixed parameters | |
| arguments = { | |
| "limit__query": 5, # Goes to query parameter | |
| "limit": 100, # Goes to body parameter | |
| "query": "john", | |
| } | |
| await tool.run(arguments) | |
| call_args = mock_client.request.call_args | |
| assert call_args is not None | |
| # Query parameter from limit__query | |
| assert call_args[1]["params"] == {"limit": 5} | |
| # Body includes limit from unsuffixed parameter | |
| expected_body = {"limit": 100, "query": "john"} | |
| assert call_args[1]["json"] == expected_body | |
| async def test_no_collisions_unchanged_behavior(self, mock_client): | |
| """Test that parameters with no collisions keep original names.""" | |
| route = HTTPRoute( | |
| path="/users/{user_id}", | |
| method="POST", | |
| operation_id="create_user", | |
| parameters=[ | |
| ParameterInfo( | |
| name="user_id", | |
| location="path", | |
| required=True, | |
| schema={"type": "integer"}, | |
| ) | |
| ], | |
| request_body=RequestBodyInfo( | |
| content_schema={ | |
| "application/json": { | |
| "type": "object", | |
| "properties": { | |
| "name": {"type": "string"}, | |
| "email": {"type": "string"}, | |
| }, | |
| "required": ["name"], | |
| } | |
| } | |
| ), | |
| ) | |
| tool = OpenAPITool( | |
| client=mock_client, | |
| route=route, | |
| name="create_user", | |
| description="Create user", | |
| parameters={}, | |
| ) | |
| # No collisions, so original parameter names should work | |
| arguments = { | |
| "user_id": 123, # Path parameter (no suffix needed) | |
| "name": "John", # Body parameter | |
| "email": "john@example.com", | |
| } | |
| await tool.run(arguments) | |
| call_args = mock_client.request.call_args | |
| assert call_args is not None | |
| assert call_args[1]["url"] == "/users/123" | |
| expected_body = {"name": "John", "email": "john@example.com"} | |
| assert call_args[1]["json"] == expected_body | |