带有单例的 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 和 pytest-asyncio 插件进行测试，除了这种情况外，它的工作原理与 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

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

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

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

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

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

pytest-asyncio with singletons causes conflicting event loops

python-3.x

python-asyncio

pytest-asyncio