green snake
Photo by Pixabay on Pexels.com

品質を底上げするFastAPIテスト戦略:pytest・TestClient/httpx・依存差し替え・DBロールバック・モック・契約テスト・負荷試験


要約(インバーテッドピラミッド)

  • 単体・API・統合・契約・負荷というテスト層を用意し、段階的にカバーする。
  • pytestでフィクスチャを設計し、TestClienthttpx.AsyncClient を使い分ける。
  • 依存性注入(Depends)を活かし、app.dependency_overridesで簡潔に差し替える。
  • DBはトランザクションのロールバックや専用エンジンで汚染を防ぎ、外部APIはrespxなどでモックする。
  • OpenAPIを検証する契約テストと、pytest-benchmarklocustでの負荷確認で運用の安心感を高める。

誰が読んで得をするか

  • 学習者Aさん(卒研・個人開発)
    小さく始められるテストの雛形が欲しい。まずはAPIの成功・失敗の2パターンを自動化したい。
  • 小規模チームBさん(受託3名)
    仕様変更が多い。依存差し替え、DBロールバック、外部APIモックで、壊れた場所をすぐ特定したい。
  • SaaS開発Cさん(スタートアップ)
    本番トラブルが怖い。OpenAPI準拠の契約テストと、ベンチマーク・負荷試験をCIに組み込みたい。

1. テスト層の地図(まず全体像)

  • 単体テスト:関数・サービス層のロジックを高速に確認。
  • APIテスト:GET/POSTなどの入出力(成功・失敗・境界)を確認。
  • 統合テスト:DB・外部APIなど実資源を使って振る舞いを確認。
  • 契約テスト:OpenAPIに対する準拠性や例の再現性を確認。
  • 負荷・ベンチマーク:遅延やスループットの傾向を把握。

判断ポイント

  • まず単体とAPIから。次に依存差し替えで外部影響を切り離す。最後に統合・負荷を足す。

2. 最小セットアップ(pytestの基本)

pytest.ini 例:

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

依存例(requirements):

pytest
httpx
pytest-asyncio
respx            # httpx用モック
pytest-benchmark # 任意:ベンチ

プロジェクト構成(例):

app/
  main.py
  routers/
    articles.py
  core/
    settings.py
    security.py
  db/
    session.py
tests/
  conftest.py
  test_articles_api.py
  test_services_unit.py

3. アプリのテスト用起動(同期/非同期の使い分け)

  • 同期エンドポイント中心:fastapi.testclient.TestClient が手軽。
  • 非同期やWebSocket、async defの多用:httpx.AsyncClient を推奨。

サンプルAPI(対象):

# app/routers/articles.py
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel

router = APIRouter(prefix="/articles", tags=["articles"])

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

FAKE = {1: {"id": 1, "title": "hello"}}

@router.get("/{aid}", response_model=Article)
def get_article(aid: int):
    a = FAKE.get(aid)
    if not a:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="not found")
    return a

main.py

# app/main.py
from fastapi import FastAPI
from app.routers import articles

app = FastAPI(title="Testable API")
app.include_router(articles.router)

4. TestClientで同期APIを素早く検証

# tests/test_articles_api.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_get_article_ok():
    r = client.get("/articles/1")
    assert r.status_code == 200
    body = r.json()
    assert body["id"] == 1
    assert "title" in body

def test_get_article_404():
    r = client.get("/articles/999")
    assert r.status_code == 404
    assert r.json()["detail"] == "not found"

判断ポイント

  • 成功と失敗を最小ペアで書く。境界値(最小/最大/存在しない)の追加で堅牢になる。

5. httpx.AsyncClientで非同期テスト

# tests/test_async_api.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_get_article_async():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        res = await ac.get("/articles/1")
        assert res.status_code == 200

判断ポイント

  • AsyncClient(app=app)でASGI内蔵モード。外部サーバ起動なしで高速に回る。

6. 依存差し替え(Dependsの真価)

  • DBセッションや認証などの依存は app.dependency_overrides で容易に差し替えられる。
  • これにより、認証・権限・DBを切り離してAPIの振る舞いだけを検証できる。

例:認証依存を差し替える

# app/core/security.py(想定)
from fastapi import Depends, HTTPException, status

