FastAPIテスト戦略：pytest、TestClient、依存差し替え、DB、契約テスト、負荷試験

FastAPIのテストは、すべてを同じ方法で確認するほど保守しにくくなります。

関数の計算、HTTPの入出力、データベースとの連携、API仕様、性能を別の層として扱うと、失敗の原因を絞り込みやすくなります。

まず主要APIの正常系と失敗系を小さく固定し、変更リスクに合わせて統合テストや契約テストを足すのが現実的です。

FastAPI自体の特徴や設計判断から確認したい場合は、FastAPIの設計ポイントを整理した入門記事も参照してください。

最初に決める5つのテスト層

テスト層とは、どの範囲を動かし、何が正しいと確認するかを分けたものです。

単体テストだけではHTTPやDBの接続ミスを見つけられず、統合テストだけでは失敗箇所の特定に時間がかかります。

各層に役割を持たせ、同じ振る舞いを重複して細かく検証しないことが、実行時間と保守負担を抑える基本です。

pytestで始める最小構成

最初の構成では、テスト探索の場所とファイル名を固定します。

# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
addopts = -q

同期のAPIテストにはpytestとHTTPXがあれば始められます。

非同期テスト、外部HTTPのモック、性能計測は必要になった段階で追加します。

pytest
httpx
anyio
respx             # 外部HTTPを置き換える場合
pytest-benchmark  # 局所的な性能を測る場合

説明を一貫させるため、以降は次の小さなAPIをテスト対象にします。

from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI(title="Testable API")

class Article(BaseModel):
    id: int
    title: str

def get_article_repository() -> dict[int, dict]:
    return {1: {"id": 1, "title": "hello"}}

@app.get("/articles/{article_id}", response_model=Article)
def read_article(
    article_id: int,
    repository: dict = Depends(get_article_repository),
):
    article = repository.get(article_id)
    if article is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="not found",
        )
    return article

依存性注入は、ルートが必要とするDBセッションやリポジトリを外から渡せる形にする仕組みです。

本番用の依存をテスト用に置き換えられるため、テスト対象とDBや認証を切り分けられます。

TestClientでHTTPの入出力を確認する

TestClientは、実際のHTTPサーバーを起動せず、同期関数としてリクエストを送れるテストクライアントです。

アプリ側のルートがasync defでも、テスト内で別の非同期処理を直接待つ必要がなければ利用できます。

import pytest
from fastapi.testclient import TestClient

from app.main import app

@pytest.fixture
def client():
    with TestClient(app) as test_client:
        yield test_client

def test_get_article_success(client):
    response = client.get("/articles/1")

    assert response.status_code == 200
    assert response.json() == {"id": 1, "title": "hello"}

def test_get_article_not_found(client):
    response = client.get("/articles/999")

    assert response.status_code == 404
    assert response.json()["detail"] == "not found"

正常系ではステータスだけでなく、利用側が必要とする項目と値を確認します。

失敗系では、存在しないIDや不正な入力を与え、公開済みのエラー形式を固定します。

AsyncClientへ切り替える条件

テスト関数の中で非同期DBドライバーなどを直接awaitする場合は、テスト自体を非同期にします。

現在のFastAPI公式例では、HTTPXのASGITransportへアプリを渡し、そのtransportをAsyncClientに設定します。

import pytest
from httpx import ASGITransport, AsyncClient

from app.main import app

@pytest.mark.anyio
async def test_get_article_async():
    transport = ASGITransport(app=app)
    async with AsyncClient(
        transport=transport,
        base_url="http://test",
    ) as client:
        response = await client.get("/articles/1")

    assert response.status_code == 200

TestClientとAsyncClientの違いは、アプリのルートが同期か非同期かではなく、テスト関数がほかの非同期処理を待つ必要があるかで判断します。

ASGITransportだけではアプリのlifespanイベントを起動しないため、起動時にDB接続などを準備するアプリではlifespanを含むテスト構成が必要です。

