Chapter 13

ユニットテスト

smithonisan
smithonisan
2021.07.09に更新

最後に、これまで書いてきたコードをテストするテストコードを書いていきます。

ユニットテストは通常それ自体が仕様を表す「ドキュメント」となり得るものです。しかし、FastAPIには強力なドキュメントでもあり、リアルなデータで動作するSwagger UIがあります。

とはいえ、Swagger UIだけではコードの変更時などにすべての挙動をチェックするのは困難です。コードのリグレッションをチェックする目的で、ユニットテストを書くのはとても有意義でしょう。

テスト関連ライブラリのインストール

DB周りで非同期処理を行っているため、テストも非同期処理に対応させる必要があります。いくつかのPythonライブラリをインストールします。

本書ではPythonで有名なユニットテストフレームワークの Pytest を使っていきます。Pytestを非同期用に拡張する、 pytest-asyncio をインストールします。

DBについては、前章までのproductionコードではMySQLを使用してきました。しかし、テストのたびにMySQLにデータベースを作成・削除すると、Dockerにより環境が閉じ込められているとはいえ、オーバーヘッドが大きいと言えます。そのため、ここではファイルベースのSQLiteをベースとした、SQLiteのオンメモリモードを使用することとします。

MySQLの非同期クライアントとして aiomysql をインストールしましたが、同様にSQLiteの非同期クライアントとして aiosqlite をインストールします。

本章のユニットテストでは、定義したFastAPIの関数を直接呼ぶのではなく、HTTPインターフェイスを使い、実際のリクエストとレスポンスを検証していきます。非同期HTTPクライアントの httpx をインストールします。

docker-compose up されて demo-app が立ち上がっている状態で、以下のコマンドを実行します。

shell
$ docker-compose exec demo-app poetry add -D pytest-asyncio aiosqlite httpx

ここで、 -D はpoetryの「開発用モード」を指定するオプションです。開発用モードでは、production用のデプロイではスキップされる、テストや開発時のローカル環境のみで使用するライブラリをインストールします。これによって本番環境では不要なライブラリをインストールせずに済み、コンテナでインストールする場合も結果的にコンテナのイメージサイズを減らしたり、ビルド時間を短縮することが可能です。

上記コマンドによて各ライブラリがインストールされ、 pyproject.tomlpoetry.lock が更新されます。

pyproject.toml
[tool.poetry]
name = "demo-app"
version = "0.1.0"
description = ""
authors = ["Your Name <you@example.com>"]

[tool.poetry.dependencies]
python = "^3.9"
fastapi = "^0.65.1"
uvicorn = {extras = ["standard"], version = "^0.13.4"}
SQLAlchemy = "^1.4.20"
aiomysql = "^0.0.21"

[tool.poetry.dev-dependencies]
aiosqlite = "^0.17.0"
pytest-asyncio = "^0.15.1"
httpx = "^0.18.2"

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

上記のように、 [tool.poetry.dev-dependencies] にライブラリが新たに追加されているはずです。

DB接続とテストクライアントの準備

ユニットテスト用に、プロジェクト直下に tests ディレクトリを作成しましょう。
空ファイルの __init__.py と、テストファイル test_main.py を作成します。結果的に、以下のようなディレクトリ構成になるはずです。

(project root)
├── Dockerfile
├── docker-compose.yaml
├── poetry.lock
├── pyproject.toml
├── api
│   ├── __init__.py
│   ├── db.py
│   ├── main.py
│   ├── migrate_db.py
│   ├── cruds
│   ├── models
│   ├── routers
│   └── schemas
└── tests
    ├── __init__.py
    └── test_main.py

最初に、Pytestの フィクスチャ(fixture) を定義していきます。

フィクスチャは、テスト関数の前処理や後処理を定義することができる関数です。xUnit系のユニットテストツールで言うところの、 setup()teardown() に相当するものですが、Pythonには yield 文がありますので、これらをまとめて1つの関数として定義することができます。(ジェネレータとして定義)

テスト用にDBの接続をすべて定義する必要があるため、少々複雑になっています。以下の処理を行います。

  1. Async用のengineとsessionを作成
  2. テスト用にオンメモリのSQLiteテーブルを初期化(関数ごとにリセット)
  3. DIを使ってFastAPIのDBの向き先をテスト用DBに変更
  4. テスト用に非同期HTTPクライアントを返却
