Asynchrone Tests¶

Sie haben bereits gesehen, wie Sie Ihre FastAPI-Anwendungen mit dem bereitgestellten TestClient testen. Bisher haben Sie nur gesehen, wie man synchrone Tests schreibt, ohne asynchrone Funktionen zu verwenden.

Die Möglichkeit, in Ihren Tests asynchrone Funktionen zu verwenden, könnte beispielsweise nützlich sein, wenn Sie Ihre Datenbank asynchron abfragen. Stellen Sie sich vor, Sie möchten das Senden von Requests an Ihre FastAPI-Anwendung testen und dann überprüfen, ob Ihr Backend die richtigen Daten erfolgreich in die Datenbank geschrieben hat, während Sie eine asynchrone Datenbankbibliothek verwenden.

Schauen wir uns an, wie wir das machen können.

pytest.mark.anyio¶

Wenn wir in unseren Tests asynchrone Funktionen aufrufen möchten, müssen unsere Testfunktionen asynchron sein. AnyIO stellt hierfür ein nettes Plugin zur Verfügung, mit dem wir festlegen können, dass einige Testfunktionen asynchron aufgerufen werden sollen.

HTTPX¶

Auch wenn Ihre FastAPI-Anwendung normale def-Funktionen anstelle von async def verwendet, handelt es sich darunter immer noch um eine asynchrone Anwendung.

Der TestClient macht unter der Haube magisches, um die asynchrone FastAPI-Anwendung in Ihren normalen def-Testfunktionen, mithilfe von Standard-Pytest aufzurufen. Aber diese Magie funktioniert nicht mehr, wenn wir sie in asynchronen Funktionen verwenden. Durch die asynchrone Ausführung unserer Tests können wir den TestClient nicht mehr in unseren Testfunktionen verwenden.

Der TestClient basiert auf HTTPX und glücklicherweise können wir ihn direkt verwenden, um die API zu testen.

Beispiel¶

Betrachten wir als einfaches Beispiel eine Dateistruktur ähnlich der in Größere Anwendungen und Testen:

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

Die Datei main.py hätte als Inhalt:

from fastapi import FastAPI

app = FastAPI()


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

Die Datei test_main.py hätte die Tests für main.py, das könnte jetzt so aussehen:

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app

transport = ASGITransport(app=app)


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

Es ausführen¶

Sie können Ihre Tests wie gewohnt ausführen mit:

$ pytest

---> 100%

Details¶

Der Marker @pytest.mark.anyio teilt pytest mit, dass diese Testfunktion asynchron aufgerufen werden soll:

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app

transport = ASGITransport(app=app)


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

Tipp

Beachten Sie, dass die Testfunktion jetzt async def ist und nicht nur def wie zuvor, wenn Sie den TestClient verwenden.

Dann können wir einen AsyncClient mit der App erstellen und mit await asynchrone Requests an ihn senden.

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app

transport = ASGITransport(app=app)


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

Das ist das Äquivalent zu:

response = client.get('/')

... welches wir verwendet haben, um unsere Requests mit dem TestClient zu machen.

Tipp

Beachten Sie, dass wir async/await mit dem neuen AsyncClient verwenden – der Request ist asynchron.

Achtung

Falls Ihre Anwendung auf Lifespan-Events angewiesen ist, der AsyncClient löst diese Events nicht aus. Um sicherzustellen, dass sie ausgelöst werden, verwenden Sie LifespanManager von florimondmanca/asgi-lifespan.

Andere asynchrone Funktionsaufrufe¶

Da die Testfunktion jetzt asynchron ist, können Sie in Ihren Tests neben dem Senden von Requests an Ihre FastAPI-Anwendung jetzt auch andere asynchrone Funktionen aufrufen (und awaiten), genau so, wie Sie diese an anderer Stelle in Ihrem Code aufrufen würden.

Tipp

Wenn Sie einen RuntimeError: Task attached to a different loop erhalten, wenn Sie asynchrone Funktionsaufrufe in Ihre Tests integrieren (z. B. bei Verwendung von MongoDBs MotorClient), dann denken Sie daran, Objekte zu instanziieren, die einen Event Loop nur innerhalb asynchroner Funktionen benötigen, z. B. einen @app.on_event("startup")-Callback.