はじめに
どうも、水無月せきなです。
本記事は、シリーズ「FastAPI × SQLModelで作るTodoアプリ」の最終回です。
前回までで、FastAPIとSQLModelを用いたアプリの構成と実装について紹介しました。
最終回となる今回は、以下の内容を取り上げます。
- モックを使ったユースケース層のテスト
- pytest-postgresql によるDB分離
- パラメタライズの活用とconftest.pyによる共通化
- APIエンドポイントのテストと依存関係のオーバーライド
シリーズ一覧
- 🛠️FastAPI × SQLModelで作るTodoアプリ①:開発環境とプロジェクトのセットアップ
- 🗄️FastAPI × SQLModelで作るTodoアプリ②:AlembicによるDBマイグレーション入門
- 📝FastAPI × SQLModelで作るTodoアプリ③:アーキテクチャと実装の詳細
- 🧪FastAPI × SQLModelで作るTodoアプリ④:テストの手法と実装
リポジトリ
開発環境
Windows 11 24H2
WSL2
Docker Desktop 4.43.1
Cursor 1.2.2
Python 3.12
PostgreSQL 15
モックの使用
TodoUsecaseのテストにおいて、TodoRepositoryのモックを作成しています。
todos_data: list[list[TodoRead]] = [
# 空のリスト
[],
# 単一のTodo
[TodoRead(id=1, title="test", description="test", completed=False)],
# 複数のTodo
[
TodoRead(id=1, title="task1", description="desc1", completed=False),
TodoRead(id=2, title="task2", description="desc2", completed=True),
TodoRead(id=3, title="task3", description="desc3", completed=False),
],
]
@pytest.mark.parametrize("todos_data", todos_data)
def test_get_todos_success_cases(
mocker: MockerFixture,
todos_data: list[TodoRead],
) -> None:
"""get_todos()の正常ケースをテスト"""
# モックを作成
mock_repository = mocker.Mock()
mock_repository.get_all_todos.return_value = todos_data
# モックを使用してUsecaseを作成
usecase = TodoUsecase(mock_repository)
# メソッドを実行
result = usecase.get_todos()
# 結果を検証
assert result == todos_data
# モックの呼び出しを検証
mock_repository.get_all_todos.assert_called_once()
mocker.patch
でモックするパターンが多いようですが、動的な生成ということで、これでも動くようです。
DBへの接続を伴うテスト
Todoアプリのテストということで、DBへのアクセス(CRUD処理)が伴います。
この時、以下の点が問題になります。
- 開発用DBとテスト用DBの分離
- テストデータの作成
今回は、ライブラリの力とpytestの仕組みで解決しました!
開発用DBとテスト用DBの分離
pytest-postgresqlは、テスト用DBの作成から削除までを行ってくれるライブラリです。
postgres_user = os.environ["POSTGRES_USER"]
postgres_password = os.environ["POSTGRES_PASSWORD"]
postgres_server = os.environ["POSTGRES_SERVER"]
postgres_port = os.environ["POSTGRES_PORT"]
postgres_db = os.environ["POSTGRES_DB"]
postgresql_noproc = factories.postgresql_noproc(
user=postgres_user,
password=postgres_password,
host=postgres_server,
port=postgres_port,
)
postgresql_fixture = factories.postgresql(
"postgresql_noproc",
)
@pytest.fixture
def get_test_engine(postgresql_fixture: Any) -> Engine:
"""テスト用のEngineを取得するFixture"""
# 接続URIを作成
uri = (
f"postgresql://"
f"{postgresql_fixture.info.user}:{postgresql_fixture.info.password}@{postgresql_fixture.info.host}:{postgresql_fixture.info.port}"
f"/{postgresql_fixture.info.dbname}"
)
return create_engine(uri)
@pytest.fixture
def get_test_session(get_test_engine: Engine) -> Generator[Session, None, None]:
"""テスト用のSessionを取得するFixture"""
# テーブルを作成する
SQLModel.metadata.create_all(get_test_engine)
with Session(get_test_engine) as session:
yield session
PostgresSQL自体はDockerで既に立てているので、postgresql_noproc
というfixture
を使います。
(他に二つfixture
があり、状況に応じて使い分ける必要があるようです)
後は、テスト用のセッションを実際のテストケースで使用するだけです。
# デコレータは省略
def test_get_all_todos_success_cases(
self,
get_test_session: Session,
create_test_todo_data: list[Todo],
expected_todos: list[TodoRead],
) -> None:
# リポジトリを作成
repository = TodoRepository(get_test_session)
# メソッドを実行
todos = repository.get_all_todos()
# 結果を検証
assert todos == expected_todos
参考
テストデータの作成
以下のような方法で行っています。
- テストデータを作成する
fixture
を定義 - テストの
parameterize
で、1. で定義したfixture
に渡したい変数をfixture
と同じ名前で宣言する(テスト関数の引数にもfixture
を入れる) -
parameterize
の引数でindirect
にfixture
の名前を指定する
コードにすると下記のようになります。
@pytest.fixture
def create_test_todo_data(get_test_session: Session, request: pytest.FixtureRequest) -> list[Todo]:
"""テスト用のTodoデータを作成するFixture"""
todos = [Todo.model_validate(t) for t in request.param]
for todo in todos:
get_test_session.add(todo)
get_test_session.commit()
get_test_session.refresh(todo)
return todos
@pytest.mark.parametrize(
"todo_id, create_test_todo_data, expected_todo",
[
pytest.param(
1,
[TodoCreate(title="test", description="test", completed=False)],
Todo(id=1, title="test", description="test", completed=False),
id="get_todo_in_single_todo",
),
],
indirect=["create_test_todo_data"],
)
def test_get_todo_by_id_success_cases(
self,
get_test_session: Session,
todo_id: int,
create_test_todo_data: list[Todo],
expected_todo: Todo,
) -> None:
"""_get_todo_by_id()をテスト"""
# リポジトリを作成
repository = TodoRepository(get_test_session)
# メソッドを実行
todo = repository._get_todo_by_id(todo_id)
# 結果を検証
assert todo == expected_todo
変数としてfixture
の名前で宣言していても、indirect
に指定していないとfixture
に値が渡されないので、気を付けてください。
conftest.py
ここまでDBに関して触れてきたfixture
については、conftest.py
に書いています。
内容は下記に記載しておくので、適宜ご参照ください。
app/tests/conftest.py
import os
from collections.abc import Generator
from typing import Any
import pytest
from pytest_postgresql import factories
from sqlalchemy import Engine
from sqlmodel import Session, SQLModel, create_engine
from app.models.todo import Todo
# 環境変数を取得
postgres_user = os.environ["POSTGRES_USER"]
postgres_password = os.environ["POSTGRES_PASSWORD"]
postgres_server = os.environ["POSTGRES_SERVER"]
postgres_port = os.environ["POSTGRES_PORT"]
postgres_db = os.environ["POSTGRES_DB"]
# ライブラリを使ってテスト用のDBをセットアップするためのfixtureを作成
# 既定ではdbname=testでDBが作成されテストケースの実行毎にDBは削除され1から再作成される
postgresql_noproc = factories.postgresql_noproc(
user=postgres_user,
password=postgres_password,
host=postgres_server,
port=postgres_port,
)
postgresql_fixture = factories.postgresql(
"postgresql_noproc",
)
@pytest.fixture
def get_test_engine(postgresql_fixture: Any) -> Engine:
"""テスト用のEngineを取得するFixture"""
# 接続URIを作成
uri = (
f"postgresql://"
f"{postgresql_fixture.info.user}:{postgresql_fixture.info.password}@{postgresql_fixture.info.host}:{postgresql_fixture.info.port}"
f"/{postgresql_fixture.info.dbname}"
)
return create_engine(uri)
@pytest.fixture
def get_test_session(get_test_engine: Engine) -> Generator[Session, None, None]:
"""テスト用のSessionを取得するFixture"""
# テーブルを作成する
SQLModel.metadata.create_all(get_test_engine)
with Session(get_test_engine) as session:
yield session
@pytest.fixture
def create_test_todo_data(get_test_session: Session, request: pytest.FixtureRequest) -> list[Todo]:
"""テスト用のTodoデータを作成するFixture"""
todos = [Todo.model_validate(t) for t in request.param]
for todo in todos:
get_test_session.add(todo)
get_test_session.commit()
get_test_session.refresh(todo)
return todos
APIのテスト
最後に、APIのテストにも触れましょう。
DBが関わるので、テスト用のセッションを使うように細工します。
def get_test_app(session: Session) -> FastAPI:
"""テスト用のFastAPIアプリケーションを作成"""
app = FastAPI()
app.include_router(router)
# データベースセッションの依存性を上書き
app.dependency_overrides[get_session] = lambda: session
return app
# デコレータは省略
def test_get_todos(
self,
get_test_session: Session,
create_test_todo_data: list[Todo],
expected_todo: list[TodoRead],
) -> None:
"""GET /todos エンドポイントのテスト"""
# テスト用アプリケーションを作成
app = get_test_app(get_test_session)
client = TestClient(app)
# APIを呼び出し
response = client.get("/todos")
# 結果を検証
assert response.status_code == 200
assert response.json() == jsonable_encoder(expected_todo)
app.dependency_overrides
によって、FastAPIが解決する依存関係の上書きができます1。
これにより、Depends
でDBへのセッションが注入される際に、そのセッションがテスト用のものになります。
依存関係の上書き後は、TestClient
を作成してリクエストを行うだけです。
おわりに
これで本シリーズ「FastAPI × SQLModelで作るTodoアプリ」はひと通り完結です。
実装の工夫点やテスト手法が、どなたかの参考になれば幸いです。
今後も機能追加や改善を行っていく予定なので、進展があればまた記事にまとめたいと思います。
その際は、またお読みいただけると嬉しいです!
参考資料
- FastAPIのテストをしっかり書いてみる
- [Python] pytest でモックを使う方法(pytest-mock)
- pytestを使ったPythonテストの基本と応用
- Pythonでpytestを用いた例外とエラーテストの完全ガイド
- 【Python】pytest-mockでユニットテストをモック化する
- pytest でテストケース毎に DB を自動的に初期化して、テスト開発体験を向上させる
- 【Pytest】fixtureとparametrizeでテストケースごとに事前・事後処理を行う