Python pytest怎么对FastAPI进行异步测试_使用httpx与pytest-asyncio
为什么不能直接用 requests 测试 FastAPI 的 async endpoint
这里有个常见的误区：直接用 requests 库去测试 FastAPI 的异步端点。问题出在哪儿？关键在于，requests 是一个同步阻塞的库，而 FastAPI 中那些用 async def 定义的路由函数，其内部逻辑是异步的。如果你直接调用 requests.get(“https://localhost:8000/”)，很可能会触发一个运行时错误，典型的提示是：RuntimeError: There is no current event loop in thread ‘MainThread’。这背后的核心原因在于，同步请求无法正确“等待”（await）路由内部的协程逻辑，比如一个 await database.fetch_one(...) 操作。即便你手动启动了事件循环，测试结果也往往是不可靠的，无法真实反映异步接口的行为。
用 httpx.AsyncClient + pytest-asyncio 是最轻量可靠的组合
那么，正确的姿势是什么？答案是 httpx.AsyncClient 与 pytest-asyncio 的组合。这套组合拳之所以被推崇，原因很直接：httpx.AsyncClient 原生支持异步 HTTP 调用，能够真实模拟客户端的行为；而 pytest-asyncio 提供的 @pytest.mark.asyncio 修饰器，可以自动管理事件循环的生命周期，省去了手动调用 asyncio.run() 或 loop.run_until_complete() 的麻烦。两者配合，代码既干净，调试也友好，还能完美兼容 pytest 强大的 fixture 机制。
首先，安装必要的依赖：
pip install httpx pytest-asyncio
为了省去在每个测试函数上都写 @pytest.mark.asyncio 的麻烦，可以在 conftest.py 文件或测试文件的顶部添加如下配置：
立即学习“Python免费学习笔记（深入）”；
pytest_plugins = [“pytest_asyncio”]
一个基础的测试示例如下：
import pytest
from httpx import AsyncClient
from myapp.main import app # 假设你的 FastAPI 实例叫 app
@pytest.mark.asyncio
async def test_read_root():
async with AsyncClient(app=app, base_url=“https://www.php.cn/link/9688c999c6508777280b6e8074ad82fa”) as ac:
response = await ac.get(“/”)
assert response.status_code == 200
assert response.json() == {“hello”: “world”}
- 参数
app=app表示直接使用 ASGI 模式连接应用，无需启动真实的服务器，测试速度快且环境可控。 base_url参数必须设置（哪怕只是一个占位符），否则调用ac.get(“/foo”)时会因为缺少主机信息而报MissingSchema错误。- 务必使用
async with语句来管理客户端，否则客户端不会被正确清理，可能引发警告甚至资源泄漏。
如何在测试中复用 FastAPI 的依赖（比如数据库 session）
FastAPI 中通过 Depends() 注入的依赖，在测试环境中并不会自动生效。这就需要我们显式地进行覆盖。常见的做法是利用 pytest 的 fixture 机制，将生产环境的依赖（如数据库连接）替换为测试专用的版本，例如内存 SQLite 或测试专用的 Session。
假设你的路由函数是这样的：def read_items(db: Session = Depends(get_db))。在测试时，可以这样覆盖依赖：
from myapp.dependencies import get_db
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
@pytest.fixture(scope=“function”)
def test_session():
engine = create_async_engine(“sqlite+aiosqlite:///:memory:”, echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
return AsyncSessionLocal()
@pytest.mark.asyncio
async def test_read_items(test_session):
app.dependency_overrides[get_db] = lambda: test_session
async with AsyncClient(app=app, base_url=“https://www.php.cn/link/9688c999c6508777280b6e8074ad82fa”) as ac:
response = await ac.get(“/items/”)
assert response.status_code == 200
app.dependency_overrides.clear() # ⚠️ 必须清理，否则会污染其他测试
app.dependency_overrides是一个字典，赋值后必须手动调用.clear()进行清理，pytest 不会自动重置它。- 如果被覆盖的依赖本身是异步函数（比如
async def get_db()），那么你覆盖的 lambda 函数也需要返回一个可等待对象（awaitable），或者直接使用lambda: get_db()（注意：这会返回一个协程，需要在依赖解析时被 await）。 - 内存 SQLite 可能不支持某些特定数据库（如 PostgreSQL）的特性（例如
RETURNING子句），在涉及复杂 SQL 的场景下，建议使用pytest-async-sqlalchemy这类工具，或者连接真实的测试数据库。
遇到 “Event loop closed” 或 “Task was destroyed but it is pending” 怎么办
这类错误在异步测试中并不少见，通常是因为在 fixture 中创建了未被正确等待（await）的异步任务，或者客户端、会话等资源没有被正确关闭。最常见的两个“坑”是：
- 在 fixture 中使用了
asyncio.create_task(...)但没有后续的await task或task.cancel()，导致任务在事件循环关闭前仍然挂起。 - 多个测试用例共享了一个未做隔离的全局异步资源（例如一个单例数据库连接池），当一个测试结束时，连接池可能还在处理请求。
解决办法其实很直接：
- 为所有异步 fixture 都加上
scope=“function”参数，确保每个测试函数都使用独立的资源实例，避免跨测试复用。 - 尽量避免使用
create_task，改用await直接调用协程；如果确实需要并发，使用asyncio.gather(...)并确保所有任务都被等待完成。 - 仔细检查
AsyncClient是否都在async with代码块内使用——遗漏会导致底层的httpcore.AsyncConnectionPool未能正常关闭。
真正让调试变得棘手的地方在于，这些错误可能不是每次测试都会出现，尤其是在 CI 环境中，它们往往是偶发的。所以，一个黄金法则是：只要测试涉及异步 fixture 或后台任务，一律按照“每个测试独占其资源”的原则来设计，这样可以最大程度避免不可预知的竞态条件和资源冲突。
