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