1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

はじめに

どうも、水無月せきなです。

本記事は、シリーズ「FastAPI × SQLModelで作るTodoアプリ」の最終回です。
前回までで、FastAPIとSQLModelを用いたアプリの構成と実装について紹介しました。

最終回となる今回は、以下の内容を取り上げます。

  • モックを使ったユースケース層のテスト
  • pytest-postgresql によるDB分離
  • パラメタライズの活用とconftest.pyによる共通化
  • APIエンドポイントのテストと依存関係のオーバーライド

シリーズ一覧

リポジトリ

開発環境

Windows 11 24H2
WSL2
Docker Desktop 4.43.1
Cursor 1.2.2
Python 3.12
PostgreSQL 15

モックの使用

TodoUsecaseのテストにおいて、TodoRepositoryのモックを作成しています。

app/tests/usecases/test_todo_usecase.py
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でモックするパターンが多いようですが、動的な生成ということで、これでも動くようです。

類似例:https://qiita.com/Jazuma/items/830d6778d932c1c87481#%E3%82%AA%E3%83%96%E3%82%B8%E3%82%A7%E3%82%AF%E3%83%88%E3%82%92%E3%83%A2%E3%83%83%E3%82%AF%E3%81%99%E3%82%8B

DBへの接続を伴うテスト

Todoアプリのテストということで、DBへのアクセス(CRUD処理)が伴います。
この時、以下の点が問題になります。

  • 開発用DBとテスト用DBの分離
  • テストデータの作成

今回は、ライブラリの力とpytestの仕組みで解決しました!

開発用DBとテスト用DBの分離

pytest-postgresqlは、テスト用DBの作成から削除までを行ってくれるライブラリです。

使い方(app/tests/conftest.py)
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があり、状況に応じて使い分ける必要があるようです)

後は、テスト用のセッションを実際のテストケースで使用するだけです。

テストケースでの使用(app/tests/repositories/test_todo_repository.py)
# デコレータは省略
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

参考

テストデータの作成

以下のような方法で行っています。

  1. テストデータを作成するfixtureを定義
  2. テストのparameterizeで、1. で定義したfixtureに渡したい変数をfixtureと同じ名前で宣言する(テスト関数の引数にもfixtureを入れる)
  3. parameterizeの引数でindirectfixtureの名前を指定する

コードにすると下記のようになります。

app/tests/conftest.py
@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
app/tests/repositories/test_todo_repository.py
@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
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が関わるので、テスト用のセッションを使うように細工します。

app/tests/routers/test_todo_router.py
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アプリ」はひと通り完結です。
実装の工夫点やテスト手法が、どなたかの参考になれば幸いです。

今後も機能追加や改善を行っていく予定なので、進展があればまた記事にまとめたいと思います。
その際は、またお読みいただけると嬉しいです!

参考資料

  1. ちなみに、app.dependency_overrides = {}とすると、上書きを取り消せるようです。参考

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?