コンテンツへスキップ

FastAPIとSQLModelを使ったアプリケーションのテスト

FastAPISQLModelに関するこの章のグループを終えるにあたり、FastAPIとSQLModelを使ったアプリケーションの自動テストを実装する方法を学びましょう。✅

ヒントとコツも含まれています。🎁

FastAPIアプリケーション

前の章で構築したシンプルなFastAPIアプリケーションの1つを使用しましょう。

同じコンセプトヒントコツは、より複雑なアプリケーションにも適用できます。

ヒーローモデルを使用したアプリケーションを使用しますが、チームモデルは使用せず、セッションを取得するための依存関係を使用します。

このセッション依存関係を持つことがどれほど役立つかを今から見ていきます。✨

👀 ファイルの完全プレビュー
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)
    for key, value in hero_data.items():
        setattr(db_hero, key, value)
    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}

ファイル構造

次に、アプリケーション全体を含む1つのファイルmain.pyと、コード構造と複数ファイルの同じアイデアでテストを含む1つのファイルtest_main.pyを持つ、複数のファイルを持つPythonプロジェクトを作成します。

ファイル構造は次のとおりです。

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

FastAPIアプリケーションのテスト

FastAPIアプリケーションでテストを行ったことがない場合は、まずテストに関するFastAPIドキュメントを確認してください。

次に、ここに進むことができます。最初のステップは、依存関係のrequestspytestをインストールすることです。

同じPython環境で実行してください。

$ python -m pip install requests pytest

---> 100%

基本的なテストコード

FastAPIアプリケーションが新しいヒーローを正しく作成していることを確認するために必要な基本的なテストコードだけを使用して、簡単なテストから始めましょう。

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 👇
  1. mainモジュールからappをインポートします。

  2. FastAPIのapp用のTestClientを作成し、変数clientに入れます。

  3. 次に、このclientを使用して、APIと通信し、新しいヒーローを作成するPOST HTTP操作を送信します。

  4. 次に、レスポンスからJSONデータを取得し、変数dataに入れます。

  5. 次に、assertステートメントで結果のテストを開始し、レスポンスのステータスコードが200であることを確認します。

  6. 作成されたヒーローのname"Deadpond"であることを確認します。

  7. 作成されたヒーローのsecret_name"Dive Wilson"であることを確認します。

  8. 年齢を送信しなかったため、作成されたヒーローのageNoneであることを確認します。

  9. 作成されたヒーローにデータベースによって作成されたidがあるため、Noneではないことを確認します。

ヒント

コードの各行で何が行われるかを確認するには、番号付きのバブルを確認してください。

これは、後ですべてのテストに必要なコードのコアです。

ただし、今では、まだ注意を払っていない少しのロジスティクスと詳細に対処する必要があります。🤓

データベースのテスト

このテストは問題ないように見えますが、問題があります。

実行すると、非常に重要なヒーローを保存するために使用しているのと同じ本番データベースを使用し、不必要なデータを追加したり、さらに悪いことに、今後のテストで本番データを削除したりする可能性があります。

したがって、テスト専用の独立したテストデータベースを使用する必要があります。

これを行うには、データベースに使用されるURLを変更する必要があります。

ただし、APIのコードが実行されると、すでにエンジンに接続されているセッションを取得し、エンジンはすでに特定のデータベースURLを使用しています。

mainモジュールから変数をインポートしてテスト用にのみ値を変更しても、その時点でエンジンは元の値で既に作成されています。

ただし、すべてのAPIパスオペレーションはFastAPIの依存関係を使用してセッションを取得し、テストで依存関係をオーバーライドできます。

ここで、依存関係が非常に役立つようになります。

依存関係のオーバーライド

テスト用にget_session()依存関係をオーバーライドしましょう。

この依存関係は、すべてのパスオペレーションSQLModelセッションオブジェクトを取得するために使用されます。

テスト専用に異なるセッションオブジェクトを使用するようにオーバーライドします。

