Spaces:
Running
Running
File size: 9,965 Bytes
8c5bdf2 | 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 | """Tests for parameter collision handling in openapi_new."""
import httpx
import pytest
from fastmcp.client import Client
from fastmcp.experimental.server.openapi import FastMCPOpenAPI
class TestParameterCollisions:
"""Test parameter name collisions between different locations (path, query, body)."""
@pytest.fixture
def collision_spec(self):
"""OpenAPI spec with parameter name collisions."""
return {
"openapi": "3.0.0",
"info": {"title": "Collision Test API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com"}],
"paths": {
"/users/{id}": {
"put": {
"operationId": "update_user",
"summary": "Update user with collision between path and body",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "integer"},
"description": "User ID in path",
}
],
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"description": "User ID in body (different from path)",
},
"name": {
"type": "string",
"description": "User name",
},
"email": {
"type": "string",
"description": "User email",
},
},
"required": ["name", "email"],
}
}
},
},
"responses": {
"200": {
"description": "User updated",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"email": {"type": "string"},
},
}
}
},
}
},
}
},
"/search": {
"get": {
"operationId": "search_with_collision",
"summary": "Search with query and header collision",
"parameters": [
{
"name": "query",
"in": "query",
"required": True,
"schema": {"type": "string"},
"description": "Search query parameter",
},
{
"name": "query",
"in": "header",
"required": False,
"schema": {"type": "string"},
"description": "Search query in header",
},
],
"responses": {
"200": {
"description": "Search results",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"results": {
"type": "array",
"items": {"type": "object"},
}
},
}
}
},
}
},
}
},
},
}
@pytest.mark.asyncio
async def test_path_body_collision_handling(self, collision_spec):
"""Test that path and body parameters with same name are handled correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
openapi_spec=collision_spec, client=client, name="Collision Test Server"
)
async with Client(server) as mcp_client:
tools = await mcp_client.list_tools()
# Find the update user tool
update_tool = next(tool for tool in tools if tool.name == "update_user")
assert update_tool is not None
# Check that both path and body 'id' parameters are included
params = update_tool.inputSchema
properties = params["properties"]
# Should have both path ID and body ID (with potential suffixing)
# The implementation should handle this collision by suffixing one of them
assert "id" in properties # One version of id
# Check for suffixed versions or verify both exist somehow
# The exact handling depends on implementation, but both should be accessible
param_names = list(properties.keys())
id_params = [name for name in param_names if "id" in name]
assert len(id_params) >= 1 # At least one id parameter
# Should also have other body parameters
assert "name" in properties
assert "email" in properties
# Required fields should include path parameter and required body fields
required = params.get("required", [])
assert "name" in required
assert "email" in required
# Path parameter should be required (may be suffixed)
id_required = any("id" in req for req in required)
assert id_required
@pytest.mark.asyncio
async def test_query_header_collision_handling(self, collision_spec):
"""Test that query and header parameters with same name are handled correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
openapi_spec=collision_spec, client=client, name="Collision Test Server"
)
async with Client(server) as mcp_client:
tools = await mcp_client.list_tools()
# Find the search tool
search_tool = next(
tool for tool in tools if tool.name == "search_with_collision"
)
assert search_tool is not None
# Check that both query and header 'query' parameters are handled
params = search_tool.inputSchema
properties = params["properties"]
# Should handle the collision somehow (suffixing or other mechanism)
param_names = list(properties.keys())
query_params = [name for name in param_names if "query" in name]
assert len(query_params) >= 1 # At least one query parameter
# Required should include the required query parameter
required = params.get("required", [])
query_required = any("query" in req for req in required)
assert query_required
@pytest.mark.asyncio
async def test_collision_resolution_maintains_functionality(self, collision_spec):
"""Test that collision resolution doesn't break basic tool functionality."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
openapi_spec=collision_spec, client=client, name="Collision Test Server"
)
async with Client(server) as mcp_client:
tools = await mcp_client.list_tools()
# Should successfully create tools despite collisions
assert len(tools) == 2
tool_names = {tool.name for tool in tools}
assert "update_user" in tool_names
assert "search_with_collision" in tool_names
# Tools should have valid schemas
for tool in tools:
assert tool.inputSchema is not None
assert tool.inputSchema["type"] == "object"
assert "properties" in tool.inputSchema
|