FastAPI lifespan state tested with httpx ASGITransport and asgi-lifespan: working versions (tested 2026-10-15)

fastapi==0.143.1
httpx==0.28.1
asgi-lifespan==2.1.0
pytest-asyncio==1.4.0
pytest==9.1.1

import sqlite3
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Request
from pydantic import BaseModel


class NoteIn(BaseModel):
    body: str


def get_db(request: Request) -> sqlite3.Connection:
    return request.app.state.db


DB = Annotated[sqlite3.Connection, Depends(get_db)]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # check_same_thread=False: sync routes run in a worker thread
    app.state.db = sqlite3.connect(":memory:", check_same_thread=False)
    app.state.db.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT NOT NULL)")
    app.state.db.commit()
    try:
        yield
    finally:
        app.state.db.close()


def create_app() -> FastAPI:
    app = FastAPI(lifespan=lifespan)

    @app.get("/notes/{note_id}")
    def read_note(note_id: int, db: DB):
        row = db.execute("SELECT body FROM notes WHERE id = ?", (note_id,)).fetchone()
        if row is None:
            raise HTTPException(404, "note not found")
        return {"id": note_id, "body": row[0]}

    @app.post("/notes", status_code=201)
    def create_note(note: NoteIn, db: DB):
        cur = db.execute("INSERT INTO notes (body) VALUES (?)", (note.body,))
        db.commit()
        return {"id": cur.lastrowid, "body": note.body}

    return app

import sqlite3

import httpx
import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport

from app import create_app, get_db


@pytest.fixture
async def client():
    # httpx's ASGITransport does not send lifespan events, so wrap the app
    app = create_app()
    async with LifespanManager(app) as manager:
        transport = ASGITransport(app=manager.app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
            yield c


async def test_create_then_read(client):
    created = await client.post("/notes", json={"body": "hello"})
    assert created.status_code == 201
    note_id = created.json()["id"]
    got = await client.get(f"/notes/{note_id}")
    assert got.json() == {"id": note_id, "body": "hello"}


async def test_missing_row_is_404(client):
    assert (await client.get("/notes/999")).status_code == 404


async def test_without_lifespan_state_is_missing():
    # No LifespanManager: app.state.db was never set, so the dependency fails
    app = create_app()
    async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        with pytest.raises(AttributeError):
            await c.get("/notes/1")


async def test_shutdown_closes_db():
    app = create_app()
    async with LifespanManager(app):
        db = app.state.db
    with pytest.raises(sqlite3.ProgrammingError):
        db.execute("SELECT 1")


async def test_each_app_gets_its_own_state():
    # Module-level app instances would share state across tests; factories do not
    async with LifespanManager(create_app()) as first, LifespanManager(create_app()) as second:
        async with httpx.AsyncClient(transport=ASGITransport(app=first.app), base_url="http://test") as c1:
            await c1.post("/notes", json={"body": "only in first"})
        async with httpx.AsyncClient(transport=ASGITransport(app=second.app), base_url="http://test") as c2:
            assert (await c2.get("/notes/1")).status_code == 404


async def test_dependency_override_skips_lifespan():
    app = create_app()
    db = sqlite3.connect(":memory:", check_same_thread=False)
    db.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT NOT NULL)")
    db.execute("INSERT INTO notes (body) VALUES ('fake')")
    app.dependency_overrides[get_db] = lambda: db
    async with httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        r = await c.get("/notes/1")
    assert r.json()["body"] == "fake"
    db.close()

[pytest]
pythonpath = .
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function