Testanwendungen mit FastAPI und SQLModel¶

Um diese Gruppe von Kapiteln über FastAPI mit SQLModel abzuschließen, lernen wir nun, wie automatisierte Tests für eine Anwendung mit FastAPI und SQLModel implementiert werden. ✅

Inklusive Tipps und Tricks. 🎁

FastAPI-Anwendung¶

Wir arbeiten mit einer der einfacheren FastAPI-Anwendungen, die wir in den vorherigen Kapiteln erstellt haben.

Alle Konzepte, Tipps und Tricks gelten auch für komplexere Anwendungen.

Wir werden die Anwendung mit den Heldenmodellen verwenden, aber ohne Teammodelle, und wir werden die Abhängigkeit nutzen, um eine Session zu erhalten.

Nun werden wir sehen, wie nützlich diese Session-Abhängigkeit ist. ✨

👀 Vorschau der vollständigen Datei

Python 3.8+

from typing import List, Optional

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


class HeroUpdate(SQLModel):
    name: Optional[str] = None
    secret_name: Optional[str] = None
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(
    *,
    session: Session = Depends(get_session),
    offset: int = 0,
    limit: int = Query(default=100, le=100),
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero


@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(
    *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

🤓 Andere Versionen und Varianten

Python 3.10+Python 3.9+

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


class HeroUpdate(SQLModel):
    name: str | None = None
    secret_name: str | None = None
    age: int | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(
    *,
    session: Session = Depends(get_session),
    offset: int = 0,
    limit: int = Query(default=100, le=100),
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero


@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(
    *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


class HeroUpdate(SQLModel):
    name: Optional[str] = None
    secret_name: Optional[str] = None
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(
    *,
    session: Session = Depends(get_session),
    offset: int = 0,
    limit: int = Query(default=100, le=100),
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero


@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(
    *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

Dateistruktur¶

Nun haben wir ein Python-Projekt mit mehreren Dateien, eine Datei main.py mit der gesamten Anwendung und eine Datei test_main.py mit den Tests, mit den gleichen Ideen aus Code-Struktur und mehrere Dateien.

Die Dateistruktur ist

.
├── project
    ├── __init__.py
    ├── main.py
    └── test_main.py

Testen von FastAPI-Anwendungen¶

Wenn Sie noch keine Tests in FastAPI-Anwendungen durchgeführt haben, lesen Sie zuerst die FastAPI-Dokumentation zum Testen.

Dann können wir hier weitermachen. Der erste Schritt ist die Installation der Abhängigkeiten, requests und pytest.

Stellen Sie sicher, dass Sie eine virtuelle Umgebung erstellen, sie aktivieren und sie dann installieren, zum Beispiel mit

$ pip install requests pytest

---> 100%

Grundlegender Testcode¶

Beginnen wir mit einem einfachen Test, nur mit dem grundlegenden Testcode, den wir benötigen, um zu überprüfen, ob die FastAPI-Anwendung korrekt einen neuen Helden erstellt.

from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine

from .main import app, get_session  # (1)!


def test_create_hero():
        # Some code here omitted, we will see it later 👈
        client = TestClient(app)  # (2)!

        response = client.post(  # (3)!
            "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
        )
        # Some code here omitted, we will see it later 👈
        data = response.json()  # (4)!

        assert response.status_code == 200  # (5)!
        assert data["name"] == "Deadpond"  # (6)!
        assert data["secret_name"] == "Dive Wilson"  # (7)!
        assert data["age"] is None  # (8)!
        assert data["id"] is not None  # (9)!

# Code below omitted 👇

Importieren Sie app aus dem Modul main.
Wir erstellen einen TestClient für die FastAPI-app und legen ihn in die Variable client.
Dann verwenden wir diesen client, um mit der API zu kommunizieren und eine POST-HTTP-Operation durchzuführen, um einen neuen Helden zu erstellen.
Dann erhalten wir die JSON-Daten aus der Antwort und legen sie in die Variable data.
Als Nächstes beginnen wir mit dem Testen der Ergebnisse mit assert-Anweisungen. Wir überprüfen, ob der Statuscode der Antwort 200 ist.
Wir überprüfen, ob der name des erstellten Helden "Deadpond" ist.
Wir überprüfen, ob der secret_name des erstellten Helden "Dive Wilson" ist.
Wir überprüfen, ob das age des erstellten Helden None ist, da wir kein Alter gesendet haben.
Wir überprüfen, ob der erstellte Held eine von der Datenbank erstellte id hat, sodass sie nicht None ist.

Tipp

Schauen Sie sich die Nummernblasen an, um zu sehen, was jede Codezeile tut.

Das ist der Kern des Codes, den wir für alle zukünftigen Tests benötigen.

Nun müssen wir uns mit ein paar logistischen Details befassen, denen wir uns bisher noch nicht gewidmet haben. 🤓

Testen der Datenbank¶

Dieser Test sieht gut aus, aber es gibt ein Problem.

Wenn wir ihn ausführen, wird dieselbe Produktionsdatenbank verwendet, die wir zum Speichern unserer sehr wichtigen Helden verwenden. Wir werden unnötige Daten hinzufügen oder schlimmer noch, in zukünftigen Tests könnten wir Produktionsdaten löschen.

Wir sollten also eine unabhängige Testdatenbank nur für die Tests verwenden.

Dazu müssen wir die verwendete URL für die Datenbank ändern.

Aber wenn der Code für die API ausgeführt wird, erhält er eine Session, die bereits mit einer Engine verbunden ist, und die Engine verwendet bereits eine bestimmte Datenbank-URL.

Selbst wenn wir die Variable aus dem main-Modul importieren und ihren Wert nur für die Tests ändern, ist die Engine zu diesem Zeitpunkt bereits mit dem ursprünglichen Wert erstellt worden.

Aber alle unsere API-Pfadoperationen erhalten die Session über eine FastAPI-Abhängigkeit, und wir können Abhängigkeiten in Tests überschreiben.

Hier erweisen sich Abhängigkeiten als sehr nützlich.

Überschreiben einer Abhängigkeit¶

Lassen Sie uns die get_session()-Abhängigkeit für die Tests überschreiben.

Diese Abhängigkeit wird von allen Pfadoperationen verwendet, um das SQLModel-Session-Objekt zu erhalten.

Wir werden sie überschreiben, um ein anderes Session-Objekt nur für die Tests zu verwenden.

So schützen wir die Produktionsdatenbank und haben mehr Kontrolle über die Daten, die wir testen.

from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine

from .main import app, get_session  # (1)!


def test_create_hero():
        # Some code here omitted, we will see it later 👈
        def get_session_override():  # (2)!
            return session  # (3)!

        app.dependency_overrides[get_session] = get_session_override  # (4)!

        client = TestClient(app)

        response = client.post(
            "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
        )
        app.dependency_overrides.clear()  # (5)!
        data = response.json()

        assert response.status_code == 200
        assert data["name"] == "Deadpond"
        assert data["secret_name"] == "Dive Wilson"
        assert data["age"] is None
        assert data["id"] is not None

# Code below omitted 👇

Importieren Sie die get_session-Abhängigkeit aus dem Modul main.
Definieren Sie die neue Funktion, die die neue Abhängigkeitsüberschreibung sein wird.
Diese Funktion gibt eine andere Session zurück als diejenige, die von der ursprünglichen get_session-Funktion zurückgegeben würde.

Wir haben noch nicht gesehen, wie dieses neue Session-Objekt erstellt wird, aber der Punkt ist, dass dies eine andere Session ist als die ursprüngliche aus der App.

Diese Session ist an eine andere Engine gebunden, und diese andere Engine verwendet eine andere URL, für eine Datenbank nur zum Testen.

Wir haben diese neue URL oder die neue Engine noch nicht definiert, aber hier sehen wir bereits, dass dieses Objekt session das von der ursprünglichen Abhängigkeit get_session() zurückgegebene Objekt überschreiben wird.
Dann hat das FastAPI-app-Objekt ein Attribut app.dependency_overrides.

Dieses Attribut ist ein Wörterbuch, und wir können Abhängigkeitsüberschreibungen darin speichern, indem wir als Schlüssel die ursprüngliche Abhängigkeitsfunktion und als Wert die neue überschreibende Abhängigkeitsfunktion übergeben.

Hier teilen wir also der FastAPI-App mit, dass sie get_session_override anstelle von get_session an allen Stellen im Code verwenden soll, die von get_session abhängen, d. h. allen Parametern mit etwas wie
```
session: Session = Depends(get_session)
```
Nachdem wir mit der Abhängigkeitsüberschreibung fertig sind, können wir die Anwendung wieder auf Normalzustand zurücksetzen, indem wir alle Werte aus diesem Wörterbuch app.dependency_overrides entfernen.

Auf diese Weise wird FastAPI die ursprüngliche Abhängigkeit verwenden, anstatt der überschriebenen, wenn eine Pfadoperation die Abhängigkeit benötigt.

Tipp

Schauen Sie sich die Nummernblasen an, um zu sehen, was jede Codezeile tut.

Erstellen der Engine und Session für Tests¶

Lassen Sie uns nun dieses Session-Objekt erstellen, das während des Testens verwendet wird.

Es wird seine eigene Engine verwenden, und diese neue Engine wird eine neue URL für die Testdatenbank verwenden.

sqlite:///testing.db

Die Testdatenbank wird also in der Datei testing.db liegen.

from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine

from .main import app, get_session  # (1)!


def test_create_hero():
    engine = create_engine(  # (2)!
        "sqlite:///testing.db", connect_args={"check_same_thread": False}
    )
    SQLModel.metadata.create_all(engine)  # (3)!

    with Session(engine) as session:  # (4)!

        def get_session_override():
            return session  # (5)!

        app.dependency_overrides[get_session] = get_session_override  # (4)!

        client = TestClient(app)

        response = client.post(
            "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
        )
        app.dependency_overrides.clear()
        data = response.json()

        assert response.status_code == 200
        assert data["name"] == "Deadpond"
        assert data["secret_name"] == "Dive Wilson"
        assert data["age"] is None
        assert data["id"] is not None
    # (6)!

Hier gibt es eine subtile Sache zu beachten.

Denken Sie daran, dass die Reihenfolge wichtig ist und wir sicherstellen müssen, dass alle SQLModel-Modelle bereits definiert und importiert sind, bevor .create_all() aufgerufen wird.

Durch den Import von etwas, *irgendetwas*, aus .main wird der Code in .main ausgeführt, einschließlich der Definition der Tabellenmodelle, und diese werden automatisch in SQLModel.metadata registriert.
Hier erstellen wir eine neue Engine, die sich vollständig von der in main.py unterscheidet.

Dies ist die Engine, die wir für die Tests verwenden werden.

Wir verwenden die neue URL der Datenbank für Tests.
```
sqlite:///testing.db
```
Und wieder verwenden wir das Verbindungsargument check_same_thread=False.
Dann rufen wir auf
```
SQLModel.metadata.create_all(engine)
```
...um sicherzustellen, dass wir alle Tabellen in der neuen Testdatenbank erstellen.

Die Tabellenmodelle sind in SQLModel.metadata registriert, nur weil wir *etwas* aus .main importiert haben und der Code in .main ausgeführt wurde, wodurch die Klassen für die Tabellenmodelle erstellt und automatisch in SQLModel.metadata registriert wurden.

Zu dem Zeitpunkt, an dem wir diese Methode aufrufen, sind die Tabellenmodelle bereits dort registriert. 💯
Hier erstellen wir das benutzerdefinierte Session-Objekt für diesen Test in einem with-Block.

Es verwendet unsere neue benutzerdefinierte Engine, sodass alles, was diese Session verwendet, die Testdatenbank verwendet.
Nun, zurück zur Abhängigkeitsüberschreibung, sie gibt einfach dasselbe Session-Objekt von außen zurück, das ist alles, das ist der ganze Trick.
Zu diesem Zeitpunkt endet der with-Block für die Test-Session, und die Session wird geschlossen, die Datei wird geschlossen usw.

Importieren von Tabellenmodellen¶

Hier erstellen wir alle Tabellen in der Testdatenbank mit

SQLModel.metadata.create_all(engine)

Aber denken Sie daran, dass die Reihenfolge wichtig ist und wir sicherstellen müssen, dass alle SQLModel-Modelle bereits definiert und importiert sind, bevor .create_all() aufgerufen wird.

In diesem Fall funktioniert alles dank einer kleinen Feinheit, die etwas Aufmerksamkeit verdient.

Da wir etwas, *irgendetwas*, aus .main importieren, wird der Code in .main ausgeführt, einschließlich der Definition der Tabellenmodelle, und diese werden automatisch in SQLModel.metadata registriert.

Auf diese Weise sind beim Aufruf von .create_all() alle Tabellenmodelle korrekt in SQLModel.metadata registriert, und alles funktioniert. 👌

In-Memory-Datenbank¶

Jetzt verwenden wir nicht die Produktionsdatenbank. Stattdessen verwenden wir eine neue Testdatenbank mit der Datei testing.db, was großartig ist.

Aber SQLite unterstützt auch die Verwendung einer In-Memory-Datenbank. Das bedeutet, dass die gesamte Datenbank nur im Speicher vorhanden ist und nie als Datei auf der Festplatte gespeichert wird.

Nach Beendigung des Programms wird die In-Memory-Datenbank gelöscht, sodass sie für eine Produktionsdatenbank nicht sehr nützlich wäre.

Aber sie eignet sich hervorragend für Tests, da sie vor jedem Test schnell erstellt und nach jedem Test schnell wieder entfernt werden kann. ✅

Da sie außerdem keine Daten in eine Datei schreiben muss und alles nur im Speicher ist, ist sie sogar noch schneller als normal. 🏎

Andere Alternativen und Ideen 👀

Bevor wir auf die Idee einer In-Memory-Datenbank kamen, hätten wir andere Alternativen und Ideen untersuchen können.

Das erste ist, dass wir die Datei nach Abschluss des Tests nicht löschen, sodass der nächste Test verbleibende Daten haben könnte. Daher wäre es richtig, die Datei direkt nach Abschluss des Tests zu löschen. 🔥

Aber wenn jeder Test eine neue Datei erstellen und sie danach löschen muss, könnte die Ausführung aller Tests etwas langsam sein.

Im Moment haben wir eine Datei testing.db, die von allen Tests verwendet wird (wir haben jetzt nur einen Test, aber wir werden mehr haben).

Wenn wir also versuchen würden, die Tests parallel auszuführen, um sie etwas zu beschleunigen, würden sie beim Versuch, dieselbe testing.db-Datei zu verwenden, kollidieren.

Natürlich könnten wir das auch beheben, indem wir einen zufälligen Namen für jede Testdatenbankdatei verwenden... aber im Fall von SQLite haben wir eine noch bessere Alternative, indem wir einfach eine In-Memory-Datenbank verwenden. ✨

Konfigurieren der In-Memory-Datenbank¶

Aktualisieren wir unseren Code, um die In-Memory-Datenbank zu verwenden.

Wir müssen nur ein paar Parameter in der Engine ändern.

from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool  # (1)!

from .main import app, get_session


def test_create_hero():
    engine = create_engine(
        "sqlite://",  # (2)!
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,  # (3)!
    )

# Code below omitted 👇

Importieren Sie StaticPool aus sqlmodel, wir werden es gleich verwenden.
Für die SQLite-URL schreiben Sie keinen Dateinamen, lassen Sie sie leer.

Anstatt also
```
sqlite:///testing.db
```
...schreiben Sie einfach
```
sqlite://
```
Dies reicht aus, um SQLModel (eigentlich SQLAlchemy) mitzuteilen, dass wir eine In-Memory-SQLite-Datenbank verwenden möchten.
Erinnern Sie sich, dass wir der Low-Level-Bibliothek, die für die Kommunikation mit SQLite zuständig ist, mit check_same_thread=False mitgeteilt haben, dass wir von verschiedenen Threads auf die Datenbank zugreifen möchten?

Da wir nun eine In-Memory-Datenbank verwenden, müssen wir SQLAlchemy auch mitteilen, dass wir dasselbe In-Memory-Datenbankobjekt von verschiedenen Threads aus verwenden möchten.

Das teilen wir ihm mit dem Parameter poolclass=StaticPool mit.

Info

Weitere Details finden Sie in der SQLAlchemy-Dokumentation zum Verwenden einer In-Memory-Datenbank in mehreren Threads.

Tipp

Schauen Sie sich die Nummernblasen an, um zu sehen, was jede Codezeile tut.

Das war's, nun läuft der Test mit der In-Memory-Datenbank, was schneller und wahrscheinlich sicherer ist.

Und alle anderen Tests können dasselbe tun.

Boilerplate-Code¶

Großartig, das funktioniert, und Sie könnten diesen gesamten Prozess in jeder Testfunktion wiederholen.

Aber wir mussten viel Boilerplate-Code hinzufügen, um die benutzerdefinierte Datenbank zu handhaben, sie im Speicher zu erstellen, die benutzerdefinierte Session und die Abhängigkeitsüberschreibung.

Müssen wir wirklich all das für jeden Test duplizieren? Nein, wir können es besser machen! 😎

Wir verwenden pytest, um die Tests auszuführen. Und pytest hat auch ein sehr ähnliches Konzept wie die Abhängigkeiten in FastAPI.

Info

Tatsächlich war pytest eines der Dinge, die das Design der Abhängigkeiten in FastAPI inspiriert haben.

Es ist eine Möglichkeit für uns, Code zu deklarieren, der vor jedem Test ausgeführt werden soll und einen Wert für die Testfunktion bereitstellt (das ist ziemlich dasselbe wie FastAPI-Abhängigkeiten).

Tatsächlich hat es auch denselben Trick, indem es erlaubt, yield anstelle von return zu verwenden, um den Wert bereitzustellen, und dann stellt pytest sicher, dass der Code nach yield ausgeführt wird, *nachdem* die Funktion mit dem Test abgeschlossen ist.

In pytest werden diese Dinge als Fixtures bezeichnet und nicht als Abhängigkeiten.

Lassen Sie uns diese Fixtures verwenden, um unseren Code zu verbessern und den doppelten Boilerplate-Code für die nächsten Tests zu reduzieren.

Pytest-Fixtures¶

Sie können mehr darüber in den pytest-Dokumenten für Fixtures lesen, aber ich gebe Ihnen ein kurzes Beispiel für das, was wir hier brauchen.

Sehen wir uns das erste Codebeispiel mit einer Fixture an.

import pytest  # (1)!
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import app, get_session


@pytest.fixture(name="session")  # (2)!
def session_fixture():  # (3)!
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session  # (4)!


def test_create_hero(session: Session):  # (5)!
    def get_session_override():
        return session  # (6)!

    app.dependency_overrides[get_session] = get_session_override

    client = TestClient(app)

    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    app.dependency_overrides.clear()
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None

Importieren Sie pytest.
Verwenden Sie den Dekorator @pytest.fixture() über der Funktion, um pytest mitzuteilen, dass dies eine Fixture-Funktion ist (entspricht einer FastAPI-Abhängigkeit).

Wir geben ihr auch den Namen "session", das wird in der Testfunktion wichtig sein.
Erstellen Sie die Fixture-Funktion. Dies entspricht einer FastAPI-Abhängigkeitsfunktion.

In dieser Fixture erstellen wir die benutzerdefinierte Engine mit der In-Memory-Datenbank, erstellen die Tabellen und erstellen die Session.

Dann yielden wir das session-Objekt.
Das, was wir returnen oder yielden, ist das, was der Testfunktion zur Verfügung steht, in diesem Fall das session-Objekt.

Hier verwenden wir yield, damit pytest zurückkommt, um "den Rest des Codes" in dieser Funktion auszuführen, sobald die Testfunktion abgeschlossen ist.

Wir haben keinen weiteren sichtbaren "Rest des Codes" nach dem yield, aber wir haben das Ende des with-Blocks, der die Session schließen wird.

Durch die Verwendung von yield wird pytest
- den ersten Teil ausführen
- das Session-Objekt erstellen
- es der Testfunktion übergeben
- die Testfunktion ausführen
- sobald die Testfunktion abgeschlossen ist, wird es hier fortgesetzt, direkt nach dem yield, und wird am Ende des with-Blocks das Session-Objekt ordnungsgemäß schließen.
Nun, in der Testfunktion, um pytest mitzuteilen, dass dieser Test die Fixture erhalten möchte, anstatt etwas wie in FastAPI zu deklarieren mit
```
session: Session = Depends(session_fixture)
```
...die Art und Weise, wie wir pytest mitteilen, welche Fixture wir wollen, ist, indem wir den exakt gleichen Namen der Fixture verwenden.

In diesem Fall haben wir sie session genannt, also muss der Parameter exakt session heißen, damit es funktioniert.

Wir fügen auch die Typannotation session: Session hinzu, damit wir Autovervollständigung und Inline-Fehlerprüfungen in unserem Editor erhalten.
Nun geben wir in der Funktion für die Abhängigkeitsüberschreibung einfach dieselbe session zurück, die von außerhalb kommt.

Das session-Objekt kommt vom Parameter, der an die Testfunktion übergeben wird, und wir verwenden es einfach wieder und geben es hier in der Abhängigkeitsüberschreibung zurück.

Tipp

Schauen Sie sich die Nummernblasen an, um zu sehen, was jede Codezeile tut.

pytest-Fixtures funktionieren sehr ähnlich wie FastAPI-Abhängigkeiten, haben aber einige geringfügige Unterschiede.

In pytest-Fixtures müssen wir einen Dekorator @pytest.fixture() oben hinzufügen.
Um eine pytest-Fixture in einer Funktion zu verwenden, müssen wir den Parameter mit dem exakt gleichen Namen deklarieren. In FastAPI müssen wir explizit Depends() mit der eigentlichen Funktion darin verwenden.

Aber abgesehen von der Art und Weise, wie wir sie deklarieren und wie wir dem Framework mitteilen, dass wir sie in der Funktion haben wollen, funktionieren sie auf sehr ähnliche Weise.

Nun erstellen wir viele Tests und verwenden diese gleiche Fixture in allen wieder, wodurch wir diesen Boilerplate-Code einsparen.

pytest wird sicherstellen, dass sie direkt vor (und nach) jeder Testfunktion ausgeführt werden. Jede Testfunktion wird also tatsächlich ihre eigene Datenbank, Engine und Session haben.

Client-Fixture¶

Großartig, diese Fixture hilft uns, viel doppelten Code zu vermeiden.

Aber im Moment müssen wir immer noch etwas Code in der Testfunktion schreiben, der für andere Tests wiederholend sein wird. Im Moment

erstellen wir die Abhängigkeitsüberschreibung.
setzen wir sie in das app.dependency_overrides.
erstellen wir den TestClient.
löschen wir die Abhängigkeitsüberschreibung(en) nach der Anfrage.

Das wird bei anderen zukünftigen Tests immer noch wiederholend sein. Können wir es verbessern? Ja! 🎉

Jede pytest-Fixture (genauso wie FastAPI-Abhängigkeiten) kann andere Fixtures anfordern.

Daher können wir eine Client-Fixture erstellen, die in allen Tests verwendet wird und die selbst die Session-Fixture anfordert.

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")  # (1)!
def client_fixture(session: Session):  # (2)!
    def get_session_override():  # (3)!
        return session

    app.dependency_overrides[get_session] = get_session_override  # (4)!

    client = TestClient(app)  # (5)!
    yield client  # (6)!
    app.dependency_overrides.clear()  # (7)!


def test_create_hero(client: TestClient):  # (8)!
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None

Erstellen Sie die neue Fixture mit dem Namen "client".
Diese Client-Fixture wiederum erfordert die Session-Fixture.
Nun erstellen wir die Abhängigkeitsüberschreibung innerhalb der Client-Fixture.
Setzen Sie die Abhängigkeitsüberschreibung in das app.dependency_overrides-Wörterbuch.
Erstellen Sie den TestClient mit der FastAPI-app.
Yielden Sie die TestClient-Instanz.

Durch die Verwendung von yield wird pytest nach Abschluss der Testfunktion zurückkommen, um den Rest des Codes nach yield auszuführen.
Das ist der Bereinigungscode, nach yield und nachdem die Testfunktion abgeschlossen ist.

Hier löschen wir die Abhängigkeitsüberschreibungen (hier ist es nur eine) in der FastAPI-app.
Nun fordert die Testfunktion die Client-Fixture an.

Und innerhalb der Testfunktion ist der Code ziemlich einfach. Wir verwenden einfach den TestClient, um Anfragen an die API zu senden, die Daten zu überprüfen und das war's.

Die Fixtures kümmern sich um den gesamten Setup- und Cleanup-Code.

Tipp

Schauen Sie sich die Nummernblasen an, um zu sehen, was jede Codezeile tut.

Nun haben wir eine Client-Fixture, die wiederum die Session-Fixture verwendet.

Und in der eigentlichen Testfunktion müssen wir nur deklarieren, dass wir diese Client-Fixture benötigen.

Weitere Tests hinzufügen¶

An diesem Punkt mag es so aussehen, als hätten wir viele Änderungen umsonst vorgenommen, um das gleiche Ergebnis zu erzielen. 🤔

Aber normalerweise werden wir viele weitere Testfunktionen erstellen. Und jetzt ist der gesamte Boilerplate-Code und die Komplexität nur einmal geschrieben, in diesen beiden Fixtures.

Fügen wir einige weitere Tests hinzu.

Python 3.8+

# Code above omitted 👆

def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.8+

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

🤓 Andere Versionen und Varianten

Python 3.10+Python 3.9+

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

Tipp

Es ist immer eine gute Idee, nicht nur den normalen Fall zu testen, sondern auch, dass ungültige Daten, Fehler und Randfälle korrekt behandelt werden.

Deshalb fügen wir hier diese beiden zusätzlichen Tests hinzu.

Jetzt können zusätzliche Testfunktionen so einfach sein wie die erste. Sie müssen nur den client-Parameter deklarieren, um die TestClient-Fixture mit allen eingerichteten Datenbankfunktionen zu erhalten. Schön! 😎

Warum zwei Fixtures?¶

Wenn wir uns den Code jetzt ansehen, könnten wir denken: Warum haben wir zwei Fixtures anstelle von nur einer mit dem gesamten Code? Und das ist völlig logisch!

Für diese Beispiele wäre das einfacher gewesen, es gibt keinen Grund, diesen Code für sie in zwei Fixtures aufzuteilen...

Aber für die nächste Testfunktion werden wir beide Fixtures benötigen, die Client- und die Session-Fixture.

Python 3.8+

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session

# Code here omitted 👈

def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.8+

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

🤓 Andere Versionen und Varianten

Python 3.10+Python 3.9+

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

In dieser Testfunktion möchten wir überprüfen, ob die Pfadoperation zum Lesen einer Liste von Helden uns tatsächlich Helden sendet.

Aber wenn die Datenbank leer ist, erhalten wir eine leere Liste, und wir wissen nicht, ob die Heldedaten korrekt gesendet werden oder nicht.

Aber wir können einige Helden erstellen in der Testdatenbank direkt bevor wir die API-Anfrage senden. ✨

Und da wir die Testdatenbank verwenden, beeinflussen wir nichts, indem wir Helden für den Test erstellen.

Dazu müssen wir

das Hero-Modell importieren
beide Fixtures anfordern, den Client und die Session.
einige Helden erstellen und sie mithilfe der Session in der Datenbank speichern.

Danach können wir die Anfrage senden und überprüfen, ob wir die Daten korrekt aus der Datenbank zurückerhalten haben. 💯

Hier ist das wichtige Detail zu beachten: Wir können Fixtures in anderen Fixtures und auch in den Testfunktionen anfordern.

Die Funktion für die Client-Fixture und die eigentliche Testfunktion erhalten beide dieselbe Session.

Hinzufügen der restlichen Tests¶

Mit denselben Ideen, dem Anfordern von Fixtures, dem Erstellen der benötigten Daten für die Tests usw., können wir nun die restlichen Tests hinzufügen. Sie sehen unseren bisherigen Arbeiten sehr ähnlich.

Python 3.8+

# Code above omitted 👆

def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

👀 Vorschau der vollständigen Datei

Python 3.8+

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

🤓 Andere Versionen und Varianten

Python 3.10+Python 3.9+

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

Tests ausführen¶

Nun können wir die Tests mit pytest ausführen und die Ergebnisse sehen.

$ pytest

============= test session starts ==============
platform linux -- Python 3.10.0, pytest-7.4.4, pluggy-1.5.0
rootdir: /home/user/code/sqlmodel-tutorial
<b>collected 7 items                              </b>

---> 100%

project/test_main.py <font color="#A6E22E">.......         [100%]</font>

<font color="#A6E22E">============== </font><font color="#A6E22E"><b>7 passed</b></font><font color="#A6E22E"> in 0.83s ===============</font>

Zusammenfassung¶

Haben Sie das alles gelesen? Wow, ich bin beeindruckt! 😎

Das Hinzufügen von Tests zu Ihrer Anwendung gibt Ihnen viel Sicherheit, dass alles korrekt funktioniert, wie Sie es beabsichtigt haben.

Und Tests sind bekanntermaßen nützlich beim Refactoring Ihres Codes, beim Ändern von Dingen, beim Hinzufügen von Funktionen. Denn Tests können viele Fehler abfangen, die sich leicht durch Refactoring einschleichen können.

Und sie geben Ihnen das Vertrauen, schneller und effizienter zu arbeiten, weil Sie wissen, dass Sie überprüfen, ob Sie nichts kaputt machen. 😅

Ich denke, Tests sind eines der Dinge, die Ihren Code und Sie als Entwickler auf das nächste professionelle Level heben. 😎

Und wenn Sie all dies gelesen und studiert haben, wissen Sie bereits viele der fortgeschrittenen Ideen und Tricks, die ich Jahre zum Lernen gebraucht habe. 🚀