带有单例的 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

从这个角度来看，您似乎正在使用一个连接到某物的全局变量。持久连接绑定到事件循环。所以如果你想继续使用一个全局变量，事件循环fixture的作用域需要和这个变量的作用域相匹配。

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

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

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