pytest运行motor异步代码报RuntimeError: Event loop is closed如何解决
错误原因
该报错是因为pytest-asyncio默认在单测执行完毕后会立即关闭当前事件循环,但AsyncIOMotorClient内部仍有未完成的异步清理任务,这些任务尝试访问已关闭的事件循环就会触发该错误。
解决方案
最规范的处理方式是通过pytest fixture管理MongoDB客户端的生命周期,在所有测试执行完成后主动关闭客户端等待内部任务清理完毕,修改后的完整代码如下:
from motor.motor_asyncio import AsyncIOMotorClient import pytest import asyncio import sys # Windows平台额外加这两行解决系统默认事件循环兼容问题 if sys.platform == "win32": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) @pytest.fixture(scope="session") def mongo_client(): client = AsyncIOMotorClient("mongodb://mongo:mongo@192.168.0.11:27017/admin?retryWrites=false") yield client # 所有测试结束后主动关闭客户端,等待内部异步资源释放 client.close() @pytest.fixture(scope="session") def db(mongo_client): return mongo_client["app"] @pytest.fixture(scope="session") def aux(db): return db["users"] async def create_user_db(collection, a: dict): x = await collection.insert_one(a) return x @pytest.mark.asyncio async def test_create(aux): form = {'username': 'c3', 'password': 'c3'} res = await create_user_db(aux, form) assert res is not None
补充说明
scope="session"表示fixture在整个测试会话中只初始化一次,避免重复创建数据库连接- 如果你不想修改原有函数的参数定义,也可以在测试文件中添加一个会话级别的fixture专门负责关闭全局的Mongo客户端,效果一致
内容的提问来源于stack exchange,提问作者Cristoff
相关产品推荐
相关产品推荐

