pytest-asyncio with singletons causes conflicting event loops

I'm using dbus for IPC. In order to have exactly one bus throughout the livetime of my program I'm using a singleton here. For the sake of demonstration I'm connecting to NetworkManager but that could be exchanged. Furthermore, I'm using asyncio for the entire project. This is a minimalistic working example of the module that will highlight the problem described below:

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',
    )

I'm using pytest with the pytest-asyncio plugin for testing which works like charm except for this case. This is a minimalistic work test module:

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()

When I execute that test, I'm getting the following exception indicating that the event loop changed:

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

I'm guessing that pytest started an event loop and during module import another event loop was started but I'm not sure. I tried enforcing using the pytest or the module event loop using asyncio.set_event_loop() but I had no success. The result stayed the same.

Is my assumption correct? How can I either enforce using a global event loop? Alternatively, how should I define the singleton to make this work with pytest?

Probably worth noting is that this singleton construct works perfectly fine in the context of the program. It's just the test which I can't figure out how to make it work.

From looking at this, looks like you're using a global variable which connects to something. Persistent connections are bound to an event loop. So if you want to continue using a global variable, the scope of the event loop fixture needs to match the scope of this variable.

I would redefine the event loop fixture to be session scoped, and create a session scoped fixture, depending on the event loop fixture, to initialize this global variable (by just calling get_bus, I suppose). The second fixture is required to ensure proper initialization ordering - i.e. that the event loop is set up properly first.

Edit: The documentation of pytest-asyncio says that event loops by default are test function scoped. Thus, they are being recreated in for each test. This default behavior can be simply overridden:

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

Recommended topics

Hot tags