Spaces:
Running
Running
File size: 9,123 Bytes
7004e72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 | """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
@pytest.fixture
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
|