FastAPI TestClient not starting lifetime in test

Question

Example code:

import os
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    print(f'Lifetime ON {os.getpid()=}')
    app.state.global_rw = 0

    _ = asyncio.create_task(infinite_1(app.state), name='my_task')
    yield 

app = FastAPI(lifespan=lifespan)

@app.get("/state/") 
async def inc(request: Request):
    return {'rw': request.app.state.global_rw}

async def infinite_1(app_rw_state):
    print('infinite_1 ON')
    while True:
        app_rw_state.global_rw += 1
        print(f'infinite_1 {app_rw_state.global_rw=}')
        await asyncio.sleep(10) 

This is all working fine, every 10 seconds app.state.global_rw is increased by one.

Test code:

from fastapi.testclient import TestClient

def test_all():
    from a_10_code import app 
    client = TestClient(app)

    response = client.get("/state/")
    assert response.status_code == 200
    assert response.json() == {'rw': 0}

Problem that I have found is that TestClient(app) will not start async def lifespan(app: FastAPI):.
Started with pytest -s a_10_test.py

So, how to start lifespan in FastAPI TestClient ?

P.S. my real code is more complex, this is just simple example for demonstration purposes.

Answer 1

The main reason that the written test fails is that it doesn't handle the asynchronous nature of the FastAPI app's lifespan context properly. In fact, the global_rw is not set due to improper initialization.

If you don't want to utilize an AsyncClient like the one by httpx you can use pytest_asyncio and the async fixture, ensuring that the FastAPI app's lifespan context correctly works and global_rw is initialized properly.

Here's the workaround:

import pytest_asyncio
import pytest
import asyncio

from fastapi.testclient import TestClient
from fastapi_lifespan import app

@pytest_asyncio.fixture(scope="module")
def client():
    with TestClient(app) as client:
        yield client

@pytest.mark.asyncio
async def test_state(client):
    response = client.get("/state/")
    assert response.status_code == 200
    assert response.json() == {"rw": 1}

    await asyncio.sleep(11)

    response = client.get("/state/")
    assert response.status_code == 200
    assert response.json() == {'rw': 2}

You can also define a conftest.py to place the fixture there to have a clean test files.

Answer 2

I think the problem might be related of an sync/async issue.

Since you're writing an async lifespan, you probably will need to use and async client with the asyncio pytest plugins.

Here I define an async_fixture get_client, that allows us to inject an async test client in your test.

import pytest_asyncio
import httpx
from typing import AsyncGenerator

@pytest_asyncio.fixture()
async def get_client() -> AsyncGenerator[httpx.AsyncClient]:
    from a_10_code import app
    transport = httpx.ASGITransport(app=app)

    async with httpx.AsyncClient(
        transport=transport,
        base_url="http://testserver"
    ) as client:
        yield client

async def test_all(get_client: httpx.AsyncClient):
    response = await get_client.get("/state")
    assert response.status_code == 200
    assert response.json() == {'rw': 0}

Read more information:

• Httpx
• Pytest Asyncio