def get_current_user():
    # 実際はJWTなどで検証
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="unauthorized")

テスト側:

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.core import security

@pytest.fixture(autouse=True)
def override_auth():
    def fake_user():
        return {"sub": "test-user", "role": "admin"}
    app.dependency_overrides[security.get_current_user] = fake_user
    yield
    app.dependency_overrides.clear()

@pytest.fixture
def client():
    return TestClient(app)

判断ポイント

  • 依存差し替えはテスト終了時に必ずクリアしてリークを防ぐ。

7. DBテスト:トランザクションで汚染しない

戦略は次の3つ。

  1. インメモリDB(学習・超高速)
  2. テスト用DB+トランザクションロールバック(実運用に近い)
  3. マイグレーションを当ててから実行(Alembic)

SQLAlchemy 2.x の例:

# tests/db_utils.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.db.session import Base  # Declarative Base
from app.db.session import get_db # 依存関数

@pytest.fixture(scope="session")
def engine():
    # 本番に近づけたい場合はPostgreSQLのテストDBを使う
    eng = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
    Base.metadata.create_all(eng)
    return eng

@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    trans = connection.begin()
    Session = sessionmaker(bind=connection, autoflush=False, autocommit=False)
    session = Session()
    yield session
    session.close()
    trans.rollback()
    connection.close()

@pytest.fixture(autouse=True)
def override_db(db_session):
    from app.main import app
    def _get_db():
        try:
            yield db_session
        finally:
            pass
    app.dependency_overrides[get_db] = _get_db
    yield
    app.dependency_overrides.clear()

判断ポイント

  • テストごとにトランザクションを開始→終了時にロールバックで元通り。高速で再現性が高い。

8. 外部HTTPのモック(httpx×respx)

外部APIに依存するテストはネットワークを切り離す。

# tests/test_external_calls.py
import respx
from httpx import Response
import pytest
from httpx import AsyncClient
from app.main import app

@respx.mock
@pytest.mark.asyncio
async def test_external_call():
    respx.get("https://api.example.com/users/1").mock(
        return_value=Response(200, json={"id": 1, "name": "Alice"})
    )
    async with AsyncClient(app=app, base_url="http://test") as ac:
        r = await ac.get("/proxy/users/1")  # アプリのプロキシAPI想定
        assert r.status_code == 200
        assert r.json()["name"] == "Alice"

判断ポイント

  • 正常・失敗(5xx/タイムアウト)の双方をモックし、リトライ・フォールバックの挙動を確認。

9. 認証・認可のテスト(スコープの確認)

  • ログインAPI:成功・失敗(メール不一致、パスワード不一致、凍結ユーザー)。
  • 保護ルート:トークンなし→401、不足スコープ→403、十分なスコープ→200。

例:

def test_protected_401(client):
    res = client.get("/protected")
    assert res.status_code == 401

def test_protected_403(client, monkeypatch):
    from app.core import security
    def low_scope_user():
        return {"sub": "u", "scopes": {"articles:read"}}
    app = security  # 参照短縮
    from app.main import app as fastapi_app
    fastapi_app.dependency_overrides[security.get_current_user] = low_scope_user
    res = client.post("/articles")  # writeが必要だと仮定
    assert res.status_code == 403

判断ポイント

  • ルートごとに必要スコープを最小に。テストもスコープ差し替えで明瞭にする。

10. バリデーションと例外ハンドラ

  • Pydanticのエラー形状や、共通エラーレスポンスの整形を固定する。
  • 422 Unprocessable Entity(入力不正)とカスタム例外(業務エラー)を分けて確認。
def test_validation_422(client):
    res = client.post("/users", json={"email": "not-an-email"})
    assert res.status_code == 422
    body = res.json()
    assert "detail" in body

判断ポイント

  • エラーの形が変わるとクライアントが壊れる。スナップショットで固定しても良い。

11. OpenAPIに対する契約テスト

  • スキーマの整合性:必須・型・エンム値。
  • 例(Examples)の再現性:サンプルリクエストで期待どおりか。

簡易検証:

# tests/test_openapi_contract.py
from app.main import app

def test_openapi_basic():
    spec = app.openapi()
    assert spec["info"]["title"] == "Testable API"
    paths = spec["paths"]
    assert "/articles/{aid}" in paths