共有リソースの初期化と解放は、FastAPIのlifespan設計で詳しく整理しています。

依存関係をテストごとに差し替える

app.dependency_overridesは、元の依存関数をキー、テスト用関数を値に持つ辞書です。

外部認証や本番DBを呼ばずにルートの振る舞いを確認できますが、差し替えが次のテストへ残ると結果が順序依存になります。

import pytest
from fastapi.testclient import TestClient

from app.main import app, get_article_repository

@pytest.fixture
def isolated_client():
    test_repository = {
        10: {"id": 10, "title": "fixture article"}
    }
    app.dependency_overrides[get_article_repository] = (
        lambda: test_repository
    )
    try:
        with TestClient(app) as test_client:
            yield test_client
    finally:
        app.dependency_overrides.pop(
            get_article_repository, None
        )

def test_reads_fixture_data(isolated_client):
    response = isolated_client.get("/articles/10")

    assert response.status_code == 200
    assert response.json()["title"] == "fixture article"

個別のキーをpopすれば、同じアプリに設定された無関係な差し替えまで消さずに済みます。

認証なしの401を試すテストと、認証済み利用者へ差し替えるテストも、別のフィクスチャに分けます。

認証差し替えをautouse=Trueで全テストへ適用すると、401を確認したいテストまで認証済みになるためです。

失敗の境界を先に表にする

APIテストは、実装の行をなぞるより、利用側から見える条件と結果を表にしてから書くと漏れを見つけやすくなります。

ステータスコードだけを固定すると、必要なデータが欠けてもテストが通ります。

一方でレスポンス全体を無条件にスナップショット化すると、利用側に影響しない変更まで失敗になるため、契約として守る項目を選びます。

DBテストは接続境界まで設計する

DBテストの目的は、単にデータを保存できることではなく、本番と同じ制約やトランザクション境界で処理が成立するかを確認することです。

SQLAlchemy 2系で、テスト用Sessionを外側のトランザクションへ参加させる基本形は次のとおりです。

import pytest
from sqlalchemy.orm import Session

@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(
        bind=connection,
        join_transaction_mode="create_savepoint",
    )
    try:
        yield session
    finally:
        session.close()
        transaction.rollback()
        connection.close()

このロールバックが戻せるのは、同じ接続の外側トランザクションに参加した変更です。

アプリが別接続を開く処理やバックグラウンド処理まで自動で元に戻るわけではないため、その境界には専用DBの再作成や明示的な削除処理を用意します。

インメモリSQLiteをTestClientから共有する場合は、同じメモリDBを参照できる接続プール設定も必要です。

from sqlalchemy import create_engine
from sqlalchemy.pool import StaticPool

engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

この構成は高速な確認用であり、本番がPostgreSQLなどであれば、DB固有の制約を確かめる統合テストも残します。

外部HTTPは成功と障害を置き換える

テストダブルは、外部サービスの代わりにテストから制御する応答を返す仕組みです。

HTTPXを使う処理では、respxでURLと応答を固定し、実ネットワークや相手サービスの状態からテストを切り離せます。

import httpx
import pytest
import respx

async def fetch_user(
    client: httpx.AsyncClient, user_id: int
) -> dict:
    response = await client.get(
        f"https://api.example.com/users/{user_id}"
    )
    response.raise_for_status()
    return response.json()

@respx.mock
@pytest.mark.anyio
async def test_fetch_user_success():
    route = respx.get(
        "https://api.example.com/users/1"
    ).mock(
        return_value=httpx.Response(
            200, json={"id": 1, "name": "Alice"}
        )
    )

    async with httpx.AsyncClient() as client:
        user = await fetch_user(client, 1)

    assert route.called
    assert user["name"] == "Alice"

成功応答だけでなく、タイムアウト、接続失敗、5xxも別のテストにします。

