Cómo llamar a una función asíncrona en pytest_sessionfinish()?
Estoy utilizando pytest-asyncio
.
Tengo el siguiente archivo conftest.py
:
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()
La función initialise_db()
se conectará a mi base de datos y la limpiará antes de que se ejecuten todas mis pruebas.
Ahora, quiero cerrar el bucle de eventos y también cerrar la conexión a mi base de datos una vez que todas las pruebas estén completas. He intentado agregar la siguiente función a conftest.py
:
def pytest_sessionfinish(session, exitstatus):
asyncio.get_event_loop().close()
mongo.disconnect_client()
Sin embargo, esta nueva función tiene dos problemas:
asyncio.get_event_loop().close()
genera una advertencia:DeprecationWarning: There is no current event loop
.-
mongo.disconnect_client()
es una función asíncrona. Si cambiopytest_sessionfinish
a una función asíncrona y usoawait
al cerrar la base de datos, obtengo la advertencia:RuntimeWarning: coroutine 'pytest_sessionfinish' was never awaited
. Como esto se llama dentro de pytest, no puedo cambiarlo para que espere a menos que edite el código fuente. Por supuesto, si no lo convierto en una función asíncrona, obtengo la advertencia:RuntimeWarning: coroutine 'disconnect_client' was never awaited
.
¿Cómo puedo resolver estos dos problemas?
davy.ai
Para resolver el primer problema, en lugar de usar
asyncio.get_event_loop()
enpytest_sessionfinish()
, puedes pasar la variable fixtureevent_loop
como argumento a la función y cerrarla utilizandoloop.close()
:Para resolver el segundo problema, puedes usar
pytest.mark.asyncio
para crear una variable fixture asíncrona que esperará a la desconexión de la conexión de la base de datos:En el ejemplo anterior,
disconnect_db
es una variable fixture asíncrona que esperará a que se desconecte la conexión de la base de datos. Se llama en la función de pruebatest_something
, que está marcada conpytest.mark.asyncio
. Esto asegurará quedisconnect_db
se espere antes de que finalice la prueba, resolviendo el problema deRuntimeWarning
.Cabe destacar que también puedes usar el marcador
pytest.mark.asyncio
en la variable fixtureinitialise_db
, si es necesario.