そうすることで、本番データベースを保護し、テストしているデータをより適切に制御できます。

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 👇
  1. mainモジュールからget_session依存関係をインポートします。

  2. 新しい依存関係のオーバーライドになる新しい関数を定義します。

  3. この関数は、元のget_session関数によって返されるものとは異なるセッションを返します。

    この新しいセッションオブジェクトがどのように作成されるかはまだ見ていませんが、重要なのは、これがアプリの元のセッションとは異なるセッションであるということです。

    このセッションは異なるエンジンに接続されており、その異なるエンジンは、テスト専用のデータベース用に異なるURLを使用します。

    その新しいURLも新しいエンジンもまだ定義していませんが、ここで、このオブジェクトsessionが元の依存関係get_session()によって返されるものをオーバーライドすることがすでにわかります。

  4. 次に、FastAPI appオブジェクトには、app.dependency_overridesという属性があります。

    この属性は辞書であり、キーとして元の依存関係関数を渡し、として新しいオーバーライド依存関係関数を渡すことで、依存関係のオーバーライドを入れることができます。

    したがって、ここでは、コード内のget_sessionに依存するすべての場所、つまり次のようなものを持つすべてのパラメーターで、FastAPIアプリにget_sessionの代わりにget_session_overrideを使用するように指示しています。

    session: Session = Depends(get_session)
    
  5. 依存関係のオーバーライドが完了したら、この辞書app.dependency_overrides内のすべての値を削除して、アプリケーションを通常の状態に戻すことができます。

    このように、パスオペレーション関数が依存関係を必要とするたびに、FastAPIはオーバーライドの代わりに元のものを使用します。

ヒント

コードの各行で何が行われるかを確認するには、番号付きのバブルを確認してください。

テスト用のEngineとSessionを作成する

次に、テスト中に使用されるセッションオブジェクトを作成しましょう。

独自のエンジンを使用し、この新しいエンジンはテストデータベースの新しいURLを使用します。

sqlite:///testing.db

したがって、テストデータベースはファイルtesting.dbにあります。

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)!
  1. ここで、注意すべき微妙な点があります。

    順番が重要であることを覚えておいてください。.create_all()を呼び出す前に、すべてのSQLModelモデルがすでに定義され、インポートされていることを確認する必要があります。

    この行では、.mainから何か、何かをインポートすることにより、テーブルモデルの定義を含め、.mainのコードが実行され、SQLModel.metadataに自動的に登録されます。

  2. ここで、main.pyのエンジンとは完全に異なる新しいエンジンを作成します。

    これがテストに使用するエンジンです。

    テスト用のデータベースの新しいURLを使用します

    sqlite:///testing.db
    

    また、接続引数check_same_thread=Falseを使用します。

  3. 次に、呼び出します

    SQLModel.metadata.create_all(engine)
    

    ...新しいテストデータベースにすべてのテーブルを作成することを確認するためです。

    テーブルモデルは、.mainから何かをインポートしただけで、SQLModel.metadataに登録されます。.mainのコードが実行され、テーブルモデルのクラスが作成され、自動的にSQLModel.metadataに登録されました。

    したがって、このメソッドを呼び出す時点までに、テーブルモデルはすでにそこに登録されています。💯

  4. ここでは、このテスト用のカスタムセッションオブジェクトをwithブロック内で作成します。

    これは、私たちが作成した新しいカスタムエンジンを使用するため、このセッションを使用するものはすべてテスト用データベースを使用することになります。

  5. さて、依存関係のオーバーライドに戻りますが、これは外部から同じセッションオブジェクトを返すだけで、それだけがトリックです。

  6. この時点で、テスト用のセッションwithブロックが終了し、セッションが閉じられ、ファイルが閉じられるなどします。

テーブルモデルのインポート

ここでは、テスト用データベースにすべてのテーブルを作成します。

SQLModel.metadata.create_all(engine)

ただし、順序が重要であることと、.create_all()を呼び出す前に、すべてのSQLModelモデルがすでに定義され、インポートされていることを確認する必要があることを忘れないでください。

この場合、少し注意が必要な微妙な点を除いて、すべてがうまくいきます。

.mainから何か、何でもインポートすると、テーブルモデルの定義を含め、.mainのコードが実行され、それが自動的にSQLModel.metadataに登録されます。

そうすることで、.create_all()を呼び出すと、すべてのテーブルモデルSQLModel.metadataに正しく登録され、すべてがうまくいくでしょう。👌

メモリデータベース

ここでは、本番データベースを使用していません。代わりに、testing.dbファイルを使用した新しいテスト用データベースを使用しています。これは素晴らしいことです。

