FastAPI自学教程（71） - 异步测试实践

1.基础用法

import pytest
from httpx import ASGITransport, AsyncClient
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), 
        base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

使用AsyncClient替代TestClient处理异步请求，支持非阻塞I/O操作
必须添加@pytest.mark.anyio标记声明异步测试函数
通过ASGITransport直接挂载FastAPI应用实例，绕过HTTP协议栈提升测试速度

流程图：

graph TD
    A[启动AsyncClient] --> B[建立ASGI连接]
    B --> C[发送异步请求]
    C --> D[接收响应数据]
    D --> E[断言验证结果]
    E --> F[关闭连接]

2.进阶配置

2.1 生命周期事件处理

from asgi_lifespan import LifespanManager

async def test_lifespan():
    async with LifespanManager(app) as manager:
        if manager.should_exit:
            raise RuntimeError("启动失败")

使用asgi-lifespan库触发startup/shutdown事件
验证应用初始化阶段数据库连接池等资源的加载状态

2.2 多级异步依赖验证

async def test_nested_async():
    async with AsyncClient(...) as ac:
        # 并行发送多个请求
        req1 = ac.get("/api1")
        req2 = ac.post("/api2")
        res1, res2 = await asyncio.gather(req1, req2)
        # 验证关联业务逻辑

利用asyncio.gather实现并发请求测试
支持数据库事务的原子性验证

3.注意事项

事件循环管理：
- 避免在测试函数外创建需要事件循环的对象（如数据库连接池）
- 使用async with上下文管理器自动清理资源

异常捕获策略：

1
2
3

with pytest.raises(HTTPException) as exc_info:
    await ac.get("/error-endpoint")
assert exc_info.value.status_code == 500

验证特定HTTP异常状态码

Mock策略：

1 2	from unittest.mock import AsyncMock app.dependency_overrides[real_service] = AsyncMock(return_value=mock_data)

使用AsyncMock模拟异步依赖项

4.生产环境建议

性能基准测试：
- 使用pytest-benchmark测量请求延迟，建议单接口平均响应<200ms
- 通过ab工具模拟1000+并发连接压力测试

持续集成配置：

# .github/workflows/tests.yml
- name: Run async tests
  run: |
    pip install -r requirements.txt
    pytest --cov=app --cov-report=xml

集成覆盖率报告与异步测试流水线

测试命令示例：

# 安装测试依赖
pip install httpx pytest-asyncio anyio

# 运行测试套件
pytest -v --cov=app tests/