Function test_lifespan_state_async_cm

tests/test_routing.py:645–690

(test_client_factory: TestClientFactory)

def test_lifespan_state_async_cm(test_client_factory: TestClientFactory) -> None:
    startup_complete = False
    shutdown_complete = False

    class State(TypedDict):
        count: int
        items: list[int]

    async def hello_world(request: Request) -> Response:
        # modifications to the state should not leak across requests
        assert request.state.count == 0
        # modify the state, this should not leak to the lifespan or other requests
        request.state.count += 1
        # since state.items is a mutable object this modification _will_ leak across
        # requests and to the lifespan
        request.state.items.append(1)
        return PlainTextResponse("hello, world")

    @contextlib.asynccontextmanager
    async def lifespan(app: Starlette) -> AsyncIterator[State]:
        nonlocal startup_complete, shutdown_complete
        startup_complete = True
        state = State(count=0, items=[])
        yield state
        shutdown_complete = True
        # modifications made to the state from a request do not leak to the lifespan
        assert state["count"] == 0
        # unless of course the request mutates a mutable object that is referenced
        # via state
        assert state["items"] == [1, 1]

    app = Router(
        lifespan=lifespan,
        routes=[Route("/", hello_world)],
    )

    assert not startup_complete
    assert not shutdown_complete
    with test_client_factory(app) as client:
        assert startup_complete
        assert not shutdown_complete
        client.get("/")
        # Calling it a second time to ensure that the state is preserved.
        client.get("/")
    assert startup_complete
    assert shutdown_complete


def test_raise_on_startup(test_client_factory: TestClientFactory) -> None: