直接用 @pytest.mark.asyncio 会报 runtimeerror: event loop is closed,因为 pytest 默认不管理事件循环生命周期,testclient 与测试函数可能复用或冲突已关闭的 loop;正确做法是改用 httpx.asyncclient 配合 pytest-asyncio 和 lifespan 管理。

为什么直接用 pytest.mark.asyncio 会报 RuntimeError: Event loop is closed
FastAPI 的接口函数是 async def,但默认 pytest 运行在同步上下文中。直接调用 await client.get("/path") 会触发 RuntimeError: Event loop is closed,因为 pytest 默认不管理事件循环生命周期,且多个测试间 event loop 可能被意外关闭或复用失败。
根本原因不是“没加 async”,而是测试运行器没正确启动/隔离 asyncio event loop。
- 别用
pytest-asyncio的全局@pytest.mark.asyncio装饰单个测试函数(尤其在 FastAPI + TestClient 混用时容易出 loop 冲突) - 不要手动调用
asyncio.run()—— 它强制新建 loop,而TestClient内部已绑定当前 loop,会导致 context mismatch - 必须让
TestClient和测试函数共用同一个、受控的 event loop
正确做法:用 httpx.AsyncClient + pytest-asyncio 配合 startup/shutdown
FastAPI 官方推荐测试异步路径的方式是弃用 TestClient(它是同步的),改用 httpx.AsyncClient,并配合 pytest-asyncio 管理 event loop。
关键点:app 必须支持 lifespan,否则 startup 里的数据库连接、缓存初始化等逻辑不会执行,测试环境和真实运行不一致。
- 确保你的
app使用了lifespan(不是on_event):
from fastapi import FastAPI from contextlib import asynccontextmanager <p>@asynccontextmanager async def lifespan(app: FastAPI):</p><h1>startup</h1><pre class="brush:python;toolbar:false;">await init_db() yield # shutdown await close_db()
app = FastAPI(lifespan=lifespan)
- 测试文件顶部启用
pytest-asyncio的自动 event loop 管理:
import pytest pytest_plugins = ["pytest_asyncio"]
- 测试函数用
@pytest.mark.asyncio,client 实例化为httpx.AsyncClient:
import pytest
from httpx import AsyncClient
from main import app # your FastAPI app
<p>@pytest.mark.asyncio
async def test_read_items():
async with AsyncClient(app=app, base_url="<a href="https://www.php.cn/link/9688c999c6508777280b6e8074ad82fa">https://www.php.cn/link/9688c999c6508777280b6e8074ad82fa</a>") as ac:
response = await ac.get("/items/")
assert response.status_code == 200
assert response.json() == [{"name": "foo"}]</p>
测试带依赖(如 Depends)的路由时如何 mock
FastAPI 的依赖注入在异步测试中不能靠覆盖 app.dependency_overrides 后再用 TestClient——因为 TestClient 不走完整 async stack。必须用 AsyncClient,且 mock 要在 lifespan 启动前注入。
- mock 必须在
lifespan中生效,否则依赖(比如数据库 session)仍会尝试真实连接 - 推荐方式:在测试函数内临时 override,并确保它在
AsyncClient生命周期内起作用:
from unittest.mock import AsyncMock
from fastapi import Depends
<p>@pytest.mark.asyncio
async def test_create_item_with_mock_db():
mock_db = AsyncMock()
mock_db.execute.return_value = AsyncMock()</p><pre class="brush:python;toolbar:false;">app.dependency_overrides[get_db] = lambda: mock_db # 注意:这里返回的是 callable,不是实例
async with AsyncClient(app=app, base_url="https://www.php.cn/link/9688c999c6508777280b6e8074ad82fa") as ac:
response = await ac.post("/items/", json={"name": "test"})
assert response.status_code == 200
app.dependency_overrides.clear() # 必须清理,避免污染其他测试
- 如果依赖本身是 async generator(比如
async def get_db()),mock 必须是 async iterator 或用AsyncMock模拟__aiter__ -
dependency_overrides是全局字典,跨测试不自动隔离,.clear()是硬性要求
常见错误响应码与调试建议
异步测试失败时,422 Unprocessable Entity、500 Internal Server Error 很常见,但原因往往和同步测试不同。
-
422:不是 schema 错了,而是 Pydantic v2 在 async context 下对Field(default_factory=...)的解析异常;检查 model 中是否用了未 await 的协程对象作默认值 -
500且 traceback 含Task was destroyed but it is pending!:说明有 background task(如BackgroundTasks)没被 await 或没被正确 cleanup,测试结束时 loop 关闭导致 task 中断 - 想看详细错误?在
AsyncClient初始化时加raise_app_exceptions=True(默认为 True,但显式写上更安心) - 数据库连接超时?不是 DB 慢,而是测试里忘了
await engine.dispose(),lifespan shutdown 没执行,下个测试复用已关闭的连接池
FastAPI 异步测试真正卡点不在语法,而在 event loop 生命周期、lifespan 执行时机、dependency_overrides 隔离这三者的咬合——漏掉任意一环,都会表现为“看起来该通的测试随机失败”。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











