FastAPIとSQLModelを使ったアプリケーションのテスト¶
FastAPIとSQLModelに関するこの章のグループを終えるにあたり、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ドキュメントを確認してください。
次に、ここに進むことができます。最初のステップは、依存関係のrequests
とpytest
をインストールすることです。
同じ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 👇
-
main
モジュールからapp
をインポートします。 -
FastAPIの
app
用のTestClient
を作成し、変数client
に入れます。 -
次に、この
client
を使用して、APIと通信し、新しいヒーローを作成するPOST
HTTP操作を送信します。 -
次に、レスポンスからJSONデータを取得し、変数
data
に入れます。 -
次に、
assert
ステートメントで結果のテストを開始し、レスポンスのステータスコードが200
であることを確認します。 -
作成されたヒーローの
name
が"Deadpond"
であることを確認します。 -
作成されたヒーローの
secret_name
が"Dive Wilson"
であることを確認します。 -
年齢を送信しなかったため、作成されたヒーローの
age
がNone
であることを確認します。 -
作成されたヒーローにデータベースによって作成された
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 👇
-
main
モジュールからget_session
依存関係をインポートします。 -
新しい依存関係のオーバーライドになる新しい関数を定義します。
-
この関数は、元の
get_session
関数によって返されるものとは異なるセッションを返します。この新しいセッションオブジェクトがどのように作成されるかはまだ見ていませんが、重要なのは、これがアプリの元のセッションとは異なるセッションであるということです。
このセッションは異なるエンジンに接続されており、その異なるエンジンは、テスト専用のデータベース用に異なるURLを使用します。
その新しいURLも新しいエンジンもまだ定義していませんが、ここで、このオブジェクト
session
が元の依存関係get_session()
によって返されるものをオーバーライドすることがすでにわかります。 -
次に、FastAPI
app
オブジェクトには、app.dependency_overrides
という属性があります。この属性は辞書であり、キーとして元の依存関係関数を渡し、値として新しいオーバーライド依存関係関数を渡すことで、依存関係のオーバーライドを入れることができます。
したがって、ここでは、コード内の
get_session
に依存するすべての場所、つまり次のようなものを持つすべてのパラメーターで、FastAPIアプリにget_session
の代わりにget_session_override
を使用するように指示しています。session: Session = Depends(get_session)
-
依存関係のオーバーライドが完了したら、この辞書
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)!
-
ここで、注意すべき微妙な点があります。
順番が重要であることを覚えておいてください。
.create_all()
を呼び出す前に、すべてのSQLModelモデルがすでに定義され、インポートされていることを確認する必要があります。この行では、
.main
から何か、何かをインポートすることにより、テーブルモデルの定義を含め、.main
のコードが実行され、SQLModel.metadata
に自動的に登録されます。 -
ここで、
main.py
のエンジンとは完全に異なる新しいエンジンを作成します。これがテストに使用するエンジンです。
テスト用のデータベースの新しいURLを使用します
sqlite:///testing.db
また、接続引数
check_same_thread=False
を使用します。 -
次に、呼び出します
SQLModel.metadata.create_all(engine)
...新しいテストデータベースにすべてのテーブルを作成することを確認するためです。
テーブルモデルは、
.main
から何かをインポートしただけで、SQLModel.metadata
に登録されます。.main
のコードが実行され、テーブルモデルのクラスが作成され、自動的にSQLModel.metadata
に登録されました。したがって、このメソッドを呼び出す時点までに、テーブルモデルはすでにそこに登録されています。💯
-
ここでは、このテスト用のカスタムセッションオブジェクトを
with
ブロック内で作成します。これは、私たちが作成した新しいカスタムエンジンを使用するため、このセッションを使用するものはすべてテスト用データベースを使用することになります。
-
さて、依存関係のオーバーライドに戻りますが、これは外部から同じセッションオブジェクトを返すだけで、それだけがトリックです。
-
この時点で、テスト用のセッションの
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 👇
-
sqlmodel
からStaticPool
をインポートします。後で使用します。 -
SQLite URLの場合は、ファイル名を何も書かずに、空のままにします。
つまり、代わりに
sqlite:///testing.db
...次のように書きます。
sqlite://
これは、SQLModel(実際にはSQLAlchemy)に、インメモリSQLiteデータベースを使用したいことを伝えるのに十分です。
-
低レベルライブラリに、
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
-
pytest
をインポートします。 -
関数の上部で
@pytest.fixture()
デコレーターを使用して、これがフィクスチャ関数(FastAPIの依存関係と同等)であることをpytestに伝えます。また、
"session"
という名前を付けます。これはテスト関数で重要になります。 -
フィクスチャ関数を作成します。これはFastAPIの依存関係関数と同等です。
このフィクスチャでは、インメモリデータベースを使用したカスタムエンジンを作成し、テーブルを作成し、セッションを作成します。
次に、
session
オブジェクトをyield
します。 -
return
またはyield
するものは、テスト関数で使用可能になります。この場合は、session
オブジェクトです。ここでは、
yield
を使用するため、pytestはテスト関数の実行が完了すると、この関数の「残りのコード」を実行するために戻ってきます。yield
の後には、目に見える「残りのコード」はありませんが、セッションを閉じるwith
ブロックの終わりがあります。yield
を使用すると、pytestは次のようになります。- 最初の部分を実行します。
- セッションオブジェクトを作成します。
- それをテスト関数に渡します。
- テスト関数を実行します。
- テスト関数の実行が完了すると、
yield
の直後から続行し、with
ブロックの最後にセッションオブジェクトを正しく閉じます。
-
さて、テスト関数で、FastAPIのように次のようなものを宣言する代わりに、このテストがフィクスチャを取得したいことをpytestに伝えるには、
session: Session = Depends(session_fixture)
...必要なフィクスチャをpytestに伝える方法は、フィクスチャのまったく同じ名前を使用することです。
この場合、
session
と名付けたため、動作させるには、パラメーターを正確にsession
と名付ける必要があります。また、エディターでオートコンプリートとインラインエラーチェックを取得できるように、型注釈
session: Session
を追加します。 -
さて、依存関係のオーバーライド関数では、外部から来た同じ
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
-
"client"
という名前の新しいフィクスチャを作成します。 -
このクライアントフィクスチャは、順番に、セッションフィクスチャも必要とします。
-
次に、クライアントフィクスチャ内に依存性オーバーライドを作成します。
-
app.dependency_overrides
辞書に依存性オーバーライドを設定します。 -
FastAPIの
app
を使ってTestClient
を作成します。 -
TestClient
インスタンスをyield
します。yield
を使用することで、テスト関数が完了した後、pytestはyield
の後の残りのコードを実行するために戻ってきます。 -
これは、
yield
の後、およびテスト関数が完了した後のクリーンアップコードです。ここでは、FastAPIの
app
内の依存性オーバーライド(ここでは1つだけ)をクリアします。 -
次に、テスト関数はクライアントフィクスチャを必要とします。
そして、テスト関数の中では、コードは非常にシンプルで、
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つだと思います。 😎
そして、あなたがこれをすべて読んで勉強した場合、私が学ぶのに何年もかかった高度なアイデアやコツをすでにたくさん知っていることになります。🚀