はじめに
どうも、水無月せきなです。
本記事は、シリーズ「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でテストケースごとに事前・事後処理を行う