ただし、SQLiteはインメモリデータベースもサポートしています。これは、データベース全体がメモリ内のみに存在し、ディスク上のファイルに保存されないことを意味します。

プログラムが終了すると、インメモリデータベースは削除されるため、本番データベースにはあまり役に立ちません。

しかし、テストには非常に適しています。各テストの前にすばやく作成し、各テストの後にすばやく削除できるためです。✅

また、ファイルに何も書き込む必要がなく、すべてがメモリ内のみであるため、通常よりもさらに高速になります。🏎

その他の代替案とアイデア👀

インメモリデータベースを使用するというアイデアに至る前に、他の代替案とアイデアを検討できたかもしれません。

1つ目は、テスト終了後にファイルを削除していないため、次のテストで残りのデータが残る可能性があるということです。そのため、テスト終了直後にファイルを削除するのが正しいことです。🔥

ただし、各テストで新しいファイルを作成し、その後削除する必要がある場合、すべてのテストを実行すると少し遅くなる可能性があります。

現在、すべてのテストで使用されるtesting.dbファイルがあります(現在は1つのテストしかありませんが、今後増える予定です)。

したがって、少しでも高速化しようとしてテストを同時に並列で実行しようとすると、同じtesting.dbファイルを使用しようとして衝突します。

もちろん、各テストデータベースファイルにランダムな名前を使用するなどして修正することもできますが、SQLiteの場合は、インメモリデータベースを使用するだけでもさらに優れた代替案があります。✨

インメモリデータベースの構成

インメモリデータベースを使用するようにコードを更新しましょう。

エンジンでいくつかのパラメーターを変更するだけで済みます。

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 👇
  1. sqlmodelからStaticPoolをインポートします。後で使用します。

  2. SQLite URLの場合は、ファイル名を何も書かずに、空のままにします。

    つまり、代わりに

    sqlite:///testing.db
    

    ...次のように書きます。

    sqlite://
    

    これは、SQLModel(実際にはSQLAlchemy)に、インメモリSQLiteデータベースを使用したいことを伝えるのに十分です。

  3. 低レベルライブラリに、check_same_thread=Falseを使用して、異なるスレッドからデータベースにアクセスできるようにしたいと伝えたことを覚えていますか?

    インメモリデータベースを使用しているので、SQLAlchemyにも、異なるスレッドから同じインメモリデータベースオブジェクトを使用できるようにしたいと伝える必要があります。

    poolclass=StaticPoolパラメーターでそれを伝えます。

    !!! info 複数のスレッドでメモリデータベースを使用するに関するSQLAlchemyドキュメントで詳細を読むことができます。

ヒント

コードの各行で何が行われるかを確認するには、番号付きのバブルを確認してください。

これで、テストはインメモリデータベースを使用して実行され、より高速で安全になる可能性があります。

そして、他のすべてのテストも同様に実行できます。

ボイラープレートコード

素晴らしい、うまくいきました。そして、各テスト関数でそのプロセスをすべて複製できます。

しかし、カスタムデータベース、メモリ内での作成、カスタムセッション、および依存関係のオーバーライドを処理するために、多くのボイラープレートコードを追加する必要がありました。

各テストでそれをすべて複製する必要がありますか?いいえ、もっとうまくできます!😎

テストを実行するためにpytestを使用しています。そして、pytestにはFastAPIの依存関係と非常に似た概念もあります。

情報

実際、pytestはFastAPIの依存関係の設計に影響を与えたものの1つでした。

これは、各テストの前に実行する必要があるコードを宣言し、テスト関数に値を提供する(これはFastAPIの依存関係とほぼ同じです)ための方法です。

実際、値を提供するためにreturnの代わりにyieldを使用することを許可する同じトリックがあり、その後pytestは、テストを含む関数の実行が完了したyieldの後のコードが実行されるようにします。

pytestでは、これらのものを依存関係ではなくフィクスチャと呼びます。

これらのフィクスチャを使用して、コードを改善し、次のテストで重複したボイラープレートを減らしましょう。

Pytestフィクスチャ

詳細については、フィクスチャに関するpytestドキュメントを参照してください。ここでは、必要なものの簡単な例を紹介します。

