How to call an async function in pytest_sessionfinish()?

user_4564080 0
Viewed 547

I am using pytest-asyncio.

I have the following conftest.py file:

import asyncio

import pytest
from database.mongo_db import mongo


@pytest.fixture(scope="session", autouse=True)
async def initialise_db():
    await mongo.connect_client()
    await mongo.drop_db()

@pytest.fixture(scope="session")
def event_loop():
    yield asyncio.new_event_loop()

The initialise_db() function will connect to my database and clear everything from it before all of my tests are run.

Now, I want to close the event loop and also close the connection to my database once all tests are complete. I have tried adding the following function to conftest.py:

def pytest_sessionfinish(session, exitstatus):
    asyncio.get_event_loop().close()
    mongo.disconnect_client()

However, this new function has two issues:

  1. asyncio.get_event_loop().close() raises a warning: DeprecationWarning: There is no current event loop
  2. mongo.disconnect_client() is an async function. If I change pytest_sessionfinish to an async function and use await when closing the database, then I get the warning: RuntimeWarning: coroutine 'pytest_sessionfinish' was never awaited, and this is called from within pytest, so I cannot change it to be awaited unless I edit the source code. Of course, if I don't make it an async function I get the warning: RuntimeWarning: coroutine 'disconnect_client' was never awaited.

How can I resolve these two issues?

pytest-asyncio pytest python-asyncio asynchronous python
1 Answers

  1. Don't. Pytests manages the event loop and you shouldn't (and needn't) interfere with that. [EDIT: Ok, sorry. Interfere, if you have to... For example to extend the scope of the event_loop fixture.]

  2. Fixtures should take care of their own teardown. Just define it as a yield fixture:

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

@pytest.fixture(scope="session", autouse=True)
async def initialise_db():
    await mongo.connect_client()
    await mongo.drop_db()
    yield # suspended until tests are done
    await mongo.disconnect_client() # pytest calls teardown when fixture goes out of scope

The event_loopfixture is provided by default by pytest-asyncio, but can be overridden to change the scope or provide a custom loop. See pytest-asyncio readme for details.

user_144121900
Related