RuntimeError：事件循环已关闭 - 电机，异步答案

【问题标题】：RuntimeError: Event loop is closed - motor, asyncioRuntimeError：事件循环已关闭 - 电机，异步
【发布时间】：2022-07-15 16:46:05
【问题描述】：

我无法运行这个测试，我总是有同样的错误 RuntimeError: 事件循环已关闭

我需要在这段代码中添加什么？

from motor.motor_asyncio import AsyncIOMotorClient
import pytest
import asyncio

client = AsyncIOMotorClient("mongodb://mongo:mongo@192.168.0.11:27017/admin?retryWrites=false")
db = client['app']
aux = db['users']

async def create_user_db(a: dict):
    x = await aux.insert_one(a)
    return x

@pytest.mark.asyncio
async def test_create():
    form = {'username': 'c3', 'password': 'c3'}
    res = await create_user_db(form)
    assert res != None

这是错误

【问题讨论】：

标签： python mongodb runtime-error motor-asyncio

【解决方案1】：

在您的示例中，您在“导入”时间打开数据库，但我们仍然没有事件循环。事件循环在测试用例运行时创建。

您可以将您的数据库定义为夹具并将其提供给测试功能，例如：

@pytest.fixture
def client():
    return AsyncIOMotorClient("mongodb://localhost:27017/")


@pytest.fixture
def db(client):
    return client['test']


@pytest.fixture
def collection(db):
    return db['test']


async def create_user_db(collection, a: dict):
    x = await collection.insert_one(a)
    return x



@pytest.mark.asyncio
async def test_create(collection):
    form = {'username': 'c3', 'password': 'c3'}
    res = await create_user_db(collection, form)
    assert res != None

【讨论】：

【解决方案2】：

为了解决这个问题，我必须像 in this answer 那样进行修补，但要修补客户端，我只为 conftest.py 中的测试实例修补 Motor AgnosticClient 类

import asyncio

from motor.core import AgnosticClient

AgnosticClient.get_io_loop = asyncio.get_running_loop

【讨论】：