FastAPI应用执行Pytest时出现‘event loop closed’运行时错误

开发小型FastAPI应用时，使用pytest测试接口遇到RuntimeError: Event loop is closed错误。

数据库操作代码：

async def get_user_data(token: str, client = MongoClient()):
    """ Get user data from database for given token """
    return await client.db.user_data.find_one({"token": token})

测试代码：

def test_read_user_data(auth_token):
    response = client.post("/crud/read",
        json={"token":"xxx"},
        headers=HEADERS(auth_token)
    )
    print(response.json())
    assert response.status_code == 200

错误信息：

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

    def _check_closed(self):
        if self._closed:
>           raise RuntimeError('Event loop is closed')
E           RuntimeError: Event loop is closed

/usr/lib/python3.8/asyncio/base_events.py:508: RuntimeError
================================================== short test summary info ==================================================
FAILED test_routes.py::test_read_user_data - RuntimeError: Event loop is closed

测试集中第一个测试正常，第二个测试必现错误；调换测试顺序后其他测试也会失败，失败前已有4次数据库操作成功。已尝试转为异步测试、重写事件循环，问题依旧。

疑问：是否必须使用异步测试？如何避免测试期间事件循环关闭或解决该错误？

1. 默认参数的客户端绑定问题：get_user_data函数中，默认参数client = MongoClient()会在模块加载时创建唯一实例，绑定到当时的事件循环。第一个测试结束后事件循环被关闭，第二个测试复用该客户端时，就会触发"事件循环已关闭"的错误。
2. 同步测试与异步客户端的冲突：同步TestClient在处理异步请求时，会临时创建并关闭事件循环，复用旧客户端会导致循环不匹配。

1. 修复数据库客户端的创建逻辑 #

不要用默认参数创建客户端，改用FastAPI的依赖注入管理客户端生命周期，确保每个请求/测试都使用绑定到当前事件循环的客户端：

from fastapi import Depends
from motor.motor_asyncio import AsyncIOMotorClient

async def get_mongo_client():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    try:
        yield client
    finally:
        await client.close()

async def get_user_data(token: str, client: AsyncIOMotorClient = Depends(get_mongo_client)):
    """ Get user data from database for given token """
    return await client.db.user_data.find_one({"token": token})

2. 用异步测试匹配FastAPI架构 #

并非必须，但异步测试更适配FastAPI的异步特性，能减少事件循环冲突。使用pytest-asyncio和httpx.AsyncClient：

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_read_user_data(auth_token, async_client: AsyncClient):
    response = await async_client.post(
        "/crud/read",
        json={"token":"xxx"},
        headers=HEADERS(auth_token)
    )
    print(response.json())
    assert response.status_code == 200

3. 用Fixture管理测试资源生命周期 #

为每个测试函数创建独立的客户端实例，避免复用已绑定到关闭循环的对象：

import pytest
from motor.motor_asyncio import AsyncIOMotorClient
from httpx import AsyncClient

@pytest.fixture(scope="function")
async def mongo_client():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    yield client
    await client.close()

@pytest.fixture(scope="function")
async def async_client(mongo_client, app):
    async def override_get_mongo_client():
        yield mongo_client
    app.dependency_overrides[get_mongo_client] = override_get_mongo_client
    
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac
    
    app.dependency_overrides.clear()

• 禁止使用全局/默认参数的异步客户端，这类实例会绑定到首次创建时的事件循环，循环关闭后无法复用。
• 优先使用AsyncClient而非同步TestClient，更贴合FastAPI的异步设计，减少循环管理问题。
• 用function级别的fixture管理客户端，确保每个测试都有独立的、绑定到当前事件循环的客户端实例。

内容的提问来源于stack exchange，提问作者Dave