Tests Asíncronos¶

Ya has visto cómo probar tus aplicaciones de FastAPI usando el TestClient proporcionado. Hasta ahora, solo has visto cómo escribir tests sincrónicos, sin usar funciones async.

Poder usar funciones asíncronas en tus tests puede ser útil, por ejemplo, cuando consultas tu base de datos de forma asíncrona. Imagina que quieres probar el envío de requests a tu aplicación FastAPI y luego verificar que tu backend escribió exitosamente los datos correctos en la base de datos, mientras usas un paquete de base de datos asíncrono.

Veamos cómo podemos hacer que esto funcione.

pytest.mark.anyio¶

Si queremos llamar funciones asíncronas en nuestros tests, nuestras funciones de test tienen que ser asíncronas. AnyIO proporciona un plugin útil para esto, que nos permite especificar que algunas funciones de test deben ser llamadas de manera asíncrona.

HTTPX¶

Incluso si tu aplicación de FastAPI usa funciones def normales en lugar de async def, sigue siendo una aplicación async por debajo.

El TestClient hace algo de magia interna para llamar a la aplicación FastAPI asíncrona en tus funciones de test def normales, usando pytest estándar. Pero esa magia ya no funciona cuando lo usamos dentro de funciones asíncronas. Al ejecutar nuestros tests de manera asíncrona, ya no podemos usar el TestClient dentro de nuestras funciones de test.

El TestClient está basado en HTTPX, y afortunadamente, podemos usarlo directamente para probar la API.

Ejemplo¶

Para un ejemplo simple, consideremos una estructura de archivos similar a la descrita en Aplicaciones Más Grandes y Testing:


.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

El archivo main.py tendría:

Python 3.8+


from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

El archivo test_main.py tendría los tests para main.py, podría verse así ahora:

Python 3.8+


import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

Ejecútalo¶

Puedes ejecutar tus tests como de costumbre vía:


fast →pytest

restart ↻

En Detalle¶

El marcador @pytest.mark.anyio le dice a pytest que esta función de test debe ser llamada asíncronamente:

Python 3.8+


import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

Consejo

Note que la función de test ahora es async def en lugar de solo def como antes al usar el TestClient.

Luego podemos crear un AsyncClient con la app y enviar requests asíncronos a ella, usando await.

Python 3.8+


import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

Esto es equivalente a:


response = client.get('/')

...que usábamos para hacer nuestros requests con el TestClient.

Consejo

Nota que estamos usando async/await con el nuevo AsyncClient: el request es asíncrono.

Advertencia

Si tu aplicación depende de eventos de lifespan, el AsyncClient no activará estos eventos. Para asegurarte de que se activen, usa LifespanManager de florimondmanca/asgi-lifespan.

Otras Llamadas a Funciones Asíncronas¶

Al ser la función de test asíncrona, ahora también puedes llamar (y await) otras funciones async además de enviar requests a tu aplicación FastAPI en tus tests, exactamente como las llamarías en cualquier otro lugar de tu código.

Consejo

Si encuentras un RuntimeError: Task attached to a different loop al integrar llamadas a funciones asíncronas en tus tests (por ejemplo, cuando usas MotorClient de MongoDB), recuerda crear instances de objetos que necesiten un loop de eventos solo dentro de funciones async, por ejemplo, en un callback '@app.on_event("startup").