Using @pytest.fixture(scope="module") with @pytest.mark.asyncio

Question

I think the example below is a really common use case:

1. create a connection to a database once,
2. pass this connection around to test which insert data
3. pass the connection to a test which verifies the data.

Changing the scope of @pytest.fixture(scope="module") causes ScopeMismatch: You tried to access the 'function' scoped fixture 'event_loop' with a 'module' scoped request object, involved factories.

Also, the test_insert and test_find coroutine do not need the event_loop argument because the loop is accessible already by passing the connection.

Any ideas how to fix those two issues?

import pytest

@pytest.fixture(scope="function")  # <-- want this to be scope="module"; run once!
@pytest.mark.asyncio
async def connection(event_loop):
    """ Expensive function; want to do in the module scope. Only this function needs `event_loop`!
    """
    conn await = make_connection(event_loop)
    return conn


@pytest.mark.dependency()
@pytest.mark.asyncio
async def test_insert(connection, event_loop):  # <-- does not need event_loop arg
    """ Test insert into database.

        NB does not need event_loop argument; just the connection.
    """
    _id = 0
    success = await connection.insert(_id, "data")
    assert success == True


@pytest.mark.dependency(depends=['test_insert'])
@pytest.mark.asyncio
async def test_find(connection, event_loop):  # <-- does not need event_loop arg
    """ Test database find.

        NB does not need event_loop argument; just the connection.
    """
    _id = 0
    data = await connection.find(_id)
    assert data == "data"

Bangs

don't use the default event-loop fixture that they provide, you'll have to make your own and pass it in to your fixture. their default fixture, as stated in the docs is scoped for functions so you won't be able to use it in a module fixture like you want – Maudiemaudlin

Thanks, that put me on the right track. – Bangs

Answer 1

The solution is to redefine the event_loop fixture with the module scope. Include that in the test file.

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

Bangs

Also it is better to have scope=session. You will be able to use event_loop in other session-scoped fixtures. There are no reasons to have separate loop for each module. – Amador

That’s for pointing that out, will be useful in some cases. However here get_event_loop is returning the default loop for the thread, so it’s not a problem to have that in the module scope (or function scope for that matter!) – Bangs

Perfect. Even better to add this in the conftest.py file so that it works for any test. And as @Amador suggests, scope=session. – Padilla

I got deprecation warning with Python 3.12 and resolved it by calling asyncio.new_event_loop() instead – Religionism

The proper way to do this since v0.23.0 is to enlarge the scope of the asyncio mark of your test: pytest-asyncio.readthedocs.io/en/latest/concepts.html – Together

Answer 2

Similar ScopeMismatch issue was raised in github for pytest-asyncio (link). The solution (below) works for me:

@pytest.yield_fixture(scope='class')
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

Monogenetic

Answer 3

The event loop of pytest-asyncio was, up until version 0.23.0, running within the scope of a "function". This made it so that you had to override the actual event_loop and give it a bigger scope (see accepted answer).

However, since 0.23.0 this has been deprecated, and it now supports passing the scope of the event loop directly to the asyncio mark of the test in question:

@pytest.mark.asyncio(scope="module")
async def test_find(connection):
    pass

ref: https://pytest-asyncio.readthedocs.io/en/latest/concepts.html#asyncio-event-loops

Together

Should this also work with scope="session" and autouse="True" ? I currently have this working with the following (and getting the deprecation warning) : @pytest_asyncio.fixture(scope="session", autouse=True) async def cleanup(event_loop): yield await adb.close_connection() – Joplin

Tried to use pytest.mark.asyncio(scope="session", autouse=True), but then that fixture never gets called – Joplin

defining the event loop scope can only be done on the actual test_* functions, not the fixtures themselves – Together

Ah OK. That explains :-D Thanks ! – Joplin