FastAPI自学教程（69） - 事件测试机制

1.基础测试方法

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()
client = TestClient(app)

def test_basic_event():
    # 测试标准请求响应
    response = client.get("/items/42")
    assert response.status_code == 200

• 使用TestClient进行同步测试，验证常规请求的响应状态和内容
• 流程图展示基础测试流程：

  graph TD
    A[初始化TestClient] --> B[发送模拟请求]
    B --> C[接收响应数据]
    C --> D[断言验证结果]

2.异步事件测试

import pytest
from httpx import ASGITransport, AsyncClient

async def test_async_lifespan():
    # 测试生命周期事件
    async with AsyncClient(transport=ASGITransport(app=app)) as client:
        response = await client.get("/predict")
        assert response.json()["model_loaded"] is True

• 通过AsyncClient配合ASGITransport测试异步生命周期事件
• 需使用pytest.mark.anyio标记异步测试函数
• 支持测试模型加载、数据库连接池初始化等预启动逻辑

3.生命周期事件验证

from contextlib import asynccontextmanager

@asynccontextmanager
async def test_lifespan():
    # 模拟资源加载
    app.state.cache = RedisPool()
    yield
    # 验证资源释放
    assert not hasattr(app.state, "cache")

def test_resource_cleanup():
    with pytest.raises(AttributeError):
        test_lifespan()

• 使用异步上下文管理器验证启动/关闭阶段的资源管理
• 可结合pytest断言验证资源加载和释放的正确性

4.生产环境测试建议

1. 性能基准测试：

   • 使用pytest-benchmark测量事件处理耗时
   • 建议模型加载时间<5秒（基于1GB内存占用）

2. 错误场景覆盖：

   def test_failed_initialization():
       with pytest.raises(RuntimeError):
           async def broken_lifespan():
               raise RuntimeError("Init failed")
           app = FastAPI(lifespan=broken_lifespan)

   验证初始化失败时的应用状态

3. 分布式事件测试：

   • 使用fastapi-events库测试跨服务事件分发
   • 验证消息队列的消费延迟（建议<200ms）

测试命令示例：

# 运行所有测试用例
pytest tests/ -v

# 生成覆盖率报告
pytest --cov=app --cov-report=html