強力な手段としては、スキーマから自動生成テストを行うツールもある(参考リンク参照)。


12. プロパティベーステスト(境界を自動探索)

hypothesisを使うと、境界付近の値を自動生成して検証できる。

例(単体):

# tests/test_services_unit.py
from hypothesis import given, strategies as st

def safe_div(a: int, b: int) -> float | None:
    if b == 0:
        return None
    return a / b

@given(st.integers(), st.integers())
def test_safe_division(a, b):
    r = safe_div(a, b)
    if b == 0:
        assert r is None
    else:
        assert isinstance(r, float)

判断ポイント

  • ビジネスロジックの境界が多い箇所に効果的。API層よりサービス層で使うと良い。

13. ベンチマークと負荷試験

ベンチマーク(単体・APIの局所):

# tests/test_bench.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

@pytest.mark.benchmark(min_rounds=5)
def test_get_article_bench(benchmark):
    def do():
        assert client.get("/articles/1").status_code == 200
    benchmark(do)

負荷試験(別プロセス):

  • locustk6 を使い、シナリオ(認証→一覧→詳細→作成)を記述。
  • 計測はレイテンシ分布・エラー率・スループット(RED指標)に集約。

判断ポイント

  • CIでは短いベンチで劣化検知、本番前の検証環境で負荷試験を実施。

14. WebSocket/SSEのテスト

  • WebSocket:websockets/starlette.testclientで接続→送受信を検証。
  • SSE:httpxで接続し、ストリームからイベントをパース。

WebSocket例:

# tests/test_ws.py
from starlette.testclient import TestClient
from app.main import app

def test_ws_echo():
    with TestClient(app).websocket_connect("/realtime/ws?room=test") as ws:
        ws.send_json({"msg": "hi"})
        data = ws.receive_json()
        assert "echo" in data

判断ポイント

  • ハートビートや異常切断をシミュレーションし、接続解放の漏れを検知。

15. CIでの実行と成果物

  • テスト実行:pytest -q
  • カバレッジ:coverageで主要モジュールを可視化。
  • アーティファクト:失敗時ログ、OpenAPI JSON、スクリーンショット(必要に応じて)を保存。
  • 並列実行:pytest -n autopytest-xdist)で時短。

GitHub Actions 例(概略):

name: test
on: [push, pull_request]
jobs:
  unit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install -r requirements.txt
      - run: pip install pytest pytest-asyncio respx pytest-benchmark
      - run: pytest -q

判断ポイント

  • PRで必ずテストが走る体制に。OpenAPIエクスポートやSDK生成もここに載せられる。

16. よくある落とし穴と回避策

症状 原因 対策
テストが遅い 実DB・実外部APIを毎回呼ぶ 依存差し替え・モック、トランザクションロールバック、並列実行
不安定(フレーク) 時間依存・ランダム・順序依存 固定シード、時刻注入、I/O分離、pytest-randomly
仕様ズレ ドキュメント未更新 OpenAPI契約テスト、例(Examples)の検証
本番だけ失敗 設定差分・ネットワーク .env分離、設定注入を依存化、タイムアウトを短く
クリーンアップ漏れ セッション未解放・依存override残り yieldフィクスチャ、app.dependency_overrides.clear()

17. 導入ロードマップ

  1. TestClientで主要APIの成功・失敗を2本ずつ作る。
  2. 依存差し替えを導入し、認証やDBを切り離してスピードを確保。
  3. DBはトランザクションロールバックで汚染ゼロに。外部APIはrespxでモック。
  4. OpenAPI契約テストを追加し、例とスキーマの破綻を早期検知。
  5. ベンチと負荷試験を段階導入し、劣化を監視。CIに統合する。

参考リンク


まとめ

  • テスト層を分け、まずはAPIの成功・失敗を小さく押さえる。
  • Dependsの依存差し替え、DBロールバック、外部モックで速く壊れにくいテストを実現する。
  • OpenAPI準拠の契約テストと軽いベンチをCIに組み込めば、仕様変更や性能劣化を早期に検知できる。
  • 目的は安心して変更できること。今日から2本のテストを書き、次に依存差し替えを導入してみてほしい。

投稿者 greeden

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です

日本語が含まれない投稿は無視されますのでご注意ください。(スパム対策)