如何测试使用FastAPI lifespan函数的接口？

问题根源 #

TestClient默认不会自动执行FastAPI应用的lifespan异步上下文管理器，导致接口调用时资源未完成初始化，触发KeyError。

解决方案 #

方案1：用异步测试框架手动触发Lifespan #

针对异步lifespan，需配合pytest-asyncio手动管理上下文：

步骤1：优化API代码（推荐用app.state替代全局变量）

避免全局变量引发的多测试用例状态污染，将资源挂载到应用实例的state属性：

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 初始化ML模型资源
    predictor = Predictor(model_version)
    app.state.predict_func = predictor.predict_from_features
    yield
    # 清理资源
    del app.state.predict_func

app = FastAPI(lifespan=lifespan)

@app.get("/prediction/")
async def get_prediction(model_input: str):
    prediction = app.state.predict_func(model_input)
    return {"prediction": prediction}

步骤2：编写异步测试用例

安装pytest-asyncio后，手动启动/关闭lifespan上下文：

import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

@pytest.mark.asyncio
async def test_read_prediction():
    # 启动lifespan上下文
    async with app.router.lifespan_context(app):
        model_input = "test"
        response = client.get(f"/prediction/?model_input={model_input}")
        assert response.status_code == 200
        # 可根据实际业务添加返回值断言
        # assert response.json()["prediction"] == expected_result

方案2：使用AsyncClient自动处理Lifespan（FastAPI 0.100+） #

若FastAPI版本≥0.100.0，可直接用httpx.AsyncClient，它会自动触发lifespan的启动与销毁：

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_read_prediction():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        model_input = "test"
        response = await ac.get(f"/prediction/?model_input={model_input}")
        assert response.status_code == 200

方案3：同步场景下手动模拟Lifespan #

如果你的Predictor是同步实现，可在测试前后手动初始化/清理资源：

from fastapi.testclient import TestClient
from app.main import app, Predictor, model_version

client = TestClient(app)

def test_read_prediction():
    # 手动初始化资源（模拟lifespan启动逻辑）
    app.state.predict_func = Predictor(model_version).predict_from_features
    try:
        model_input = "test"
        response = client.get(f"/prediction/?model_input={model_input}")
        assert response.status_code == 200
    finally:
        # 手动清理资源（模拟lifespan销毁逻辑）
        del app.state.predict_func

关键提示 #

• 始终优先使用app.state存储应用级资源，避免全局变量带来的测试状态污染问题。
• 异步lifespan必须配合异步测试框架才能正确执行上下文逻辑。
• AsyncClient是异步场景下最简洁的测试方案，无需手动管理lifespan生命周期。

内容的提问来源于stack exchange，提问作者lux7