diff --git a/fastapi/routing.py b/fastapi/routing.py index 5a0e0272e..b5b411d5b 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -677,42 +677,77 @@ class APIRoute(routing.Route): def _detect_route_conflicts( - new_route: "APIRoute", + new_route: "BaseRoute", existing_routes: Sequence[BaseRoute], ) -> List[str]: """ Detect and warn about route conflicts. - Checks if route patterns could shadow each other. Only warns when HTTP methods overlap. + Checks if route patterns could shadow each other. Only compares routes of the same type + and warns when HTTP methods overlap. Different route types (APIRoute vs WebSocketRoute) + are not compared as they handle different protocols. """ import warnings conflicts = [] - if not isinstance(new_route, APIRoute): + # Only check routes that have path_regex attribute (APIRoute, WebSocketRoute, etc.) + if not hasattr(new_route, "path_regex"): return conflicts + # Cache new_route attributes once before loop for performance + new_methods = getattr(new_route, "methods", None) + new_path = getattr(new_route, "path", None) + new_path_regex = getattr(new_route, "path_regex", None) + + # Skip if new route doesn't have path attribute + if new_path is None: + return conflicts + + # Build method string for warning message + methods_str = "" + if new_methods is not None: + methods_str = f" ({', '.join(sorted(new_methods))})" + for existing in existing_routes: - if not isinstance(existing, APIRoute): + # Only compare routes of the exact same type + if type(new_route) is not type(existing): continue - if not (new_route.methods & existing.methods): + # Skip if existing route doesn't have path_regex + if not hasattr(existing, "path_regex"): + continue + + # Check if methods overlap (both routes must have methods attribute) + existing_methods = getattr(existing, "methods", None) + if new_methods is not None and existing_methods is not None: + if not (new_methods & existing_methods): + continue + + # Get existing path attributes safely + existing_path = getattr(existing, "path", None) + + # Skip if existing route doesn't have path attribute + if existing_path is None: continue - if new_route.path == existing.path: + # Skip if paths are identical (route override, not a conflict) + if new_path == existing_path: continue - methods_str = ", ".join(sorted(new_route.methods)) + # Get existing path_regex attribute safely + existing_path_regex = getattr(existing, "path_regex", None) - if new_route.path_regex.match(existing.path): + # Check for shadowing patterns + if new_path_regex and new_path_regex.match(existing_path): conflicts.append( - f"Route {new_route.path} ({methods_str}) " - f"may shadow existing route {existing.path}" + f"Route {new_path}{methods_str} " + f"may shadow existing route {existing_path}" ) - elif existing.path_regex.match(new_route.path): + elif existing_path_regex and existing_path_regex.match(new_path): conflicts.append( - f"Route {new_route.path} ({methods_str}) " - f"may be shadowed by existing route {existing.path}" + f"Route {new_path}{methods_str} " + f"may be shadowed by existing route {existing_path}" ) if conflicts: diff --git a/tests/test_route_conflicts.py b/tests/test_route_conflicts.py index 9b7664508..fcde1a7c8 100644 --- a/tests/test_route_conflicts.py +++ b/tests/test_route_conflicts.py @@ -1,8 +1,9 @@ import warnings from fastapi import FastAPI -from fastapi.routing import APIRouter +from fastapi.routing import APIRouter, _detect_route_conflicts from fastapi.testclient import TestClient +from starlette.routing import Mount def test_route_conflict_warning_dynamic_before_static(): @@ -85,58 +86,6 @@ def test_no_conflict_different_paths(): assert len(w) == 0 -def test_actual_routing_behavior_static_before_dynamic(): - """Verify that static-before-dynamic actually works correctly.""" - app = FastAPI() - - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - - @app.get("/items/sync") - def sync_items(): - return {"action": "sync"} - - @app.get("/items/{item_id}") - def get_item(item_id: str): - return {"item_id": item_id} - - client = TestClient(app) - - # Static route should match - response = client.get("/items/sync") - assert response.json() == {"action": "sync"} - - # Dynamic route should match others - response = client.get("/items/123") - assert response.json() == {"item_id": "123"} - - -def test_actual_routing_behavior_dynamic_before_static(): - """Verify that dynamic-before-static causes shadowing.""" - app = FastAPI() - - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - - @app.get("/items/{item_id}") - def get_item(item_id: str): - return {"item_id": item_id} - - @app.get("/items/sync") - def sync_items(): - return {"action": "sync"} - - client = TestClient(app) - - # Dynamic route will match everything (including "sync") - response = client.get("/items/sync") - assert response.json() == {"item_id": "sync"} - - # Dynamic route should still match others - response = client.get("/items/123") - assert response.json() == {"item_id": "123"} - - def test_router_conflict_detection(): """Test conflict detection works with APIRouter.""" router = APIRouter() @@ -210,10 +159,12 @@ def test_no_duplicate_warnings_same_route(): def sync_items_v1(): return {"version": 1} - # FastAPI will overwrite the route, but should not warn about conflict with itself - # (The path regex won't match its own literal path if they're identical) + # Add the exact same route path again - FastAPI will overwrite it + @app.get("/items/sync") + def sync_items_v2(): + return {"version": 2} - # Should have 0 warnings + # Should have 0 warnings - identical paths are explicitly excluded from conflict detection assert len(w) == 0 @@ -310,3 +261,311 @@ def test_no_conflict_different_param_depth(): # Should NOT warn - completely different paths assert len(w) == 0 + + +def test_no_conflict_with_websocket_routes(): + """Test that WebSocket routes don't trigger conflict warnings.""" + from fastapi import WebSocket + + app = FastAPI() + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # WebSocket route with dynamic path + @app.websocket("/ws/{client_id}") + async def websocket_endpoint(websocket: WebSocket, client_id: str): + await websocket.accept() + await websocket.close() + + # Regular API route that could conflict if both were APIRoute + @app.get("/ws/test") + def get_ws_test(): + return {"test": "ok"} + + # Should NOT warn - WebSocket routes are not APIRoute instances + assert len(w) == 0 + + +def test_no_conflict_with_mount(): + """Test that Mount routes don't trigger conflict warnings.""" + app = FastAPI() + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + # Add a regular route first + @app.get("/static/test") + def get_static_test(): + return {"test": "ok"} + + # Add a Mount - this should not trigger warnings + # Mounting another FastAPI app creates a Mount route + sub_app = FastAPI() + app.mount("/static", sub_app) + + # Should NOT warn - Mount is not an APIRoute + route_conflict_warnings = [ + warning for warning in w if "shadow" in str(warning.message).lower() + ] + assert len(route_conflict_warnings) == 0 + + +def test_detect_route_conflicts_with_non_apiroute(): + """Test _detect_route_conflicts directly with non-APIRoute as new_route.""" + app = FastAPI() + + @app.get("/test") + def test_route(): + return {"test": "ok"} + + # Create a Mount route (not an APIRoute) + mount_route = Mount("/static", app=FastAPI(), name="static") + + # Call _detect_route_conflicts directly with a non-APIRoute + conflicts = _detect_route_conflicts(mount_route, app.routes) # type: ignore + + assert conflicts == [] + + +def test_websocket_conflict_detection(): + """Test that WebSocket routes can also be checked for conflicts.""" + from fastapi import WebSocket + + app = FastAPI() + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + + @app.websocket("/ws/{client_id}") + async def websocket_dynamic(websocket: WebSocket, client_id: str): + await websocket.accept() + await websocket.close() + + @app.websocket("/ws/test") + async def websocket_static(websocket: WebSocket): + await websocket.accept() + await websocket.close() + + # WebSocket routes are not checked yet (would need to call _detect_route_conflicts) + # For now, no warnings since it's not called during app.websocket() + assert len(w) == 0 + + +def test_route_type_isolation(): + """Test that only routes of the same type are compared for conflicts.""" + from fastapi import WebSocket + + app = FastAPI() + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + + # Add routes of different types with overlapping paths + @app.get("/test/{id}") + def get_test(id: str): + return {"id": id} + + @app.websocket("/test/ws") + async def ws_test(websocket: WebSocket): + await websocket.accept() + await websocket.close() + + @app.get("/test/static") + def get_static(): + return {"static": True} + + # Get the routes + api_routes = [r for r in app.routes if type(r).__name__ == "APIRoute"] + ws_routes = [r for r in app.routes if type(r).__name__ == "APIWebSocketRoute"] + + # Test that APIRoute conflicts are detected within same type + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + conflicts = _detect_route_conflicts(api_routes[-1], api_routes[:-1]) # type: ignore + assert len(conflicts) == 1 # /test/static vs /test/{id} + + # Test that WebSocket routes don't interfere with API routes + # Call with WebSocket route against API routes + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + ws_conflicts = _detect_route_conflicts(ws_routes[0], api_routes) # type: ignore + assert len(ws_conflicts) == 0 # Different types, no conflicts + + # Test with API route against WebSocket routes + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + api_conflicts = _detect_route_conflicts(api_routes[0], ws_routes) # type: ignore + assert len(api_conflicts) == 0 # Different types, no conflicts + + +def test_mount_route_without_path_regex(): + """Test that Mount routes (which lack path_regex) are handled safely.""" + from starlette.routing import Mount + + sub_app = FastAPI() + mount = Mount("/static", app=sub_app, name="static") + + app = FastAPI() + + @app.get("/test") + def test_route(): + return {"test": "ok"} + + conflicts = _detect_route_conflicts(mount, app.routes) # type: ignore + assert conflicts == [] + + +def test_api_route_against_mount_routes(): + """Test that API routes skip Mount routes during conflict detection.""" + from starlette.routing import Mount + + app = FastAPI() + + @app.get("/api/{resource}") + def get_resource(resource: str): + return {"resource": resource} + + # Add a Mount route + sub_app = FastAPI() + mount = Mount("/api/static", app=sub_app, name="static") + + # Get the API route + api_routes = [r for r in app.routes if type(r).__name__ == "APIRoute"] + new_route = api_routes[0] + + conflicts = _detect_route_conflicts(new_route, [mount]) # type: ignore + assert conflicts == [] + + +def test_websocket_route_without_methods(): + """Test that WebSocket routes (which have no methods attribute) can be checked for conflicts.""" + from fastapi import WebSocket + + app = FastAPI() + + @app.websocket("/ws/{client_id}") + async def websocket_endpoint(websocket: WebSocket, client_id: str): + await websocket.accept() + await websocket.close() + + @app.websocket("/ws/special") + async def websocket_special(websocket: WebSocket): + await websocket.accept() + await websocket.close() + + # Get WebSocket routes + ws_routes = [r for r in app.routes if type(r).__name__ == "APIWebSocketRoute"] + + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + conflicts = _detect_route_conflicts(ws_routes[-1], ws_routes[:-1]) # type: ignore + + # Should detect conflict even without methods attribute + assert len(conflicts) == 1 + assert "/ws/special" in conflicts[0] + assert "/ws/{client_id}" in conflicts[0] + + +def test_route_without_path_regex_attribute(): + """Test defensive code for routes that truly lack path_regex attribute.""" + from starlette.routing import BaseRoute + + class CustomRouteWithoutPathRegex(BaseRoute): + """Custom route class without path_regex attribute.""" + + def __init__(self): + self.path = "/custom" + + custom_route = CustomRouteWithoutPathRegex() + + app = FastAPI() + + @app.get("/test") + def test_route(): + return {"test": "ok"} + + conflicts = _detect_route_conflicts(custom_route, app.routes) # type: ignore + assert conflicts == [] + + +def test_route_with_path_regex_but_no_path(): + """Test defensive code for routes with path_regex but path is None.""" + import re + + from starlette.routing import BaseRoute + + class CustomRouteWithoutPath(BaseRoute): + """Custom route class with path_regex but no path.""" + + def __init__(self): + self.path_regex = re.compile(r"/custom") + self.methods = {"GET"} + self.path = None + + custom_route = CustomRouteWithoutPath() + + app = FastAPI() + + @app.get("/test") + def test_route(): + return {"test": "ok"} + + conflicts = _detect_route_conflicts(custom_route, app.routes) # type: ignore + assert conflicts == [] + + +def test_existing_route_without_path_regex_in_loop(): + """Test that existing routes without path_regex are skipped in the loop.""" + from starlette.routing import BaseRoute + + class CustomExistingRouteWithoutPathRegex(BaseRoute): + """Custom route without path_regex.""" + + def __init__(self): + self.path = "/existing" + + app = FastAPI() + + @app.get("/test/{id}") + def test_route(id: str): + return {"id": id} + + api_routes = [r for r in app.routes if type(r).__name__ == "APIRoute"] + new_route = api_routes[0] + + custom_existing = CustomExistingRouteWithoutPathRegex() + custom_existing.__class__ = type(new_route) + + conflicts = _detect_route_conflicts(new_route, [custom_existing]) # type: ignore + assert conflicts == [] + + +def test_existing_route_with_none_path_in_loop(): + """Test that existing routes with None path are skipped in the loop.""" + import re + + from starlette.routing import BaseRoute + + class CustomExistingRouteWithNonePath(BaseRoute): + """Custom route with path_regex but path is None.""" + + def __init__(self): + self.path_regex = re.compile(r"/existing") + self.path = None + self.methods = {"GET"} + + app = FastAPI() + + @app.get("/test/{id}") + def test_route(id: str): + return {"id": id} + + api_routes = [r for r in app.routes if type(r).__name__ == "APIRoute"] + new_route = api_routes[0] + + custom_existing = CustomExistingRouteWithNonePath() + custom_existing.__class__ = type(new_route) + + conflicts = _detect_route_conflicts(new_route, [custom_existing]) # type: ignore + assert conflicts == []