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

Question

我正在为IPC使用dbus。为了在我的程序的整个生命周期中只有一条总线，我在这里使用了单例。为了演示，我连接到了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-asyncio插件的pytest进行测试，除了这种情况外，它的工作方式类似于charm。这是一个简约的工作测试模块：

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

从这里可以看出，您使用的是一个全局变量，该变量连接到某个对象。持久连接被绑定到事件循环。因此，如果您希望继续使用全局变量，事件循环fixture的作用域需要与此变量的作用域匹配。

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

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

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