pytest-asyncio与单例一起使用会导致事件循环冲突。

Question

我正在使用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进行测试，这非常有效，除了这个问题。

这是一个极简的工作测试模块：

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一起工作？

值得注意的是，在程序的上下文中，此单例构造完全正常。只是测试时我无法想出如何使其正常工作。

- grmmgrmm

Answer 1

从这个情况看，看起来您正在使用一个连接到某个东西的全局变量。持久连接绑定到事件循环。因此，如果要继续使用全局变量，则事件循环装置的范围需要匹配此变量的范围。

我将重新定义事件循环装置以使其具有会话范围，并创建一个依赖于事件循环装置的会话范围装置，以初始化此全局变量（通过仅调用get_bus，我想）。第二个装置是必需的，以确保正确的初始化顺序-即先正确设置事件循环。

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

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

- Tin Tvrtković

---

谢谢提示！我会尝试并保持这个帖子的更新！ - grmmgrmm

你有状态更新吗？我尝试了事件循环，但没有用，但我会再次检查我的变量范围。 - Marc LaBelle

救命稻草！非常感谢你。 - cmc