Jeremiah Lowin commited on
Commit
72df1a6
·
1 Parent(s): ef08a5c

Add tests

Browse files
Files changed (1) hide show
  1. tests/server/openapi/test_openapi.py +284 -0
tests/server/openapi/test_openapi.py CHANGED
@@ -2114,3 +2114,287 @@ class TestRouteMapTags:
2114
  "getMetrics",
2115
  }
2116
  assert tool_names == expected_tools
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2114
  "getMetrics",
2115
  }
2116
  assert tool_names == expected_tools
2117
+
2118
+
2119
+ class TestMCPNames:
2120
+ """Tests for the mcp_names dictionary functionality."""
2121
+
2122
+ @pytest.fixture
2123
+ def mcp_names_openapi_spec(self) -> dict:
2124
+ """OpenAPI spec with various operationIds for testing naming strategies."""
2125
+ return {
2126
+ "openapi": "3.1.0",
2127
+ "info": {"title": "MCP Names Test API", "version": "1.0.0"},
2128
+ "paths": {
2129
+ "/users": {
2130
+ "get": {
2131
+ "operationId": "list_users__with_pagination",
2132
+ "summary": "Get All Users",
2133
+ "responses": {"200": {"description": "Success"}},
2134
+ },
2135
+ "post": {
2136
+ "operationId": "create_user_admin__special_permissions",
2137
+ "summary": "Create New User",
2138
+ "requestBody": {
2139
+ "required": True,
2140
+ "content": {
2141
+ "application/json": {
2142
+ "schema": {
2143
+ "type": "object",
2144
+ "properties": {"name": {"type": "string"}},
2145
+ "required": ["name"],
2146
+ }
2147
+ }
2148
+ },
2149
+ },
2150
+ "responses": {"201": {"description": "Created"}},
2151
+ },
2152
+ },
2153
+ "/users/{id}": {
2154
+ "get": {
2155
+ "operationId": "get_user_by_id__admin_only",
2156
+ "summary": "Fetch Single User Profile",
2157
+ "parameters": [
2158
+ {
2159
+ "name": "id",
2160
+ "in": "path",
2161
+ "required": True,
2162
+ "schema": {"type": "integer"},
2163
+ }
2164
+ ],
2165
+ "responses": {"200": {"description": "Success"}},
2166
+ }
2167
+ },
2168
+ "/very-long-endpoint-name": {
2169
+ "get": {
2170
+ "operationId": "this_is_a_very_long_operation_id_that_exceeds_fifty_six_characters_and_should_be_truncated",
2171
+ "summary": "This Is A Very Long Summary That Should Also Be Truncated When Used As Name",
2172
+ "responses": {"200": {"description": "Success"}},
2173
+ }
2174
+ },
2175
+ "/special": {
2176
+ "get": {
2177
+ "operationId": "special-chars@and#spaces in$operation%id",
2178
+ "summary": "Special Chars & Spaces In Summary!",
2179
+ "responses": {"200": {"description": "Success"}},
2180
+ }
2181
+ },
2182
+ },
2183
+ }
2184
+
2185
+ @pytest.fixture
2186
+ async def mock_client(self) -> httpx.AsyncClient:
2187
+ """Mock client for testing."""
2188
+
2189
+ async def _responder(request):
2190
+ return httpx.Response(200, json={"status": "ok"})
2191
+
2192
+ transport = httpx.MockTransport(_responder)
2193
+ return httpx.AsyncClient(transport=transport, base_url="http://test")
2194
+
2195
+ async def test_mcp_names_custom_mapping(self, mcp_names_openapi_spec, mock_client):
2196
+ """Test that mcp_names dictionary provides custom names for components."""
2197
+ mcp_names = {
2198
+ "list_users__with_pagination": "user_list",
2199
+ "create_user_admin__special_permissions": "admin_create_user",
2200
+ "get_user_by_id__admin_only": "user_detail",
2201
+ }
2202
+
2203
+ server = FastMCPOpenAPI(
2204
+ openapi_spec=mcp_names_openapi_spec,
2205
+ client=mock_client,
2206
+ mcp_names=mcp_names,
2207
+ )
2208
+
2209
+ # Check tools use custom names
2210
+ tools = server._tool_manager.list_tools()
2211
+ tool_names = {tool.name for tool in tools}
2212
+ assert "admin_create_user" in tool_names
2213
+
2214
+ # Check resource templates use custom names
2215
+ templates = list(server._resource_manager.get_templates().values())
2216
+ template_names = {template.name for template in templates}
2217
+ assert "user_detail" in template_names
2218
+
2219
+ # Check resources use custom names
2220
+ resources = list(server._resource_manager.get_resources().values())
2221
+ resource_names = {resource.name for resource in resources}
2222
+ assert "user_list" in resource_names
2223
+
2224
+ async def test_mcp_names_fallback_to_operation_id_short(
2225
+ self, mcp_names_openapi_spec, mock_client
2226
+ ):
2227
+ """Test fallback to operationId up to double underscore when not in mcp_names."""
2228
+ # Only provide mapping for one operationId
2229
+ mcp_names = {
2230
+ "list_users__with_pagination": "custom_user_list",
2231
+ }
2232
+
2233
+ server = FastMCPOpenAPI(
2234
+ openapi_spec=mcp_names_openapi_spec,
2235
+ client=mock_client,
2236
+ mcp_names=mcp_names,
2237
+ )
2238
+
2239
+ tools = server._tool_manager.list_tools()
2240
+ tool_names = {tool.name for tool in tools}
2241
+
2242
+ templates = list(server._resource_manager.get_templates().values())
2243
+ template_names = {template.name for template in templates}
2244
+
2245
+ resources = list(server._resource_manager.get_resources().values())
2246
+ resource_names = {resource.name for resource in resources}
2247
+
2248
+ # Custom mapped name should be used
2249
+ assert "custom_user_list" in resource_names
2250
+
2251
+ # Unmapped operationIds should use short version (up to __)
2252
+ assert "create_user_admin" in tool_names
2253
+ assert "get_user_by_id" in template_names
2254
+
2255
+ async def test_names_are_slugified(self, mcp_names_openapi_spec, mock_client):
2256
+ """Test that names are properly slugified (spaces, special chars removed)."""
2257
+ server = FastMCPOpenAPI(
2258
+ openapi_spec=mcp_names_openapi_spec,
2259
+ client=mock_client,
2260
+ )
2261
+
2262
+ resources = list(server._resource_manager.get_resources().values())
2263
+ resource_names = {
2264
+ resource.name for resource in resources if resource.name is not None
2265
+ }
2266
+
2267
+ # Special chars and spaces should be slugified
2268
+ slugified_name = next(
2269
+ (name for name in resource_names if "special" in name), None
2270
+ )
2271
+ assert slugified_name is not None
2272
+ # Should not contain special characters or spaces
2273
+ assert "@" not in slugified_name
2274
+ assert "#" not in slugified_name
2275
+ assert "$" not in slugified_name
2276
+ assert "%" not in slugified_name
2277
+ assert " " not in slugified_name
2278
+
2279
+ async def test_names_are_truncated_to_56_chars(
2280
+ self, mcp_names_openapi_spec, mock_client
2281
+ ):
2282
+ """Test that names are truncated to 56 characters maximum."""
2283
+ server = FastMCPOpenAPI(
2284
+ openapi_spec=mcp_names_openapi_spec,
2285
+ client=mock_client,
2286
+ )
2287
+
2288
+ # Check all component types
2289
+ all_names = []
2290
+
2291
+ tools = server._tool_manager.list_tools()
2292
+ all_names.extend(tool.name for tool in tools)
2293
+
2294
+ resources = list(server._resource_manager.get_resources().values())
2295
+ all_names.extend(resource.name for resource in resources)
2296
+
2297
+ templates = list(server._resource_manager.get_templates().values())
2298
+ all_names.extend(template.name for template in templates)
2299
+
2300
+ # All names should be 56 characters or less
2301
+ for name in all_names:
2302
+ assert len(name) <= 56, (
2303
+ f"Name '{name}' exceeds 56 characters (length: {len(name)})"
2304
+ )
2305
+
2306
+ # Verify that the long operationId was actually truncated
2307
+ long_name = next((name for name in all_names if len(name) > 50), None)
2308
+ assert long_name is not None, "Expected to find a truncated name for testing"
2309
+
2310
+ async def test_mcp_names_with_from_openapi_classmethod(
2311
+ self, mcp_names_openapi_spec, mock_client
2312
+ ):
2313
+ """Test mcp_names works with FastMCP.from_openapi() classmethod."""
2314
+ mcp_names = {
2315
+ "list_users__with_pagination": "openapi_user_list",
2316
+ }
2317
+
2318
+ server = FastMCP.from_openapi(
2319
+ openapi_spec=mcp_names_openapi_spec,
2320
+ client=mock_client,
2321
+ mcp_names=mcp_names,
2322
+ )
2323
+
2324
+ resources = list(server._resource_manager.get_resources().values())
2325
+ resource_names = {resource.name for resource in resources}
2326
+ assert "openapi_user_list" in resource_names
2327
+
2328
+ async def test_mcp_names_with_from_fastapi_classmethod(self):
2329
+ """Test mcp_names works with FastMCP.from_fastapi() classmethod."""
2330
+ from fastapi import FastAPI
2331
+ from pydantic import BaseModel
2332
+
2333
+ app = FastAPI(title="FastAPI MCP Names Test")
2334
+
2335
+ class User(BaseModel):
2336
+ name: str
2337
+
2338
+ @app.get("/users", operation_id="list_users__with_filters")
2339
+ async def get_users() -> list[User]:
2340
+ return [User(name="test")]
2341
+
2342
+ @app.post("/users", operation_id="create_user__admin_required")
2343
+ async def create_user(user: User) -> User:
2344
+ return user
2345
+
2346
+ mcp_names = {
2347
+ "list_users__with_filters": "fastapi_user_list",
2348
+ "create_user__admin_required": "fastapi_create_user",
2349
+ }
2350
+
2351
+ server = FastMCP.from_fastapi(
2352
+ app=app,
2353
+ mcp_names=mcp_names,
2354
+ )
2355
+
2356
+ tools = server._tool_manager.list_tools()
2357
+ tool_names = {tool.name for tool in tools}
2358
+
2359
+ resources = list(server._resource_manager.get_resources().values())
2360
+ resource_names = {resource.name for resource in resources}
2361
+
2362
+ assert "fastapi_create_user" in tool_names
2363
+ assert "fastapi_user_list" in resource_names
2364
+
2365
+ async def test_mcp_names_custom_names_are_also_truncated(
2366
+ self, mcp_names_openapi_spec, mock_client
2367
+ ):
2368
+ """Test that custom names in mcp_names are also truncated to 56 characters."""
2369
+ # Provide a custom name that's longer than 56 characters
2370
+ very_long_custom_name = "this_is_a_very_long_custom_name_that_exceeds_fifty_six_characters_and_should_be_truncated"
2371
+
2372
+ mcp_names = {
2373
+ "list_users__with_pagination": very_long_custom_name,
2374
+ }
2375
+
2376
+ server = FastMCPOpenAPI(
2377
+ openapi_spec=mcp_names_openapi_spec,
2378
+ client=mock_client,
2379
+ mcp_names=mcp_names,
2380
+ )
2381
+
2382
+ resources = list(server._resource_manager.get_resources().values())
2383
+ resource_names = {
2384
+ resource.name for resource in resources if resource.name is not None
2385
+ }
2386
+
2387
+ # Find the resource that should have the custom name
2388
+ truncated_name = next(
2389
+ (
2390
+ name
2391
+ for name in resource_names
2392
+ if "this_is_a_very_long_custom_name" in name
2393
+ ),
2394
+ None,
2395
+ )
2396
+ assert truncated_name is not None
2397
+ assert len(truncated_name) <= 56
2398
+ assert (
2399
+ len(truncated_name) == 56
2400
+ ) # Should be exactly 56 since original was longer