wehub-resource-sync
375 lines
10 KiB
Python
375 lines
10 KiB
Python
"""Unit tests for extract_route_patterns function."""
|
|
import pytest
|
|
from fastapi import APIRouter, FastAPI
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Mount, Route
|
|
|
|
from ray.serve._private.thirdparty.get_asgi_route_name import (
|
|
extract_route_patterns,
|
|
)
|
|
|
|
|
|
def has_path(patterns, path):
|
|
"""Helper to check if a path exists in patterns list."""
|
|
return any(pattern.path == path for pattern in patterns)
|
|
|
|
|
|
def get_methods_for_path(patterns, path):
|
|
"""Helper to get methods for a specific path."""
|
|
for pattern in patterns:
|
|
if pattern.path == path:
|
|
return pattern.methods
|
|
return None
|
|
|
|
|
|
def test_extract_route_patterns_fastapi_simple():
|
|
"""Test extracting route patterns from a simple FastAPI app."""
|
|
app = FastAPI()
|
|
|
|
@app.get("/")
|
|
def root():
|
|
return {"message": "root"}
|
|
|
|
@app.get("/users/{user_id}")
|
|
def get_user(user_id: str):
|
|
return {"user_id": user_id}
|
|
|
|
@app.post("/items/{item_id}")
|
|
def create_item(item_id: str):
|
|
return {"item_id": item_id}
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
# FastAPI automatically adds some default routes
|
|
assert has_path(patterns, "/")
|
|
assert has_path(patterns, "/users/{user_id}")
|
|
assert has_path(patterns, "/items/{item_id}")
|
|
# FastAPI adds OpenAPI routes
|
|
assert has_path(patterns, "/openapi.json")
|
|
assert has_path(patterns, "/docs")
|
|
|
|
|
|
def test_extract_route_patterns_nested_paths():
|
|
"""Test extracting nested parameterized routes."""
|
|
app = FastAPI()
|
|
|
|
@app.get("/api/v1/users/{user_id}/posts/{post_id}")
|
|
def get_post(user_id: str, post_id: str):
|
|
return {"user_id": user_id, "post_id": post_id}
|
|
|
|
@app.get("/api/v1/users/{user_id}/settings")
|
|
def get_settings(user_id: str):
|
|
return {"user_id": user_id}
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
assert has_path(patterns, "/api/v1/users/{user_id}/posts/{post_id}")
|
|
assert has_path(patterns, "/api/v1/users/{user_id}/settings")
|
|
|
|
|
|
def test_extract_route_patterns_with_mounts():
|
|
"""Test extracting route patterns from apps with mounted sub-apps."""
|
|
# Create a sub-app
|
|
sub_app = Starlette(
|
|
routes=[
|
|
Route("/health", lambda request: None),
|
|
Route("/status", lambda request: None),
|
|
]
|
|
)
|
|
|
|
# Create main app with mounted sub-app
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/", lambda request: None),
|
|
Mount("/admin", app=sub_app),
|
|
]
|
|
)
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
assert has_path(patterns, "/")
|
|
assert has_path(patterns, "/admin/health")
|
|
assert has_path(patterns, "/admin/status")
|
|
|
|
|
|
def test_extract_route_patterns_included_router():
|
|
"""Routes registered via `include_router` must be extracted.
|
|
|
|
On FastAPI >= 0.137 these routes are nested under an `_IncludedRouter` node
|
|
instead of being flattened into `app.routes` (see #64475).
|
|
"""
|
|
app = FastAPI()
|
|
|
|
@app.get("/direct")
|
|
def direct():
|
|
return {}
|
|
|
|
# Router with its own prefix, included without an include-time prefix.
|
|
router = APIRouter(prefix="/prefix")
|
|
|
|
@router.get("/routed/{user_id}")
|
|
def routed():
|
|
return {}
|
|
|
|
@router.websocket("/ws")
|
|
async def ws():
|
|
pass
|
|
|
|
# Router included with an include-time prefix.
|
|
other = APIRouter()
|
|
|
|
@other.post("/create")
|
|
def create():
|
|
return {}
|
|
|
|
app.include_router(router)
|
|
app.include_router(other, prefix="/other")
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
assert has_path(patterns, "/direct")
|
|
assert has_path(patterns, "/prefix/routed/{user_id}")
|
|
assert get_methods_for_path(patterns, "/prefix/routed/{user_id}") == ["GET"]
|
|
assert has_path(patterns, "/other/create")
|
|
assert get_methods_for_path(patterns, "/other/create") == ["POST"]
|
|
# WebSocket routes have no method restrictions.
|
|
assert has_path(patterns, "/prefix/ws")
|
|
assert get_methods_for_path(patterns, "/prefix/ws") is None
|
|
|
|
|
|
def test_extract_route_patterns_nested_mounts():
|
|
"""Test extracting patterns from deeply nested mounts."""
|
|
# Innermost app
|
|
inner_app = Starlette(
|
|
routes=[
|
|
Route("/details", lambda request: None),
|
|
]
|
|
)
|
|
|
|
# Middle app
|
|
middle_app = Starlette(
|
|
routes=[
|
|
Route("/list", lambda request: None),
|
|
Mount("/item", app=inner_app),
|
|
]
|
|
)
|
|
|
|
# Main app
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/", lambda request: None),
|
|
Mount("/api/v1", app=middle_app),
|
|
]
|
|
)
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
assert has_path(patterns, "/")
|
|
assert has_path(patterns, "/api/v1/list")
|
|
assert has_path(patterns, "/api/v1/item/details")
|
|
|
|
|
|
def test_extract_route_patterns_with_root_path():
|
|
"""Test extracting patterns from apps with root_path set."""
|
|
app = FastAPI(root_path="/v1")
|
|
|
|
@app.get("/")
|
|
def root():
|
|
return {}
|
|
|
|
@app.get("/users")
|
|
def get_users():
|
|
return []
|
|
|
|
@app.get("/items/{item_id}")
|
|
def get_item(item_id: str):
|
|
return {"item_id": item_id}
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
# Root path should be prepended to all routes
|
|
assert has_path(patterns, "/v1/") # Root route
|
|
assert has_path(patterns, "/v1/users")
|
|
assert has_path(patterns, "/v1/items/{item_id}")
|
|
|
|
|
|
def test_extract_route_patterns_empty_app():
|
|
"""Test extracting patterns from an app with no user-defined routes."""
|
|
app = FastAPI()
|
|
# Don't define any routes
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
# Should still have FastAPI defaults
|
|
assert has_path(patterns, "/openapi.json")
|
|
assert has_path(patterns, "/docs")
|
|
# May or may not have "/" depending on FastAPI version
|
|
|
|
|
|
def test_extract_route_patterns_starlette():
|
|
"""Test extracting patterns from a pure Starlette app."""
|
|
|
|
async def homepage(request):
|
|
return None
|
|
|
|
async def user_detail(request):
|
|
return None
|
|
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/", homepage),
|
|
Route("/users/{user_id}", user_detail),
|
|
]
|
|
)
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
assert has_path(patterns, "/")
|
|
assert has_path(patterns, "/users/{user_id}")
|
|
# Starlette shouldn't have OpenAPI routes
|
|
assert not has_path(patterns, "/openapi.json")
|
|
|
|
|
|
def test_extract_route_patterns_multiple_methods_same_path():
|
|
"""Test that methods are grouped when multiple methods use same path."""
|
|
app = FastAPI()
|
|
|
|
@app.get("/items/{item_id}")
|
|
def get_item(item_id: str):
|
|
return {"item_id": item_id}
|
|
|
|
@app.put("/items/{item_id}")
|
|
def update_item(item_id: str):
|
|
return {"item_id": item_id}
|
|
|
|
@app.delete("/items/{item_id}")
|
|
def delete_item(item_id: str):
|
|
return {"item_id": item_id}
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
# Path should appear only once with all methods grouped
|
|
path_count = sum(1 for pattern in patterns if pattern.path == "/items/{item_id}")
|
|
assert path_count == 1
|
|
|
|
# Check that all methods are present
|
|
methods = get_methods_for_path(patterns, "/items/{item_id}")
|
|
assert methods is not None
|
|
assert "GET" in methods
|
|
assert "PUT" in methods
|
|
assert "DELETE" in methods
|
|
|
|
|
|
def test_extract_route_patterns_invalid_app():
|
|
"""Test that invalid apps return empty list gracefully."""
|
|
|
|
class FakeApp:
|
|
"""An app without routes attribute."""
|
|
|
|
pass
|
|
|
|
fake_app = FakeApp()
|
|
|
|
# Should return empty list without raising exception
|
|
patterns = extract_route_patterns(fake_app)
|
|
assert patterns == []
|
|
|
|
|
|
def test_extract_route_patterns_mount_without_routes():
|
|
"""Test handling mounts that don't have sub-routes."""
|
|
from starlette.responses import PlainTextResponse
|
|
|
|
async def custom_mount(scope, receive, send):
|
|
response = PlainTextResponse("Custom mount")
|
|
await response(scope, receive, send)
|
|
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/", lambda request: None),
|
|
Mount("/custom", app=custom_mount),
|
|
]
|
|
)
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
assert has_path(patterns, "/")
|
|
assert has_path(patterns, "/custom")
|
|
# Custom mount has no method restrictions
|
|
assert get_methods_for_path(patterns, "/custom") is None
|
|
|
|
|
|
def test_extract_route_patterns_sorted_output():
|
|
"""Test that output is sorted by path."""
|
|
app = FastAPI()
|
|
|
|
@app.get("/zebra")
|
|
def zebra():
|
|
return {}
|
|
|
|
@app.get("/apple")
|
|
def apple():
|
|
return {}
|
|
|
|
@app.get("/banana")
|
|
def banana():
|
|
return {}
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
# Extract just the paths
|
|
paths = [pattern.path for pattern in patterns]
|
|
|
|
# Find the user-defined routes
|
|
user_routes = [p for p in paths if p in ["/zebra", "/apple", "/banana"]]
|
|
|
|
# Should be sorted
|
|
assert user_routes == ["/apple", "/banana", "/zebra"]
|
|
|
|
|
|
def test_extract_route_patterns_special_characters():
|
|
"""Test routes with special regex characters."""
|
|
app = FastAPI()
|
|
|
|
@app.get("/users/{user_id:path}")
|
|
def get_user_path(user_id: str):
|
|
return {"user_id": user_id}
|
|
|
|
@app.get("/items/{item_id:int}")
|
|
def get_item_int(item_id: int):
|
|
return {"item_id": item_id}
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
# Extract just the paths
|
|
paths = [pattern.path for pattern in patterns]
|
|
|
|
# FastAPI converts these to standard patterns
|
|
assert any("user_id" in p for p in paths)
|
|
assert any("item_id" in p for p in paths)
|
|
|
|
|
|
def test_extract_route_patterns_websocket_routes():
|
|
"""Test that WebSocket routes are also extracted."""
|
|
app = FastAPI()
|
|
|
|
@app.get("/http")
|
|
def http_route():
|
|
return {}
|
|
|
|
@app.websocket("/ws")
|
|
async def websocket_route(websocket):
|
|
await websocket.accept()
|
|
await websocket.close()
|
|
|
|
patterns = extract_route_patterns(app)
|
|
|
|
assert has_path(patterns, "/http")
|
|
assert has_path(patterns, "/ws")
|
|
|
|
# WebSocket route should have no method restrictions
|
|
assert get_methods_for_path(patterns, "/ws") is None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|