フィクスチャを使用した最初のコード例を見てみましょう。

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
  1. pytestをインポートします。

  2. 関数の上部で@pytest.fixture()デコレーターを使用して、これがフィクスチャ関数(FastAPIの依存関係と同等)であることをpytestに伝えます。

    また、"session"という名前を付けます。これはテスト関数で重要になります。

  3. フィクスチャ関数を作成します。これはFastAPIの依存関係関数と同等です。

    このフィクスチャでは、インメモリデータベースを使用したカスタムエンジンを作成し、テーブルを作成し、セッションを作成します。

    次に、sessionオブジェクトをyieldします。

  4. returnまたはyieldするものは、テスト関数で使用可能になります。この場合は、sessionオブジェクトです。

    ここでは、yieldを使用するため、pytestはテスト関数の実行が完了すると、この関数の「残りのコード」を実行するために戻ってきます。

    yieldの後には、目に見える「残りのコード」はありませんが、セッションを閉じるwithブロックの終わりがあります。

    yieldを使用すると、pytestは次のようになります。

    • 最初の部分を実行します。
    • セッションオブジェクトを作成します。
    • それをテスト関数に渡します。
    • テスト関数を実行します。
    • テスト関数の実行が完了すると、yieldの直後から続行し、withブロックの最後にセッションオブジェクトを正しく閉じます。
  5. さて、テスト関数で、FastAPIのように次のようなものを宣言する代わりに、このテストがフィクスチャを取得したいことをpytestに伝えるには、

    session: Session = Depends(session_fixture)
    

    ...必要なフィクスチャをpytestに伝える方法は、フィクスチャのまったく同じ名前を使用することです。

    この場合、sessionと名付けたため、動作させるには、パラメーターを正確にsessionと名付ける必要があります。

    また、エディターでオートコンプリートとインラインエラーチェックを取得できるように、型注釈session: Sessionを追加します。

  6. さて、依存関係のオーバーライド関数では、外部から来た同じsessionオブジェクトを返します。

    sessionオブジェクトは、テスト関数に渡されたパラメーターから来ており、依存関係のオーバーライドでそれを再利用して返します。

ヒント

コードの各行で何が行われるかを確認するには、番号付きのバブルを確認してください。

pytestフィクスチャはFastAPIの依存関係と非常によく似た方法で機能しますが、いくつかの小さな違いがあります。

  • pytestフィクスチャでは、上部に@pytest.fixture()のデコレーターを追加する必要があります。
  • 関数でpytestフィクスチャを使用するには、まったく同じ名前でパラメーターを宣言する必要があります。FastAPIでは、内部に実際の関数を指定してDepends()を明示的に使用する必要があります。

しかし、それらを宣言する方法と、フレームワークにそれらを関数に含めたいことを伝える方法を除けば、それらは非常によく似た方法で機能します。

これで、多くのテストを作成し、それらすべてで同じフィクスチャを再利用して、ボイラープレートコードを節約できます。

pytestは、各テスト関数の直前(および直後に終了)にそれらが実行されるようにします。したがって、各テスト関数は、実際には独自のデータベース、エンジン、およびセッションを持つことになります。

クライアントフィクスチャ

素晴らしい、そのフィクスチャは、多くの重複コードを防ぐのに役立ちます。

しかし、現在でも、他のテストで繰り返されるテスト関数にいくつかのコードを書く必要があります。現時点では、

  • 依存関係のオーバーライドを作成します。
  • それをapp.dependency_overridesに配置します。
  • TestClientを作成します。
  • リクエストを行った後で、依存関係のオーバーライドをクリアします。

それはまだ将来の他のテストで繰り返されます。改善できますか?はい!🎉

pytestフィクスチャ(FastAPIの依存関係と同じように)は、他のフィクスチャを必要とすることができます。

したがって、すべてのテストで使用され、それ自体がセッションフィクスチャを必要とするクライアントフィクスチャを作成できます。

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
  1. "client"という名前の新しいフィクスチャを作成します。

  2. このクライアントフィクスチャは、順番に、セッションフィクスチャも必要とします。

  3. 次に、クライアントフィクスチャ内に依存性オーバーライドを作成します。

  4. app.dependency_overrides辞書に依存性オーバーライドを設定します。

  5. FastAPIappを使ってTestClientを作成します。

  6. TestClientインスタンスをyieldします。

    yieldを使用することで、テスト関数が完了した後、pytestはyieldの後の残りのコードを実行するために戻ってきます。

  7. これは、yieldの後、およびテスト関数が完了した後のクリーンアップコードです。

    ここでは、FastAPIのapp内の依存性オーバーライド(ここでは1つだけ)をクリアします。

  8. 次に、テスト関数はクライアントフィクスチャを必要とします。

    そして、テスト関数の中では、コードは非常にシンプルで、TestClientを使ってAPIにリクエストを送信し、データを確認するだけです。

    フィクスチャは、すべてのセットアップクリーンアップコードを処理します。

