3

我正在使用 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？

可能值得注意的是，这个单例结构在程序上下文中工作得非常好。这只是我无法弄清楚如何使其工作的测试。

4

1 回答 1

4

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

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

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

@pytest.fixture(scope='module')
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()
于 2020-10-09T16:34:18.310 回答