如何在 pytest_sessionfinish() 中调用异步函数？

Question

我正在使用pytest-asyncio。

我有以下conftest.py 文件：

import asyncio

import pytest
from database.mongo_db import mongo


@pytest.fixture(scope="session", autouse=True)
async def initialise_db():
    await mongo.connect_client()
    await mongo.drop_db()

@pytest.fixture(scope="session")
def event_loop():
    yield asyncio.new_event_loop()

initialise_db() 函数将连接到我的数据库并在运行所有测试之前清除其中的所有内容。

现在，我想关闭事件循环并在所有测试完成后关闭与我的数据库的连接。我尝试将以下函数添加到conftest.py：

def pytest_sessionfinish(session, exitstatus):
    asyncio.get_event_loop().close()
    mongo.disconnect_client()

但是，这个新功能有两个问题：

1. asyncio.get_event_loop().close() 发出警告：DeprecationWarning: There is no current event loop
2. mongo.disconnect_client() 是一个异步函数。如果我将pytest_sessionfinish 更改为异步函数并在关闭数据库时使用await，则会收到警告：RuntimeWarning: coroutine 'pytest_sessionfinish' was never awaited，这是从 pytest 中调用的，因此除非我编辑源代码。当然，如果我不将其设为异步函数，我会收到警告：RuntimeWarning: coroutine 'disconnect_client' was never awaited。

如何解决这两个问题？

Answer 1

1. 不要。 Pytests 管理事件循环，您不应该（也不需要）干预它。 [编辑：好的，对不起。干涉，如果你必须...例如扩大 event_loop 固定装置的范围。]

2. 夹具应自行拆卸。只需将其定义为产量夹具：

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session", autouse=True)
async def initialise_db():
    await mongo.connect_client()
    await mongo.drop_db()
    yield # suspended until tests are done
    await mongo.disconnect_client() # pytest calls teardown when fixture goes out of scope

event_loopfixture 默认由 pytest-asyncio 提供，但可以被覆盖以更改范围或提供自定义循环。详见 pytest-asyncio readme。