そのときに検証するのはrespxの動作ではなく、アプリが定めた再試行回数、代替処理、利用側へ返すエラーです。

OpenAPI契約テストが守れる範囲

契約テストは、API提供側と利用側が合意したパス、入力、出力の形が変わっていないかを確認します。

次の例はOpenAPI文書に対象パスと成功応答が存在することを確かめる、最小のスモークテストです。

from app.main import app

def test_openapi_contains_article_contract():
    specification = app.openapi()
    operation = specification["paths"][
        "/articles/{article_id}"
    ]["get"]

    assert "200" in operation["responses"]

この確認だけでは、実際のレスポンスが常にスキーマへ適合することまでは保証できません。

APIテスト、OpenAPI差分の確認、実際のクライアントを使う統合テストを、変更の影響に合わせて組み合わせます。

SDKとの連携まで扱う場合は、FastAPIとOpenAPIの契約テストも参照してください。

プロパティテスト、ベンチマーク、負荷試験の違い

例を数件ずつ書いても入力範囲が広すぎる処理には、Hypothesisによるプロパティベーステストが向きます。

個々の答えを列挙するのではなく、「並べ替えても合計は同じ」「正規化を2回行っても結果は変わらない」といった処理の性質を検証します。

pytest-benchmarkは関数や単一APIの比較に使い、Locustやk6は認証、一覧、詳細、更新といった利用の流れを再現する用途に分けます。

性能値は実行環境の影響を受けるため、異なるマシンの数値をそのまま合否判定へ混ぜないようにします。

WebSocketとSSEで確認すること

WebSocketでは接続成功だけでなく、送受信の順序、認証失敗、異常切断後の解放を確認します。

from fastapi.testclient import TestClient
from app.main import app

def test_websocket_echo():
    with TestClient(app) as client:
        with client.websocket_connect(
            "/realtime/ws?room=test"
        ) as websocket:
            websocket.send_json({"message": "hello"})
            response = websocket.receive_json()

    assert response["echo"] == "hello"

SSEではContent-Type、イベント順序、再接続、途中切断時の扱いを、サービスの契約に合わせて確認します。

配信方式の選択や実装上の注意は、FastAPIのストリーミングAPI設計で比較しています。

CIでは失敗の意味が分かる単位に分ける

CIの一つのジョブへ全テストを詰め込むと、環境障害とコードの失敗を見分けにくくなります。

毎回実行する軽いテストと、環境準備が必要なテストを分けると、変更のたびに必要な確認を保ちながら待ち時間を管理できます。

FastAPIをGitHub Actionsへ組み込む流れは、FastAPIのCI/CDパイプライン入門で確認できます。

よくある失敗と切り分け方

小さく始める導入手順

1. 変更頻度と影響が大きいAPIを一つ選び、正常系と代表的な失敗系を書く。
2. DB、認証、外部HTTPを依存として分離し、必要なテストだけ差し替える。
3. 本番と同じDBが必要な境界を選び、ロールバックまたはデータ再作成を設計する。
4. OpenAPIの差分と利用側が依存するレスポンスを契約として固定する。
5. 局所的な遅延はベンチマーク、同時利用時の挙動は負荷試験で測る。

テスト本数を増やすこと自体を目標にせず、変更時に見逃したくない失敗へ対応する層を選びます。

最初の正常系と失敗系が安定したら、同じfixtureを関連APIへ広げ、CIで失敗の意味を追える状態に育ててください。

参考資料

• FastAPI公式ドキュメント：Testing
• FastAPI公式ドキュメント：Async Tests
• FastAPI公式ドキュメント：Testing Dependencies with Overrides
• HTTPX公式ドキュメント：Transports
• pytest公式ドキュメント：Fixtures
• SQLAlchemy公式ドキュメント：Session Basics
• SQLAlchemy公式ドキュメント：SQLite
• respx公式リポジトリ
• FastAPI公式ドキュメント：OpenAPI
• Locust公式サイト
• k6公式サイト