Sebastián Ramírez
2 weeks ago
committed by
GitHub
GitHub
11 changed files with 1877 additions and 316 deletions
@ -1,24 +1,14 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.routing import APIRoute |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"item_id": "Foo"}] |
|||
def custom_generate_unique_id(route: APIRoute) -> str: |
|||
return route.name |
|||
|
|||
|
|||
def use_route_names_as_operation_ids(app: FastAPI) -> None: |
|||
""" |
|||
Simplify operation IDs so that generated API clients have simpler function |
|||
names. |
|||
app = FastAPI(generate_unique_id_function=custom_generate_unique_id) |
|||
|
|||
Should be called only after all routes have been added. |
|||
""" |
|||
for route in app.routes: |
|||
if isinstance(route, APIRoute): |
|||
route.operation_id = route.name # in this case, 'read_items' |
|||
|
|||
|
|||
use_route_names_as_operation_ids(app) |
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"item_id": "Foo"}] |
|||
|
|||
File diff suppressed because it is too large
@ -0,0 +1,855 @@ |
|||
from typing import Annotated, cast |
|||
|
|||
import pytest |
|||
from fastapi import APIRouter, Body, Depends, FastAPI, Request |
|||
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse |
|||
from fastapi.routing import ( |
|||
APIRoute, |
|||
_IncludedRouter, |
|||
_iter_included_route_candidates, |
|||
_restore_fastapi_scope_key, |
|||
) |
|||
from fastapi.testclient import TestClient |
|||
from starlette.routing import BaseRoute, Host, Match, Mount, NoMatchFound, Route, Router |
|||
|
|||
|
|||
def dependency_a(): |
|||
return "a" |
|||
|
|||
|
|||
def dependency_b(): |
|||
return "b" |
|||
|
|||
|
|||
def dependency_c(): |
|||
return "c" |
|||
|
|||
|
|||
def unique_id_b(route: APIRoute) -> str: |
|||
return f"b_{route.name}" |
|||
|
|||
|
|||
def test_router_include_context_matches_flattened_include_metadata(): |
|||
callback_router = APIRouter() |
|||
|
|||
@callback_router.post("/callback") |
|||
def callback(): # pragma: no cover |
|||
return {"ok": True} |
|||
|
|||
callback_route = callback_router.routes[0] |
|||
|
|||
parent_router = APIRouter() |
|||
included_router = APIRouter( |
|||
prefix="/items", |
|||
tags=["router"], |
|||
dependencies=[Depends(dependency_a)], |
|||
responses={401: {"description": "Unauthorized"}}, |
|||
callbacks=[callback_route], |
|||
default_response_class=HTMLResponse, |
|||
strict_content_type=False, |
|||
) |
|||
|
|||
@included_router.get( |
|||
"/{item_id}", |
|||
tags=["route"], |
|||
dependencies=[Depends(dependency_b)], |
|||
responses={404: {"description": "Missing"}}, |
|||
callbacks=[callback_route], |
|||
generate_unique_id_function=unique_id_b, |
|||
) |
|||
def read_item(item_id: str, request: Request): |
|||
context = request.scope["fastapi"]["effective_route_context"] |
|||
return JSONResponse( |
|||
{ |
|||
"path": context.path, |
|||
"tags": context.tags, |
|||
"dependency_count": len(context.dependencies), |
|||
"response_codes": sorted(context.responses), |
|||
"callback_count": len(context.callbacks or []), |
|||
"deprecated": context.deprecated, |
|||
"include_in_schema": context.include_in_schema, |
|||
"response_class": context.response_class.__name__, |
|||
"generate_unique_id": context.generate_unique_id_function(context), |
|||
"strict_content_type": context.strict_content_type, |
|||
"has_dependency_overrides_provider": ( |
|||
context.dependency_overrides_provider |
|||
is app.router.dependency_overrides_provider |
|||
), |
|||
} |
|||
) |
|||
|
|||
parent_router.include_router( |
|||
included_router, |
|||
prefix="/api", |
|||
tags=["include"], |
|||
dependencies=[Depends(dependency_c)], |
|||
responses={400: {"description": "Bad request"}}, |
|||
callbacks=[callback_route], |
|||
deprecated=True, |
|||
include_in_schema=False, |
|||
) |
|||
|
|||
app = FastAPI() |
|||
app.include_router(parent_router) |
|||
response = TestClient(app).get("/api/items/foo") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"path": "/api/items/{item_id}", |
|||
"tags": ["include", "router", "route"], |
|||
"dependency_count": 3, |
|||
"response_codes": [400, 401, 404], |
|||
"callback_count": 3, |
|||
"deprecated": True, |
|||
"include_in_schema": False, |
|||
"response_class": "HTMLResponse", |
|||
"generate_unique_id": "b_read_item", |
|||
"strict_content_type": False, |
|||
"has_dependency_overrides_provider": True, |
|||
} |
|||
|
|||
|
|||
def test_live_route_addition_uses_include_metadata_for_runtime_and_openapi(): |
|||
calls: list[str] = [] |
|||
|
|||
def included_dependency(): |
|||
calls.append("dependency") |
|||
|
|||
router = APIRouter() |
|||
app = FastAPI() |
|||
app.include_router( |
|||
router, |
|||
prefix="/api", |
|||
tags=["included"], |
|||
dependencies=[Depends(included_dependency)], |
|||
responses={418: {"description": "Teapot"}}, |
|||
) |
|||
|
|||
@router.get("/later") |
|||
def read_later(): |
|||
return {"later": True} |
|||
|
|||
client = TestClient(app) |
|||
response = client.get("/api/later") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == {"later": True} |
|||
assert calls == ["dependency"] |
|||
operation = client.get("/openapi.json").json()["paths"]["/api/later"]["get"] |
|||
assert operation["tags"] == ["included"] |
|||
assert operation["responses"]["418"] == {"description": "Teapot"} |
|||
|
|||
|
|||
def test_openapi_cache_updates_after_live_route_addition(): |
|||
router = APIRouter() |
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
client = TestClient(app) |
|||
|
|||
first_schema = client.get("/openapi.json").json() |
|||
assert "/api/later" not in first_schema["paths"] |
|||
|
|||
@router.get("/later") |
|||
def read_later(): # pragma: no cover |
|||
return {"later": True} |
|||
|
|||
second_schema = client.get("/openapi.json").json() |
|||
assert "/api/later" in second_schema["paths"] |
|||
|
|||
|
|||
def test_nested_router_added_after_parent_inclusion_is_live(): |
|||
parent_router = APIRouter() |
|||
child_router = APIRouter() |
|||
app = FastAPI() |
|||
app.include_router(parent_router, prefix="/api") |
|||
parent_router.include_router(child_router, prefix="/child", tags=["child"]) |
|||
|
|||
@child_router.get("/items") |
|||
def read_items(): |
|||
return ["item"] |
|||
|
|||
client = TestClient(app) |
|||
response = client.get("/api/child/items") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == ["item"] |
|||
operation = client.get("/openapi.json").json()["paths"]["/api/child/items"]["get"] |
|||
assert operation["tags"] == ["child"] |
|||
|
|||
|
|||
def test_repeated_deep_inclusions_handle_all_concrete_paths(): |
|||
shared_router = APIRouter() |
|||
|
|||
@shared_router.get("/items") |
|||
def read_items(): |
|||
return [] |
|||
|
|||
parent_router = APIRouter() |
|||
parent_router.include_router(shared_router, prefix="/a") |
|||
parent_router.include_router(shared_router, prefix="/b") |
|||
|
|||
app = FastAPI() |
|||
app.include_router(parent_router, prefix="/v1") |
|||
app.include_router(parent_router, prefix="/v2") |
|||
|
|||
client = TestClient(app) |
|||
paths = ["/v1/a/items", "/v1/b/items", "/v2/a/items", "/v2/b/items"] |
|||
for path in paths: |
|||
response = client.get(path) |
|||
assert response.status_code == 200 |
|||
assert response.json() == [] |
|||
assert set(client.get("/openapi.json").json()["paths"]) == set(paths) |
|||
|
|||
|
|||
def test_url_path_for_uses_effective_context_for_live_included_route(): |
|||
router = APIRouter() |
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
|
|||
@router.get("/items/{item_id}", name="read_item") |
|||
def read_item(item_id: str): # pragma: no cover |
|||
return {"item_id": item_id} |
|||
|
|||
assert app.url_path_for("read_item", item_id="abc") == "/api/items/abc" |
|||
|
|||
|
|||
def test_url_path_for_uses_distinct_repeated_inclusion_contexts(): |
|||
router = APIRouter() |
|||
|
|||
@router.get("/items/{item_id}", name="read_item") |
|||
def read_item(item_id: str): # pragma: no cover |
|||
return {"item_id": item_id} |
|||
|
|||
parent_router = APIRouter() |
|||
parent_router.include_router(router, prefix="/v1") |
|||
parent_router.include_router(router, prefix="/v2") |
|||
|
|||
assert parent_router.url_path_for("read_item", item_id="abc") == "/v1/items/abc" |
|||
assert ( |
|||
parent_router.routes[1].url_path_for("read_item", item_id="abc") |
|||
== "/v2/items/abc" |
|||
) |
|||
|
|||
|
|||
def test_indirect_router_inclusion_cycles_are_rejected(): |
|||
parent_router = APIRouter() |
|||
child_router = APIRouter() |
|||
|
|||
parent_router.include_router(child_router, prefix="/child") |
|||
|
|||
with pytest.raises(AssertionError, match="already includes this router"): |
|||
child_router.include_router(parent_router, prefix="/parent") |
|||
|
|||
parent_router = APIRouter() |
|||
child_router = APIRouter() |
|||
grandchild_router = APIRouter() |
|||
|
|||
parent_router.include_router(child_router, prefix="/child") |
|||
child_router.include_router(grandchild_router, prefix="/grandchild") |
|||
|
|||
with pytest.raises(AssertionError, match="already includes this router"): |
|||
grandchild_router.include_router(parent_router, prefix="/parent") |
|||
|
|||
|
|||
def test_original_api_route_subclass_instance_is_called_after_inclusion(): |
|||
class TrackingRoute(APIRoute): |
|||
calls = 0 |
|||
|
|||
async def handle(self, scope, receive, send): |
|||
self.calls += 1 |
|||
await super().handle(scope, receive, send) |
|||
|
|||
router = APIRouter(route_class=TrackingRoute) |
|||
|
|||
@router.get("/items") |
|||
def read_items(): |
|||
return [] |
|||
|
|||
original_route = router.routes[0] |
|||
assert isinstance(original_route, TrackingRoute) |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
|
|||
response = TestClient(app).get("/api/items") |
|||
|
|||
assert response.status_code == 200 |
|||
assert original_route.calls == 1 |
|||
|
|||
|
|||
def test_original_api_route_get_route_handler_is_called_after_inclusion(): |
|||
class TrackingRoute(APIRoute): |
|||
calls = 0 |
|||
|
|||
def get_route_handler(self): |
|||
handler = super().get_route_handler() |
|||
|
|||
async def custom_handler(request): |
|||
self.calls += 1 |
|||
return await handler(request) |
|||
|
|||
return custom_handler |
|||
|
|||
router = APIRouter(route_class=TrackingRoute) |
|||
|
|||
@router.get("/items") |
|||
def read_items(): |
|||
return [] |
|||
|
|||
original_route = router.routes[0] |
|||
assert isinstance(original_route, TrackingRoute) |
|||
original_route.calls = 0 |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
|
|||
response = TestClient(app).get("/api/items") |
|||
|
|||
assert response.status_code == 200 |
|||
assert original_route.calls == 1 |
|||
|
|||
|
|||
def test_original_api_route_matches_is_called_after_inclusion(): |
|||
class HeaderRoute(APIRoute): |
|||
calls = 0 |
|||
|
|||
def matches(self, scope): |
|||
self.calls += 1 |
|||
headers = dict(scope.get("headers", [])) |
|||
if headers.get(b"x-match") != b"yes": |
|||
return Match.NONE, {} |
|||
return super().matches(scope) |
|||
|
|||
router = APIRouter(route_class=HeaderRoute) |
|||
|
|||
@router.get("/items") |
|||
def read_items(): |
|||
return [] |
|||
|
|||
original_route = router.routes[0] |
|||
assert isinstance(original_route, HeaderRoute) |
|||
original_route.calls = 0 |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
client = TestClient(app) |
|||
|
|||
assert client.get("/api/items").status_code == 404 |
|||
assert client.get("/api/items", headers={"x-match": "yes"}).status_code == 200 |
|||
assert original_route.calls >= 2 |
|||
|
|||
|
|||
def test_effective_route_context_is_available_in_scope_during_request(): |
|||
router = APIRouter() |
|||
|
|||
@router.get("/items") |
|||
def read_items(request: Request): |
|||
fastapi_scope = request.scope.get("fastapi") |
|||
assert isinstance(fastapi_scope, dict) |
|||
return { |
|||
"has_context": "effective_route_context" in fastapi_scope, |
|||
"path": fastapi_scope["effective_route_context"].path, |
|||
} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
|
|||
response = TestClient(app).get("/api/items") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == {"has_context": True, "path": "/api/items"} |
|||
|
|||
|
|||
def test_original_api_router_matches_is_called_after_inclusion(): |
|||
class HeaderRouter(APIRouter): |
|||
calls = 0 |
|||
|
|||
def matches(self, scope): |
|||
self.calls += 1 |
|||
headers = dict(scope.get("headers", [])) |
|||
if headers.get(b"x-router-match") != b"yes": |
|||
return Match.NONE, {} |
|||
return super().matches(scope) |
|||
|
|||
router = HeaderRouter() |
|||
|
|||
@router.get("/items") |
|||
def read_items(): |
|||
return [] |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
client = TestClient(app) |
|||
|
|||
assert client.get("/api/items").status_code == 404 |
|||
assert ( |
|||
client.get("/api/items", headers={"x-router-match": "yes"}).status_code == 200 |
|||
) |
|||
assert router.calls >= 2 |
|||
|
|||
|
|||
def test_original_nested_api_router_subclasses_are_called_after_inclusion(): |
|||
class TrackingRouter(APIRouter): |
|||
calls = 0 |
|||
|
|||
async def handle(self, scope, receive, send): |
|||
self.calls += 1 |
|||
await super().handle(scope, receive, send) |
|||
|
|||
parent_router = TrackingRouter() |
|||
child_router = TrackingRouter() |
|||
|
|||
@child_router.get("/items") |
|||
def read_items(): |
|||
return [] |
|||
|
|||
parent_router.include_router(child_router, prefix="/child") |
|||
app = FastAPI() |
|||
app.include_router(parent_router, prefix="/api") |
|||
|
|||
response = TestClient(app).get("/api/child/items") |
|||
|
|||
assert response.status_code == 200 |
|||
assert parent_router.calls == 1 |
|||
assert child_router.calls == 1 |
|||
|
|||
|
|||
def test_router_and_include_prefix_path_params_reach_endpoint_and_openapi(): |
|||
router = APIRouter(prefix="/tenants/{tenant_id}") |
|||
|
|||
@router.get("/items/{item_id}") |
|||
def read_item(version: int, tenant_id: int, item_id: int): |
|||
return {"version": version, "tenant_id": tenant_id, "item_id": item_id} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api/{version}") |
|||
|
|||
client = TestClient(app) |
|||
response = client.get("/api/1/tenants/2/items/3") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == {"version": 1, "tenant_id": 2, "item_id": 3} |
|||
|
|||
operation = client.get("/openapi.json").json()["paths"][ |
|||
"/api/{version}/tenants/{tenant_id}/items/{item_id}" |
|||
]["get"] |
|||
assert {parameter["name"] for parameter in operation["parameters"]} == { |
|||
"version", |
|||
"tenant_id", |
|||
"item_id", |
|||
} |
|||
|
|||
|
|||
def test_effective_body_fields_from_app_router_include_and_route_match_openapi(): |
|||
def app_body_dependency(app_body: Annotated[str, Body()]): |
|||
return app_body |
|||
|
|||
def router_body_dependency(router_body: Annotated[int, Body()]): |
|||
return router_body |
|||
|
|||
def include_body_dependency(include_body: Annotated[bool, Body()]): |
|||
return include_body |
|||
|
|||
app = FastAPI(dependencies=[Depends(app_body_dependency)]) |
|||
router = APIRouter(dependencies=[Depends(router_body_dependency)]) |
|||
|
|||
@router.post("/items") |
|||
def create_item(route_body: Annotated[float, Body()]): |
|||
return {"route_body": route_body} |
|||
|
|||
app.include_router( |
|||
router, |
|||
prefix="/api", |
|||
dependencies=[Depends(include_body_dependency)], |
|||
) |
|||
|
|||
client = TestClient(app) |
|||
response = client.post( |
|||
"/api/items", |
|||
json={ |
|||
"app_body": "app", |
|||
"router_body": 1, |
|||
"include_body": True, |
|||
"route_body": 2.5, |
|||
}, |
|||
) |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == {"route_body": 2.5} |
|||
|
|||
schema = client.get("/openapi.json").json() |
|||
request_body_schema = schema["paths"]["/api/items"]["post"]["requestBody"][ |
|||
"content" |
|||
]["application/json"]["schema"] |
|||
body_ref = request_body_schema["$ref"].removeprefix("#/components/schemas/") |
|||
body_schema = schema["components"]["schemas"][body_ref] |
|||
assert set(body_schema["required"]) == { |
|||
"app_body", |
|||
"router_body", |
|||
"include_body", |
|||
"route_body", |
|||
} |
|||
assert set(body_schema["properties"]) == { |
|||
"app_body", |
|||
"router_body", |
|||
"include_body", |
|||
"route_body", |
|||
} |
|||
|
|||
|
|||
def test_later_full_match_wins_over_earlier_included_partial_match(): |
|||
get_router = APIRouter() |
|||
post_router = APIRouter() |
|||
|
|||
@get_router.get("/items") |
|||
def read_items(): # pragma: no cover |
|||
return {"method": "get"} |
|||
|
|||
@post_router.post("/items") |
|||
def create_item(): |
|||
return {"method": "post"} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(get_router, prefix="/api") |
|||
app.include_router(post_router, prefix="/api") |
|||
|
|||
response = TestClient(app).post("/api/items") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == {"method": "post"} |
|||
|
|||
|
|||
def test_included_partial_match_returns_405_when_no_later_full_match_exists(): |
|||
router = APIRouter() |
|||
|
|||
@router.get("/items") |
|||
def read_items(): # pragma: no cover |
|||
return [] |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
|
|||
response = TestClient(app).post("/api/items") |
|||
|
|||
assert response.status_code == 405 |
|||
assert response.headers["allow"] == "GET" |
|||
|
|||
|
|||
def test_included_slash_redirect_does_not_block_later_exact_match(): |
|||
redirect_router = APIRouter() |
|||
exact_router = APIRouter() |
|||
|
|||
@redirect_router.get("/items/") |
|||
def read_items_with_slash(): # pragma: no cover |
|||
return {"path": "slash"} |
|||
|
|||
@exact_router.get("/items") |
|||
def read_items_without_slash(): |
|||
return {"path": "exact"} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(redirect_router, prefix="/api") |
|||
app.include_router(exact_router, prefix="/api") |
|||
|
|||
response = TestClient(app).get("/api/items", follow_redirects=False) |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == {"path": "exact"} |
|||
|
|||
|
|||
def test_failed_included_match_does_not_leak_effective_context_to_later_route(): |
|||
class RejectingRoute(APIRoute): |
|||
def matches(self, scope): |
|||
return Match.NONE, {} |
|||
|
|||
rejecting_router = APIRouter(route_class=RejectingRoute) |
|||
fallback_router = APIRouter() |
|||
|
|||
@rejecting_router.get("/items") |
|||
def rejected_item(): # pragma: no cover |
|||
return {"source": "rejected"} |
|||
|
|||
@fallback_router.get("/items") |
|||
def fallback_item(request: Request): |
|||
fastapi_scope = request.scope.get("fastapi", {}) |
|||
context = fastapi_scope.get("effective_route_context") |
|||
return { |
|||
"source": "fallback", |
|||
"context_path": getattr(context, "path", None), |
|||
} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(rejecting_router, prefix="/api") |
|||
app.include_router(fallback_router, prefix="/api") |
|||
|
|||
response = TestClient(app).get("/api/items") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.json() == {"source": "fallback", "context_path": "/api/items"} |
|||
|
|||
|
|||
def test_included_starlette_mount_keeps_prefix_runtime_and_url_path_for(): |
|||
def mounted_endpoint(request): |
|||
return PlainTextResponse("mounted") |
|||
|
|||
router = APIRouter( |
|||
routes=[ |
|||
Mount( |
|||
"/mounted", |
|||
routes=[Route("/items/{item_id}", mounted_endpoint, name="read_item")], |
|||
name="mounted", |
|||
) |
|||
] |
|||
) |
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
|
|||
client = TestClient(app) |
|||
response = client.get("/api/mounted/items/abc") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.text == "mounted" |
|||
assert ( |
|||
app.url_path_for("mounted:read_item", item_id="abc") == "/api/mounted/items/abc" |
|||
) |
|||
|
|||
|
|||
def test_included_starlette_host_keeps_prefix_runtime_and_url_path_for(): |
|||
def hosted_endpoint(request): |
|||
return PlainTextResponse("hosted") |
|||
|
|||
hosted_app = Router( |
|||
routes=[Route("/items/{item_id}", hosted_endpoint, name="read_item")] |
|||
) |
|||
router = APIRouter( |
|||
routes=[Host("{subdomain}.example.com", hosted_app, name="hosted")] |
|||
) |
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
|
|||
client = TestClient(app, base_url="http://api.example.com") |
|||
response = client.get("/api/items/abc") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.text == "hosted" |
|||
url = app.url_path_for("hosted:read_item", subdomain="api", item_id="abc") |
|||
assert str(url) == "/api/items/abc" |
|||
assert url.host == "api.example.com" |
|||
|
|||
|
|||
def test_restore_fastapi_scope_key_ignores_non_dict_fastapi_scope(): |
|||
scope = {"fastapi": "not-a-dict"} |
|||
|
|||
_restore_fastapi_scope_key(scope, "effective_route_context", object()) |
|||
|
|||
assert scope == {"fastapi": "not-a-dict"} |
|||
|
|||
|
|||
@pytest.mark.anyio |
|||
async def test_included_api_route_without_app_scope_returns_405_response(): |
|||
router = APIRouter() |
|||
|
|||
@router.get("/items") |
|||
def read_items(): # pragma: no cover |
|||
return {"items": []} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
included_router = cast(_IncludedRouter, app.router.routes[-1]) |
|||
effective_context = next(included_router.effective_route_contexts()) |
|||
route = effective_context.original_route |
|||
messages = [] |
|||
|
|||
async def receive(): # pragma: no cover |
|||
return {"type": "http.request", "body": b"", "more_body": False} |
|||
|
|||
async def send(message): |
|||
messages.append(message) |
|||
|
|||
scope = { |
|||
"type": "http", |
|||
"method": "POST", |
|||
"path": "/api/items", |
|||
"raw_path": b"/api/items", |
|||
"root_path": "", |
|||
"scheme": "http", |
|||
"query_string": b"", |
|||
"headers": [], |
|||
"fastapi": {"effective_route_context": effective_context}, |
|||
} |
|||
|
|||
await route.handle(scope, receive, send) |
|||
|
|||
assert messages[0]["type"] == "http.response.start" |
|||
assert messages[0]["status"] == 405 |
|||
assert dict(messages[0]["headers"])[b"allow"] == b"GET" |
|||
|
|||
|
|||
def test_effective_api_route_context_does_not_match_websocket_scope(): |
|||
router = APIRouter() |
|||
|
|||
@router.get("/items") |
|||
def read_items(): # pragma: no cover |
|||
return {"items": []} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
included_router = cast(_IncludedRouter, app.router.routes[-1]) |
|||
effective_context = next(included_router.effective_route_contexts()) |
|||
|
|||
match, child_scope = effective_context.matches( |
|||
{ |
|||
"type": "websocket", |
|||
"path": "/api/items", |
|||
"root_path": "", |
|||
} |
|||
) |
|||
|
|||
assert match == Match.NONE |
|||
assert child_scope == {} |
|||
|
|||
|
|||
def test_effective_api_route_context_url_path_for_no_match(): |
|||
router = APIRouter() |
|||
|
|||
@router.get("/items/{item_id}") |
|||
def read_item(item_id: str): # pragma: no cover |
|||
return {"item_id": item_id} |
|||
|
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
included_router = cast(_IncludedRouter, app.router.routes[-1]) |
|||
effective_context = next(included_router.effective_route_contexts()) |
|||
|
|||
with pytest.raises(NoMatchFound): |
|||
effective_context.url_path_for("missing", item_id="abc") |
|||
|
|||
with pytest.raises(NoMatchFound): |
|||
included_router.url_path_for("missing", item_id="abc") |
|||
|
|||
|
|||
def test_included_starlette_host_without_prefix_keeps_original_app(): |
|||
def hosted_endpoint(request): |
|||
return PlainTextResponse("hosted") |
|||
|
|||
hosted_app = Router( |
|||
routes=[Route("/items/{item_id}", hosted_endpoint, name="read_item")] |
|||
) |
|||
router = APIRouter( |
|||
routes=[Host("{subdomain}.example.com", hosted_app, name="hosted")] |
|||
) |
|||
app = FastAPI() |
|||
app.include_router(router) |
|||
|
|||
client = TestClient(app, base_url="http://api.example.com") |
|||
response = client.get("/items/abc") |
|||
|
|||
assert response.status_code == 200 |
|||
assert response.text == "hosted" |
|||
|
|||
|
|||
class UnknownRoute(BaseRoute): |
|||
def matches(self, scope): # pragma: no cover |
|||
return Match.NONE, {} |
|||
|
|||
async def handle(self, scope, receive, send): # pragma: no cover |
|||
raise AssertionError("UnknownRoute should not be handled") |
|||
|
|||
def url_path_for(self, name, /, **path_params): # pragma: no cover |
|||
raise NoMatchFound(name, path_params) |
|||
|
|||
|
|||
@pytest.mark.anyio |
|||
async def test_included_unknown_route_is_ignored_and_can_return_default_404(): |
|||
router = APIRouter(routes=[UnknownRoute()]) |
|||
app = FastAPI() |
|||
app.include_router(router, prefix="/api") |
|||
included_router = cast(_IncludedRouter, app.router.routes[-1]) |
|||
|
|||
assert included_router.effective_candidates() == [] |
|||
|
|||
messages = [] |
|||
|
|||
async def receive(): # pragma: no cover |
|||
return {"type": "http.request", "body": b"", "more_body": False} |
|||
|
|||
async def send(message): |
|||
messages.append(message) |
|||
|
|||
scope = { |
|||
"type": "http", |
|||
"method": "GET", |
|||
"path": "/api/missing", |
|||
"raw_path": b"/api/missing", |
|||
"root_path": "", |
|||
"scheme": "http", |
|||
"query_string": b"", |
|||
"headers": [], |
|||
"fastapi": {}, |
|||
} |
|||
|
|||
await included_router._handle_selected(scope, receive, send) |
|||
|
|||
assert messages[0]["type"] == "http.response.start" |
|||
assert messages[0]["status"] == 404 |
|||
|
|||
|
|||
def test_no_prefix_include_validation_sees_effective_starlette_route_candidates(): |
|||
def endpoint(request): # pragma: no cover |
|||
return PlainTextResponse("ok") |
|||
|
|||
child_router = APIRouter(routes=[Route("/items", endpoint, name="read_items")]) |
|||
parent_router = APIRouter() |
|||
parent_router.include_router(child_router, prefix="/child") |
|||
|
|||
candidates = list(_iter_included_route_candidates(parent_router.routes)) |
|||
|
|||
assert cast(Route, candidates[0]).path == "/child/items" |
|||
|
|||
|
|||
def test_apirouter_matches_fallback_without_include_context(): |
|||
router = APIRouter() |
|||
|
|||
def read_items(request): # pragma: no cover |
|||
return PlainTextResponse("items") |
|||
|
|||
router.add_route("/items", read_items) |
|||
|
|||
assert router.matches({"type": "http", "path": "/items", "root_path": ""}) == ( |
|||
Match.NONE, |
|||
{}, |
|||
) |
|||
|
|||
|
|||
@pytest.mark.anyio |
|||
async def test_apirouter_handle_fallback_without_include_context(): |
|||
router = APIRouter() |
|||
|
|||
def read_items(request): |
|||
return PlainTextResponse("items") |
|||
|
|||
router.add_route("/items", read_items) |
|||
messages = [] |
|||
|
|||
async def receive(): # pragma: no cover |
|||
return {"type": "http.request", "body": b"", "more_body": False} |
|||
|
|||
async def send(message): |
|||
messages.append(message) |
|||
|
|||
scope = { |
|||
"type": "http", |
|||
"method": "GET", |
|||
"path": "/items", |
|||
"raw_path": b"/items", |
|||
"root_path": "", |
|||
"scheme": "http", |
|||
"query_string": b"", |
|||
"headers": [], |
|||
} |
|||
|
|||
await router.handle(scope, receive, send) |
|||
|
|||
assert messages[0]["type"] == "http.response.start" |
|||
assert messages[0]["status"] == 200 |
|||
assert messages[1]["body"] == b"items" |
|||
Loading…
Reference in new issue