From 9607d6ff949850cd58de17e1af4643ad83ee3545 Mon Sep 17 00:00:00 2001 From: d3vyce Date: Thu, 22 Oct 2026 13:05:40 -0400 Subject: [PATCH] feat: run the app lifespan in create_async_client --- docs/module/pytest.md | 31 ++++++++ src/fastapi_toolsets/pytest/utils.py | 47 +++++++++++-- tests/test_pytest.py | 101 +++++++++++++++++++++++++++ 3 files changed, 174 insertions(+), 5 deletions(-) diff --git a/docs/module/pytest.md b/docs/module/pytest.md index 99d1653..af478b5 100644 --- a/docs/module/pytest.md +++ b/docs/module/pytest.md @@ -52,6 +52,37 @@ async with create_async_client( ... ``` +### Lifespan + +The client does not run the app's lifespan by default. Pass `lifespan=True` to run startup before the client opens and shutdown after it closes: + +```python +@asynccontextmanager +async def lifespan(app): + yield {"cache": {}} + + +app = FastAPI(lifespan=lifespan) + + +@app.get("/cache") +async def read_cache(request: Request): + return request.state.cache + + +@pytest.fixture +async def http_client(): + async with create_async_client(app=app, lifespan=True) as c: + yield c +``` + +State yielded by the lifespan is available on `request.state`, as a shallow copy per request just like under uvicorn. Shutdown runs even when the test fails, and an exception raised by startup or shutdown propagates unchanged. The client works as a fixture under both the anyio plugin and pytest-asyncio. + +The lifespan is called through `app.router.lifespan_context`, so `app` must be a FastAPI or Starlette app, and middleware does not see the lifespan events. + +!!! warning "Shutdown runs too" + A lifespan that tears down shared resources, such as one wired by `db.install(app)`, disposes the engine when the client closes. Open a client with `lifespan=True` only where that is what you want. + ## Database sessions Use [`create_worker_database`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_worker_database) + [`create_db_session`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_db_session) to get a fully isolated `AsyncSession` for each test: diff --git a/src/fastapi_toolsets/pytest/utils.py b/src/fastapi_toolsets/pytest/utils.py index 84c6f16..9aa64e7 100644 --- a/src/fastapi_toolsets/pytest/utils.py +++ b/src/fastapi_toolsets/pytest/utils.py @@ -2,7 +2,7 @@ import os from collections.abc import AsyncGenerator, Callable -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, nullcontext from typing import Any from weakref import WeakKeyDictionary @@ -61,6 +61,28 @@ def _pop_overrides(app: Any, overrides: dict[Any, Any], owner: Any) -> None: del _override_layers[app] +@asynccontextmanager +async def _run_lifespan(app: Any) -> AsyncGenerator[Any, None]: + """Run *app*'s lifespan around the block and yield *app* bound to its state.""" + context = app.router.lifespan_context(app) + state = await context.__aenter__() or {} + try: + yield _with_state(app, state) + finally: + # Like a server, shut down cleanly even when the block raised. + await context.__aexit__(None, None, None) + + +def _with_state(app: Any, state: dict[str, Any]) -> Any: + """Wrap *app* so each request scope gets a copy of the lifespan *state*.""" + + async def wrapped(scope: Any, receive: Any, send: Any) -> None: + scope["state"] = state.copy() + await app(scope, receive, send) + + return wrapped + + def _get_xdist_worker(default_test_db: str) -> str: """Return the pytest-xdist worker name, or *default_test_db* when not running under xdist. @@ -201,6 +223,8 @@ async def create_async_client( app: Any, base_url: str = "http://test", dependency_overrides: dict[Callable[..., Any], Callable[..., Any]] | None = None, + *, + lifespan: bool = False, **kwargs: Any, ) -> AsyncGenerator[AsyncClient, None]: """Create an async httpx client for testing FastAPI applications. @@ -211,6 +235,9 @@ async def create_async_client( dependency_overrides: Optional mapping of original dependencies to their test replacements. Applied via ``app.dependency_overrides`` before yielding and restored to their previous state after. + lifespan: Run the app's startup before yielding and its shutdown + after, through ``app.router.lifespan_context``. State yielded by + the lifespan is exposed on ``request.state``. Defaults to False. **kwargs: Additional keyword arguments forwarded to :class:`httpx.AsyncClient` (e.g. ``headers``, ``cookies``, ``auth``, ``timeout``). @@ -255,17 +282,27 @@ async def override(): ) as c: yield c ``` + + Example with the app lifespan: + ```python + @pytest.fixture + async def client(): + async with create_async_client(app, lifespan=True) as c: + yield c + ``` """ overrides = dependency_overrides or {} owner = object() - transport = ASGITransport(app=app) try: if overrides: _push_overrides(app, overrides, owner) - async with AsyncClient( - transport=transport, base_url=base_url, **kwargs - ) as client: + async with ( + _run_lifespan(app) if lifespan else nullcontext(app) as asgi_app, + AsyncClient( + transport=ASGITransport(app=asgi_app), base_url=base_url, **kwargs + ) as client, + ): yield client finally: if overrides: diff --git a/tests/test_pytest.py b/tests/test_pytest.py index d439764..add4cdb 100644 --- a/tests/test_pytest.py +++ b/tests/test_pytest.py @@ -1,7 +1,9 @@ """Tests for ``fastapi_toolsets.pytest``: generated fixtures, clients and databases.""" +import asyncio import uuid from collections.abc import Callable +from contextlib import asynccontextmanager from typing import Any, cast import pytest @@ -296,6 +298,105 @@ async def headers_endpoint(request: Request): assert client.is_closed +def _lifespan_app(events: list[str], *, fail_on: str | None = None) -> FastAPI: + """An app whose lifespan logs to *events*, seeds state and may raise.""" + + @asynccontextmanager + async def lifespan(app: FastAPI): + events.append("startup") + if fail_on == "startup": + raise ValueError("startup boom") + yield {"greeting": "hello"} + events.append("shutdown") + if fail_on == "shutdown": + raise ValueError("shutdown boom") + + app = FastAPI(lifespan=lifespan) + + @app.get("/state") + async def state_endpoint(request: Request): + touched = getattr(request.state, "touched", False) + request.state.touched = True + return {"greeting": request.state.greeting, "touched": touched} + + return app + + +class TestLifespan: + """``lifespan=True`` runs the app's startup and shutdown around the client.""" + + @pytest.mark.anyio + async def test_startup_state_and_shutdown(self): + """State reaches requests as a per-request copy; shutdown waits for exit.""" + events: list[str] = [] + + async with create_async_client(_lifespan_app(events), lifespan=True) as client: + assert events == ["startup"] + for _ in range(2): + response = await client.get("/state") + assert response.json() == {"greeting": "hello", "touched": False} + + assert events == ["startup", "shutdown"] + + @pytest.mark.anyio + async def test_off_by_default(self): + events: list[str] = [] + + async with create_async_client(_lifespan_app(events)): + pass + + assert events == [] + + @pytest.mark.anyio + async def test_startup_failure_raises_and_restores_overrides(self): + events: list[str] = [] + app = _lifespan_app(events, fail_on="startup") + original = _dep("original") + + with pytest.raises(ValueError, match="startup boom"): + async with create_async_client( + app, dependency_overrides={original: _dep("x")}, lifespan=True + ): + pass # pragma: no cover + + assert events == ["startup"] + assert original not in app.dependency_overrides + + @pytest.mark.anyio + async def test_shutdown_failure_raises(self): + events: list[str] = [] + + with pytest.raises(ValueError, match="shutdown boom"): + async with create_async_client( + _lifespan_app(events, fail_on="shutdown"), lifespan=True + ): + pass + + assert events == ["startup", "shutdown"] + + @pytest.mark.anyio + async def test_error_in_block_is_not_wrapped_and_shutdown_runs(self): + events: list[str] = [] + + with pytest.raises(KeyError): + async with create_async_client(_lifespan_app(events), lifespan=True): + raise KeyError("test failure") + + assert events == ["startup", "shutdown"] + + @pytest.mark.anyio + async def test_enter_and_exit_in_different_tasks(self): + """pytest-asyncio runs fixture setup and teardown in separate tasks.""" + events: list[str] = [] + context = create_async_client(_lifespan_app(events), lifespan=True) + + client = await asyncio.create_task(context.__aenter__()) + assert (await client.get("/state")).json()["greeting"] == "hello" + await asyncio.create_task(context.__aexit__(None, None, None)) + + assert events == ["startup", "shutdown"] + + class TestDependencyOverrides: """Overrides are layered per client and put back exactly as they were."""