如何使用LifespanManager异步测试FastAPI反向代理

按照FastAPI官方文档，我需要用LifespanManager进行异步测试，编写了如下lifespan函数：

@asynccontextmanager
async def lifespan(_app: FastAPI):
    async with httpx.AsyncClient(base_url=env.proxy_url, transport=httpx.MockTransport(dummy_response)) as client:
        yield {'client': client}
        await client.aclose()

我要测试名为proxy的反向代理接口，编写测试代码后遇到一系列问题：

1. 初始测试代码报错：

TypeError: 'FastAPI' object does not support the asynchronous context manager protocol

2. 修改代码后出现属性错误：

FAILED tests/regression/test_proxy.py::test_proxy_get_request - AttributeError: 'State' object has no attribute 'client'
手动设置state后仍无法在proxy函数的request中获取client；

3. 参考建议修正state问题后，又出现TaskGroup未处理错误，根源为：

E asyncio.exceptions.CancelledError: Cancelled by cancel scope 105cdb8f0

以下是完整测试代码：

import pytest
import pytest_asyncio
from fastapi import FastAPI
from contextlib import asynccontextmanager
from asgi_lifespan import LifespanManager
import httpx
from httpx import Response
import importlib
import uvicorn

import proxy
import env

def dummy_response(_request):
    res = Response(200, content="Mock response")
    res.headers['Content-Type'] = 'text/plain; charset=utf-8'
    return res

@pytest_asyncio.fixture
async def _client():
    with pytest.MonkeyPatch.context() as monkeypatch:
        monkeypatch.setenv("PROXY_URL", "http://proxy")

        importlib.reload(env)
        importlib.reload(proxy)

    @asynccontextmanager
    async def lifespan(_app: FastAPI):
        async with httpx.AsyncClient(
             base_url=env.proxy_url, 
             transport=httpx.MockTransport(dummy_response)) as client:
            yield {'client': client} # startup
            await client.aclose() # shutdown

    app = FastAPI(lifespan=lifespan)
    app.add_route("/proxy/path", proxy.proxy)

    async with LifespanManager(app) as manager:
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=manager.app), base_url=env.proxy_url) as client:
                yield client

@pytest.mark.asyncio
async def test_proxy_get_request(_client):
        response = await _client.get(f"/proxy/path", params={"query": "param"})
        assert response.status_code == 200

解决方案 #

1. 修复"FastAPI对象不支持异步上下文管理器协议"错误 #

该错误是因为直接将FastAPI实例当作异步上下文管理器使用，正确方式是通过LifespanManager统一管理应用生命周期，你的代码已引入该工具，需确保fixture中上下文嵌套逻辑正确。

2. 修复"State对象没有client属性"错误 #

FastAPI的lifespan函数中，yield字典的方式不会自动将值注入app.state，需直接将client赋值给应用状态；同时接口函数要通过request.app.state而非request.state获取client：

调整lifespan函数：

@asynccontextmanager
async def lifespan(_app: FastAPI):
    async with httpx.AsyncClient(
         base_url=env.proxy_url, 
         transport=httpx.MockTransport(dummy_response)) as client:
        _app.state.client = client  # 直接绑定到应用状态
        yield  # 无需返回字典
        # 移除手动调用client.aclose()，async with会自动处理关闭

确保proxy接口函数正确取值：

from fastapi import Request

async def proxy(request: Request):
    client = request.app.state.client
    # 后续代理逻辑实现

3. 修复CancelledError错误 #

该错误源于生命周期管理的嵌套冲突，调整fixture的上下文结构，确保LifespanManager覆盖整个测试周期，同时区分测试客户端与代理客户端的URL：

修改后的完整fixture代码：

@pytest_asyncio.fixture
async def _client():
    with pytest.MonkeyPatch.context() as monkeypatch:
        monkeypatch.setenv("PROXY_URL", "http://proxy")
        importlib.reload(env)
        importlib.reload(proxy)

    @asynccontextmanager
    async def lifespan(_app: FastAPI):
        async with httpx.AsyncClient(
             base_url=env.proxy_url, 
             transport=httpx.MockTransport(dummy_response)) as client:
            _app.state.client = client
            yield

    app = FastAPI(lifespan=lifespan)
    app.add_route("/proxy/path", proxy.proxy)

    async with LifespanManager(app):
        # 测试客户端使用固定测试URL，避免与代理目标URL混淆
        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app), 
            base_url="http://testserver"
        ) as client:
            yield client

内容的提问来源于stack exchange，提问作者roberto tomás