tests/test_main.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from api.db import get_db, Base
from api.main import app

ASYNC_DB_URL = "sqlite+aiosqlite:///:memory:"


@pytest.fixture
async def async_client() -> AsyncClient:
    # Async用のengineとsessionを作成
    async_engine = create_async_engine(ASYNC_DB_URL, echo=True)
    async_session = sessionmaker(
        autocommit=False, autoflush=False, bind=async_engine, class_=AsyncSession
    )

    # テスト用にオンメモリのSQLiteテーブルを初期化(関数ごとにリセット)
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    # DIを使ってFastAPIのDBの向き先をテスト用DBに変更
    async def get_test_db():
        async with async_session() as session:
            yield session

    app.dependency_overrides[get_db] = get_test_db

    # テスト用に非同期HTTPクライアントを返却
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

DBの初期化が api/migrate_db.py と少し異なっていますが、これは先に説明したとおり、同期的に処理する Engine と非同期に処理する AsyncEngine の書き方の違いによるものです。

ここで、肝になるのが 12章 DB操作(CRUDs) で説明した、 get_db のoverrideです。
ルーターは以下のように定義していました。

api/routers/tasks.py
@router.post("/tasks", response_model=task_schema.TaskCreateResponse)
async def create_task(
    task_body: task_schema.TaskCreate, db: AsyncSession = Depends(get_db)
):

この get_db 関数は通常 api/db.py からインポートされるものです。しかし、フィクスチャの中で app.dependency_overrides[get_db] = get_test_db と定義することで、上記のAPIがコールされたときに、 get_db の代わりに get_test_db を使うようにoverrideしています。これがまさに、DIの力によるものです。

テストを書く (1)

それでは、実際にテストコードを書いていきましょう。

非同期のPytest関数として、 @pytest.mark.asyncio デコレータを持つ async def で始まるコルーチンを作成します。

tests/test_main.py
import starlette.status


@pytest.mark.asyncio
async def test_create_and_read(async_client):
    response = await async_client.post("/tasks", json={"title": "テストタスク"})
    assert response.status_code == starlette.status.HTTP_200_OK
    response_obj = response.json()
    assert response_obj["title"] == "テストタスク"

    response = await async_client.get("/tasks")
    assert response.status_code == starlette.status.HTTP_200_OK
    response_obj = response.json()
    assert len(response_obj) == 1
    assert response_obj[0]["title"] == "テストタスク"
    assert response_obj[0]["done"] is False

コルーチンの引数に test_create_and_read(async_client) として先ほど定義した async_client フィクスチャを定義します。そうすると、フィクスチャの返り値が入った状態でこのコルーチンが実行されますので、 async_client.post() というようにクライアントを利用することが可能です。

このコルーチンでは、最初に POST コールによってTODOタスクを作成し、2つ目の GET コールによって作成したTODOタスクを確認しています。

それぞれ最初に json={"title": "テストタスク"} で渡したタスクが返却されていることが確認できます。

テストを書く (2)

追加で、完了フラグを使ったテストも追加してみましょう。

12章 DB操作(CRUDs) で説明したように、完了フラグのON/OFFを複数回コールした場合に正しいステータスコードが返ってくることを、シナリオとしてテストしています。

tests/test_main.py
@pytest.mark.asyncio
async def test_done_flag(async_client):
    response = await async_client.post("/tasks", json={"title": "テストタスク2"})
    assert response.status_code == starlette.status.HTTP_200_OK
    response_obj = response.json()
    assert response_obj["title"] == "テストタスク2"

    # 完了フラグを立てる
    response = await async_client.put("/tasks/1/done")
    assert response.status_code == starlette.status.HTTP_200_OK

    # 既に完了フラグが立っているので400を返却
    response = await async_client.put("/tasks/1/done")
    assert response.status_code == starlette.status.HTTP_400_BAD_REQUEST

    # 完了フラグを外す
    response = await async_client.delete("/tasks/1/done")
    assert response.status_code == starlette.status.HTTP_200_OK

    # 既に完了フラグが外れているので404を返却
    response = await async_client.delete("/tasks/1/done")
    assert response.status_code == starlette.status.HTTP_404_NOT_FOUND