带有单例的 pytest-asyncio 会导致事件循环冲突

Question

我使用 dbus 进行 IPC。为了在程序的整个生命周期中只有一辆总线，我在这里使用单例。为了演示，我连接到 NetworkManager，但可以交换。此外，我正在asyncio整个项目中使用。这是该模块的一个简约工作示例，它将突出显示下面描述的问题：

import asyncio  # noqa

from dbus_next.aio import MessageBus
from dbus_next import BusType


BUS = None


async def get_bus():
    # Returns a BUS singleton
    global BUS
    if not BUS:
        BUS = await MessageBus(bus_type=BusType(2)).connect()

    return BUS


async def introspect():
    # Get the dbus singleton and call a method on that singleton
    bus = await get_bus()
    return await bus.introspect(
        'org.freedesktop.NetworkManager',
        '/org/freedesktop/NetworkManager',
    )

我正在使用pytest该pytest-asyncio插件进行测试，除了这种情况之外，它的工作原理就像魅力一样。这是一个简约的工作测试模块：

import pytest

from example import introspect


@pytest.mark.asyncio
async def test_example_first():
    # With only this first call the test passes
    await introspect()


@pytest.mark.asyncio
async def test_example_second():
    # This second call will lead to the exception below.
    await introspect()

当我执行该测试时，出现以下异常，表明事件循环已更改：

example.py:22: in introspect
    '/org/freedesktop/NetworkManager',
../.local/lib/python3.7/site-packages/dbus_next/aio/message_bus.py:133: in introspect
    return await asyncio.wait_for(future, timeout=timeout)
/usr/lib/python3.7/asyncio/tasks.py:403: in wait_for
    fut = ensure_future(fut, loop=loop)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

coro_or_future = <Future pending>

    def ensure_future(coro_or_future, *, loop=None):
        """Wrap a coroutine or an awaitable in a future.
    
        If the argument is a Future, it is returned directly.
        """
        if coroutines.iscoroutine(coro_or_future):
            if loop is None:
                loop = events.get_event_loop()
            task = loop.create_task(coro_or_future)
            if task._source_traceback:
                del task._source_traceback[-1]
            return task
        elif futures.isfuture(coro_or_future):
            if loop is not None and loop is not futures._get_loop(coro_or_future):
>               raise ValueError('loop argument must agree with Future')
E               ValueError: loop argument must agree with Future

我猜测 pytest 启动了一个事件循环，并且在模块导入期间启动了另一个事件循环，但我不确定。我尝试强制使用pytest或使用模块事件循环asyncio.set_event_loop()，但没有成功。结果还是一样。

我的假设正确吗？我如何强制使用全局事件循环？或者，我应该如何定义单例才能使其工作pytest？

可能值得注意的是，这个单例构造在程序的上下文中工作得非常好。这只是一个测试，我不知道如何让它发挥作用。

Answer 1

从这个角度来看，您似乎正在使用连接到某些内容的全局变量。持久连接绑定到事件循环。因此，如果您想继续使用全局变量，事件循环固定装置的范围需要与该变量的范围相匹配。

我会将事件循环固定装置重新定义为会话范围，并根据事件循环固定装置创建一个会话范围固定装置，以初始化此全局变量（get_bus我想，只需调用 ）。需要第二个装置来确保正确的初始化顺序 - 即首先正确设置事件循环。

编辑：文档pytest-asyncio说默认情况下事件循环是测试函数范围的。因此，每次测试都会重新创建它们。可以简单地覆盖此默认行为：

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