不能直接用 requests 测试 FastAPI 的 async endpoint，因为 requests 是同步阻塞库，无法正确 await 异步路由逻辑，且会触发事件循环未运行错误；应使用 httpx.AsyncClient + pytest-asyncio 组合。
为什么不能直接用 requests 测试 FastAPI 的 async endpoint
因为 requests 是同步阻塞的，而 FastAPI 的路由函数（如 @app.get("/")）是 async def，直接用 requests.get("http://localhost:8000/") 会触发事件循环未运行的错误，典型报错是：RuntimeError: There is no current event loop in thread 'MainThread'。更关键的是，即使你手动启动 loop，也无法正确 await 路由内部的协程逻辑（比如 await database.fetch_one(...)），测试结果不可靠。
用 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
在 conftest.py 或测试文件顶部加配置（避免每次写 @pytest.mark.asyncio）：
立即学习“Python免费学习笔记（深入）”；
pytest_plugins = ["pytest_asyncio"]
测试示例：
import pytest
from httpx import AsyncClient
from myapp.main import app # 假设你的 FastAPI 实例叫 app
<p>@pytest.mark.asyncio
async def test_read_root():
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("/")
assert response.status_code == 200
assert response.json() == {"hello": "world"}-
app=app表示走 ASGI 模式直连，不启动真实服务器，快且可控 -
base_url必须设置（哪怕只是占位），否则ac.get("/foo")会因无 host 报MissingSchema - 务必用
async with，否则 client 不会正确 cleanup，可能引发 warning 或资源泄漏
如何在测试中复用 FastAPI 的依赖（比如数据库 session）
FastAPI 的 Depends() 在测试里不会自动注入——除非你显式覆盖。常见做法是用 pytest fixture 替换生产依赖，例如把数据库 session 改为内存 SQLite 或 TestSession。
假设你的路由用了 def read_items(db: Session = Depends(get_db))，测试时这样覆盖：
from myapp.dependencies import get_db
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
<p>@pytest.fixture(scope="function")
def test_session():
engine = create_async<em>engine("sqlite+aiosqlite:///:memory:", echo=False)
AsyncSessionLocal = sessionmaker(engine, class</em>=AsyncSession, expire_on_commit=False)
return AsyncSessionLocal()</p><p>@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="<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
app.dependency_overrides.clear() # ⚠️ 必须清理，否则污染其他测试-
app.dependency_overrides是字典，赋值后必须手动.clear()，pytest 不自动重置 - 如果依赖本身是 async（比如
async def get_db()），确保你覆盖的 lambda 也返回 awaitable，或直接用lambda: get_db()（注意：它返回 coroutine，需在依赖解析时被 await） - 内存 SQLite 不支持某些 PostgreSQL 特性（如
RETURNING），复杂 SQL 场景建议用pytest-async-sqlalchemy或真实测试库
遇到 “Event loop closed” 或 “Task was destroyed but it is pending” 怎么办
这类错误通常出现在 fixture 中创建了未 await 的 task，或 client/session 没被正确关闭。最常见两个坑：
- 在 fixture 里用了
asyncio.create_task(...)但没await task或task.cancel()，导致 task 挂在 loop 关闭前 - 多个测试共用一个未隔离的全局 async resource（比如单例数据库连接池），一个测试结束时 pool 还在发请求
解决办法很直接：
- 所有 async fixture 都加上
scope="function"，避免跨测试复用 - 不用
create_task，改用await直接调用；若真需并发，用asyncio.gather(...)并确保 await 完 - 检查
AsyncClient是否都在async with块内使用——漏掉会导致底层httpcore.AsyncConnectionPool未 shutdown
真正难 debug 的点在于：这些错误不一定每轮都出现，尤其在 CI 环境下偶发。所以只要涉及 async fixture 或 background task，一律按“每个测试独占资源”来设计。
