Saltar al contenido

Tests Asíncronos

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

Poder usar funciones asíncronas en tus tests podría ser útil, por ejemplo, cuando estás consultando 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ó correctamente los datos en la base de datos, mientras usas una librería de base de datos asíncrona.

Veamos cómo podemos hacer que 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 ordenado para esto, que nos permite especificar que algunas funciones de test deben ser llamadas de forma asíncrona.

HTTPX

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

El TestClient hace algo de magia por dentro 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 la usamos dentro de funciones asíncronas. Al ejecutar nuestros tests de forma asíncrona, ya no podemos usar el TestClient dentro de nuestras funciones de test.

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

Ejemplo

Para un ejemplo simple, consideremos una estructura de archivos similar a la descrita en Bigger Applications y Testing:

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

El archivo main.py tendría:

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:

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:

En Detalle

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

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

Ten en cuenta que la función de test ahora es async def en lugar de solo def como antes cuando se usaba el TestClient.

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

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 nuestras requests con el TestClient.

Consejo

Ten en cuenta que estamos usando async/await con el nuevo AsyncClient - la request es asíncrona.

Aviso

Si tu aplicación depende de lifespan events, el AsyncClient no disparará estos eventos. Para asegurar que se disparen, usa LifespanManager de florimondmanca/asgi-lifespan.

Otras Llamadas a Funciones Asíncronas

Como la función de test ahora es asíncrona, ahora también puedes llamar (y hacer await) a otras funciones async aparte 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 (ej. cuando usas MotorClient de MongoDB), recuerda instanciar objetos que necesitan un event loop solo dentro de funciones async, ej. un callback @app.on_event("startup").