pythonasynchronouspytestpython-asynciofastapi

FastAPI TestClient not starting lifetime in test


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.


Solution

  • 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.