ヒント

コードの各行で何が行われるかを確認するには、番号付きのバブルを確認してください。

これで、セッションフィクスチャを順番に使用するクライアントフィクスチャができました。

そして、実際のテスト関数では、このクライアントフィクスチャが必要であることを宣言するだけで済みます。

テストを追加する

現時点では、同じ結果を得るために、何もせずに多くの変更を行っただけのように見えるかもしれません。🤔

しかし、通常は他の多くのテスト関数を作成します。そして今では、すべてのボイラープレートと複雑さは、それらの2つのフィクスチャで一度だけ記述されます。

さらにいくつかのテストを追加しましょう。

# 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 👇
👀 ファイルの完全プレビュー
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

ヒント

通常のケースだけでなく、無効なデータエラー、およびコーナーケースが正しく処理されることもテストすることは常に良いアイデアです。

そのため、ここに2つの追加のテストを追加します。

これで、追加のテスト関数は最初のものと同じくらいシンプルにできます。データベースの設定がすべて完了したTestClientフィクスチャを取得するために、clientパラメータを宣言するだけで済みます。いいね! 😎

なぜ2つのフィクスチャなのか

ここで、コードを見て、すべてのコードを1つだけの代わりに、なぜ2つのフィクスチャに入れるのかと思うかもしれません。そして、それは完全に理にかなっています!

これらの例では、よりシンプルだったでしょう。それらのために、コードを2つのフィクスチャに分離する必要はありません...

ただし、次のテスト関数では、クライアントセッションの両方のフィクスチャが必要になります。

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 👇
👀 ファイルの完全プレビュー
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

このテスト関数では、ヒーローのリストを読み取るための*パス操作*が実際にヒーローを送信していることを確認したいと考えています。

ただし、データベースが空の場合、空のリストが返され、ヒーローデータが正しく送信されているかどうかはわかりません。

しかし、APIリクエストを送信する直前に、テストデータベースにいくつかのヒーローを作成できます。✨

そして、テストデータベースを使用しているため、テストのためにヒーローを作成しても何も影響しません。

そのためには、次のことを行う必要があります。

  • Heroモデルをインポートします。
  • クライアントセッションの両方のフィクスチャが必要です。
  • セッションを使用して、いくつかのヒーローを作成し、データベースに保存します。

その後、リクエストを送信して、データベースから実際にデータが正しく返ってきたことを確認できます。💯

ここで注目すべき重要な詳細は、他のフィクスチャテスト関数でもフィクスチャを要求できることです。

クライアントフィクスチャの関数と実際のテスト関数は両方とも同じセッションを受け取ります。

残りのテストを追加する

同じアイデアを使用し、フィクスチャを要求したり、テストに必要なデータを作成したりなどして、残りのテストを追加できます。これらは、これまでに行ってきたことと非常によく似ています。

# 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
👀 ファイルの完全プレビュー
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

テストを実行する

これで、pytestでテストを実行して結果を確認できます。

$ pytest

============= test session starts ==============
platform linux -- Python 3.7.5, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
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>

まとめ

全部読みましたか?すごい、感動しました! 😎

アプリケーションにテストを追加すると、すべてが意図したとおりに正しく機能しているという多くの確信が得られます。

そして、テストは、コードをリファクタリングしたり、変更したり、機能を追加したりするときに非常に役立ちます。なぜなら、テストはリファクタリングによって簡単に導入される可能性のある多くのエラーをキャッチするのに役立つからです。

そして、何も壊していないかどうかを確認していることがわかっているので、より速く、より効率的に作業する自信を与えてくれます。😅

テストは、あなたのコードと開発者としてのあなたを次のプロフェッショナルレベルに引き上げるものの1つだと思います。 😎

そして、あなたがこれをすべて読んで勉強した場合、私が学ぶのに何年もかかった高度なアイデアやコツをすでにたくさん知っていることになります。🚀