Browse Source

fix: fix pydanticV2.5 tests

pull/10646/head
Nikita Pastukhov 2 years ago
parent
commit
3e6288647f
  1. 17
      fastapi/routing.py
  2. 3
      tests/test_annotated.py
  3. 62
      tests/test_router_events.py

17
fastapi/routing.py

@ -120,19 +120,6 @@ def _prepare_response_content(
return dataclasses.asdict(res)
return res
def _merge_lifespan_context(
original_context: Lifespan[Any], nested_context: Lifespan[Any]
) -> Lifespan[Any]:
@asynccontextmanager
async def merged_lifespan(app: AppType) -> AsyncIterator[Mapping[str, Any]]:
async with original_context(app) as maybe_self_context:
async with nested_context(app) as maybe_nested_context:
yield {**(maybe_self_context or {}), **(maybe_nested_context or {})}
return merged_lifespan
async def serialize_response(
*,
field: Optional[ModelField] = None,
@ -1299,10 +1286,6 @@ class APIRouter(routing.Router):
self.add_event_handler("startup", handler)
for handler in router.on_shutdown:
self.add_event_handler("shutdown", handler)
self.lifespan_context = _merge_lifespan_context(
self.lifespan_context,
router.lifespan_context,
)
def get(
self,

3
tests/test_annotated.py

@ -57,7 +57,7 @@ foo_is_short = {
{
"ctx": {"min_length": 1},
"loc": ["query", "foo"],
"msg": "String should have at least 1 characters",
"msg": "String should have at least 1 character",
"type": "string_too_short",
"input": "",
"url": match_pydantic_error_url("string_too_short"),
@ -75,7 +75,6 @@ foo_is_short = {
]
}
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[

62
tests/test_router_events.py

@ -2,7 +2,7 @@ from contextlib import asynccontextmanager
from typing import AsyncGenerator, Dict
import pytest
from fastapi import APIRouter, FastAPI, Request
from fastapi import APIRouter, FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel
@ -109,63 +109,3 @@ def test_app_lifespan_state(state: State) -> None:
assert response.json() == {"message": "Hello World"}
assert state.app_startup is True
assert state.app_shutdown is True
def test_router_nested_lifespan_state(state: State) -> None:
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
state.app_startup = True
yield {"app": True}
state.app_shutdown = True
@asynccontextmanager
async def router_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
state.router_startup = True
yield {"router": True}
state.router_shutdown = True
@asynccontextmanager
async def subrouter_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
state.sub_router_startup = True
yield {"sub_router": True}
state.sub_router_shutdown = True
sub_router = APIRouter(lifespan=subrouter_lifespan)
router = APIRouter(lifespan=router_lifespan)
router.include_router(sub_router)
app = FastAPI(lifespan=lifespan)
app.include_router(router)
@app.get("/")
def main(request: Request) -> Dict[str, str]:
assert request.state.app
assert request.state.router
assert request.state.sub_router
return {"message": "Hello World"}
assert state.app_startup is False
assert state.router_startup is False
assert state.sub_router_startup is False
assert state.app_shutdown is False
assert state.router_shutdown is False
assert state.sub_router_shutdown is False
with TestClient(app) as client:
assert state.app_startup is True
assert state.router_startup is True
assert state.sub_router_startup is True
assert state.app_shutdown is False
assert state.router_shutdown is False
assert state.sub_router_shutdown is False
response = client.get("/")
assert response.status_code == 200, response.text
assert response.json() == {"message": "Hello World"}
assert state.app_startup is True
assert state.router_startup is True
assert state.sub_router_startup is True
assert state.app_shutdown is True
assert state.router_shutdown is True
assert state.sub_router_shutdown is True

Loading…
Cancel
Save