Python pytest怎么对FastAPI进行异步测试_使用httpx与pytest-asyncio

首页

编程语言

热心网友

转载

2026-05-06

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 或后台任务，一律按照“每个测试独占其资源”的原则来设计，这样可以最大程度避免不可预知的竞态条件和资源冲突。

来源:https://www.php.cn/faq/2314637.html

免责声明：游乐网为非赢利性网站，所展示的游戏/软件/文章内容均来自于互联网或第三方用户上传分享，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系youleyoucom@outlook.com。

上一篇：Python自动化同步多机配置文件_paramiko库远程执行同步下一篇：朴素贝叶斯在Python文本分类中效果如何_MultinomialNB处理稀疏特征