es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Cómo llamar a una función asíncrona en pytest_sessionfinish()?

04 Aug, 2023 Programación

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:

  1. asyncio.get_event_loop().close() genera una advertencia: DeprecationWarning: There is no current event loop.

  2. mongo.disconnect_client() es una función asíncrona. Si cambio pytest_sessionfinish a una función asíncrona y uso await 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?

Tags:  , , , ,

Answer

  1. Avatar for davy.ai
    davy.ai

    Para resolver el primer problema, en lugar de usar asyncio.get_event_loop() en pytest_sessionfinish(), puedes pasar la variable fixture event_loop como argumento a la función y cerrarla utilizando loop.close():

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

    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:

    import pytest

@pytest.fixture(scope="session")
async def disconnect_db():
    await mongo.disconnect_client()

@pytest.mark.asyncio
async def test_something(disconnect_db):
    assert True

    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 prueba test_something, que está marcada con pytest.mark.asyncio. Esto asegurará que disconnect_db se espere antes de que finalice la prueba, resolviendo el problema de RuntimeWarning.

    Cabe destacar que también puedes usar el marcador pytest.mark.asyncio en la variable fixture initialise_db, si es necesario.

Comments are closed.