pytest-patterns

Pytest のパターン

使用するタイミング

以下の場合にこのスキルを有効にしてください。

• サービスクラスやリポジトリークラスの単体テストを作成するとき
• httpx.AsyncClient を使用して FastAPI エンドポイントの統合テストを作成するとき
• pytest フィクスチャや conftest ファイルを作成またはリファクタリングするとき
• テストデータ用の factory_boy ファクトリーを設定するとき
• pytest-asyncio を使用して非同期コードをテストするとき
• 外部サービス（HTTP API、メール、キュー）をモックするとき
• 入力のバリエーションに対するパラメーター化テストを追加するとき
• テストカバレッジを監査または改善するとき

以下の場合はこのスキルを使用しないでください。

• フロントエンドの React コンポーネントやフックのテスト（react-testing-patterns を使用）
• Playwright による E2E ブラウザーテスト（e2e-testing を使用）
• TDD のレッド・グリーン・リファクターのワークフローの徹底（tdd-workflow を使用）
• アプリケーションコードの作成（python-backend-expert を使用）

手順

テストの構成

tests/
├── conftest.py              # Root conftest: DB session, async client, auth helpers
├── unit/
│   ├── conftest.py          # Unit-specific fixtures (mocked repos, services)
│   ├── services/
│   │   ├── test_user_service.py
│   │   └── test_order_service.py
│   └── repositories/
│       └── test_user_repository.py
├── integration/
│   ├── conftest.py          # Integration-specific fixtures (test DB, seeding)
│   ├── test_users_api.py
│   └── test_orders_api.py
└── factories/
    ├── __init__.py
    ├── user_factory.py
    └── order_factory.py

命名規則：

• テストファイル：test_<module>.py
• テストクラス：Test<Feature>（関連するテストをまとめ、__init__ は定義しない）
• テスト関数：test_<action>_<expected_outcome> または test_<scenario>
• フィクスチャ：内容を表す名詞（db_session、authenticated_client、sample_user）

マーカーの規則：

# pyproject.toml
[tool.pytest.ini_options]
markers = [
    "unit: Unit tests (no DB, no network)",
    "integration: Integration tests (real DB, real HTTP)",
    "slow: Tests that take > 1 second",
]
asyncio_mode = "auto"

テストの一部を実行するには、pytest -m unit、pytest -m integration、pytest -m "not slow" を使用します。

フィクスチャの設計

Conftest の階層

フィクスチャはカスケードします。ルートの conftest.py で共有フィクスチャを定義し、サブディレクトリーの conftest ファイルで階層固有のフィクスチャを追加します。

ルートの conftest（tests/conftest.py）：

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from app.main import app
from app.database import get_db

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture(scope="session")
async def engine():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest.fixture
async def db_session(engine):
    async with async_sessionmaker(engine, class_=AsyncSession)() as session:
        yield session
        await session.rollback()

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session
    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

フィクスチャのスコープ

スコープ	用途	例
function（デフォルト）	テストごとに分離されたデータ	db_session、sample_user
class	テストクラス内で共有	service_instance
module	テストファイル内で共有	seeded_database
session	テスト実行全体で共有	engine、anyio_backend

ルール：

• データの分離のため、デフォルトでは function スコープを使用します
• session スコープは、コストの高いステートレスなリソース（エンジン、イベントループ）に限って使用します
• 変更可能なデータに session スコープを使用してはいけません。テスト同士が干渉します
• yield するフィクスチャでは、ロールバック、削除、クローズなどのクリーンアップを行います

認証フィクスチャ

@pytest.fixture
def auth_headers():
    """Return authorization headers for a standard test user."""
    token = create_test_token(user_id=1, role="member")
    return {"Authorization": f"Bearer {token}"}

@pytest.fixture
async def authenticated_client(client, auth_headers):
    """AsyncClient pre-configured with auth headers."""
    client.headers.update(auth_headers)
    return client

@pytest.fixture
def admin_headers():
    """Return authorization headers for an admin user."""
    token = create_test_token(user_id=99, role="admin")
    return {"Authorization": f"Bearer {token}"}

ファクトリーパターン

一貫性があり、上書き可能なテストデータを用意するには factory_boy を使用します。

import factory
from app.models import User, Order

class UserFactory(factory.Factory):
    class Meta:
        model = User

    id = factory.Sequence(lambda n: n + 1)
    email = factory.LazyAttribute(lambda o: f"user{o.id}@example.com")
    display_name = factory.Faker("name")
    role = "member"
    is_active = True

class OrderFactory(factory.Factory):
    class Meta:
        model = Order

    id = factory.Sequence(lambda n: n + 1)
    user_id = factory.LazyAttribute(lambda o: UserFactory().id)
    total_cents = factory.Faker("random_int", min=100, max=100000)
    status = "pending"

テストでの使用例：

def test_user_defaults():
    user = UserFactory()
    assert user.is_active is True
    assert user.role == "member"

def test_user_override():
    admin = UserFactory(role="admin", display_name="Admin User")
    assert admin.role == "admin"

def test_user_batch():
    users = UserFactory.build_batch(5)
    assert len(users) == 5

SQLAlchemy との統合（DB にデータを永続化する統合テスト向け）：

class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = None  # Set per-test via conftest

    # ... fields same as above

conftest でセッションを設定します。

@pytest.fixture(autouse=True)
def set_factory_session(db_session):
    UserFactory._meta.sqlalchemy_session = db_session
    OrderFactory._meta.sqlalchemy_session = db_session

API 統合テスト

テスト用データベースを使用しつつ、実際のアプリに対して httpx.AsyncClient で FastAPI エンドポイントをテストします。

import pytest
from httpx import AsyncClient

class TestUsersAPI:
    """Integration tests for /api/v1/users endpoints."""

    async def test_create_user_success(self, authenticated_client: AsyncClient):
        response = await authenticated_client.post("/api/v1/users", json={
            "email": "[email protected]",
            "display_name": "New User",
        })
        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "[email protected]"
        assert "id" in data

    async def test_create_user_duplicate_email(self, authenticated_client, sample_user):
        response = await authenticated_client.post("/api/v1/users", json={
            "email": sample_user.email,
            "display_name": "Duplicate",
        })
        assert response.status_code == 409
        assert "already exists" in response.json()["detail"]

    async def test_list_users_pagination(self, authenticated_client):
        response = await authenticated_client.get("/api/v1/users?limit=10&cursor=0")
        assert response.status_code == 200
        data = response.json()
        assert "items" in data
        assert "next_cursor" in data

    async def test_get_user_not_found(self, authenticated_client):
        response = await authenticated_client.get("/api/v1/users/99999")
        assert response.status_code == 404

    async def test_unauthenticated_request(self, client):
        response = await client.get("/api/v1/users")
        assert response.status_code == 401

主なパターン：

• 保護されたエンドポイントには authenticated_client を使用し、認証テストには通常の client を使用します
• まずステータスコードをアサートし、次にレスポンスボディをアサートします
• エラーケース（404、409、422、401、403）をテストします
• ページネーションのパラメーターをテストします
• 正確なタイムスタンプや自動生成 ID をアサートしてはいけません（"id" in data を使用します）

非同期テスト

pyproject.toml で asyncio_mode = "auto" を設定すると、すべての async def test_* 関数が自動的に実行されます。

async def test_async_service_call(db_session):
    service = UserService(db_session)
    user = await service.create_user(email="[email protected]", display_name="Test")
    assert user.id is not None

async def test_concurrent_operations(db_session):
    service = UserService(db_session)
    import asyncio
    results = await asyncio.gather(
        service.get_user(1),
        service.get_user(2),
        service.get_user(3),
    )
    assert len(results) == 3

よくある落とし穴：

• sync フィクスチャと async フィクスチャを不用意に混在させないでください。非同期フィクスチャは非同期テストでのみ使用できます
• 一貫性を保つため、常に anyio を直接使わず pytest-asyncio を使用します
• テストがハングする場合は、await されていないコルーチンや async キーワードの不足を確認します

モック戦略

外部サービスはモックする：はい

from unittest.mock import AsyncMock, patch

async def test_send_notification(db_session):
    with patch("app.services.notification.EmailClient") as mock_email:
        mock_email.return_value.send = AsyncMock(return_value=True)
        service = NotificationService(db_session)
        result = await service.notify_user(user_id=1, message="Hello")
        assert result is True
        mock_email.return_value.send.assert_called_once()

データベースはモックしない：いいえ

# BAD: Mocking the database hides real query issues
async def test_user_service(mock_db):
    mock_db.execute.return_value = MockResult([user_dict])  # Don't do this

# GOOD: Use a real test database (SQLite or PostgreSQL in Docker)
async def test_user_service(db_session):
    service = UserService(db_session)
    user = await service.create_user(email="[email protected]", display_name="Test")
    fetched = await service.get_user(user.id)
    assert fetched.email == "[email protected]"

モックする対象：

モックする	モックしない
HTTP API（respx または unittest.mock を使用）	データベースクエリー
メール／SMS サービス	SQLAlchemy セッション
ファイルストレージ（S3、GCS）	リポジトリーメソッド（統合テスト内）
メッセージキュー（Redis、RabbitMQ）	Pydantic による検証
時刻／datetime（freezegun）	FastAPI の依存性注入
乱数／UUID の生成	ORM のリレーションシップ

HTTP モックでの respx の使用：

import respx
from httpx import Response

@respx.mock
async def test_external_api_call():
    respx.get("https://api.example.com/data").mock(
        return_value=Response(200, json={"key": "value"})
    )
    service = ExternalDataService()
    result = await service.fetch_data()
    assert result["key"] == "value"

パラメーター化テスト

複数の入力と出力の組み合わせをテストするには @pytest.mark.parametrize を使用します。

@pytest.mark.parametrize("email,is_valid", [
    ("[email protected]", True),
    ("[email protected]", True),
    ("[email protected]", True),
    ("", False),
    ("not-an-email", False),
    ("@missing-local.com", False),
    ("user@", False),
])
def test_email_validation(email, is_valid):
    if is_valid:
        assert validate_email(email) is True
    else:
        with pytest.raises(ValidationError):
            validate_email(email)

読みやすい出力にするため、ID を指定してパラメーター化します：

@pytest.mark.parametrize("status,expected_code", [
    pytest.param("active", 200, id="active-user-ok"),
    pytest.param("suspended", 403, id="suspended-user-forbidden"),
    pytest.param("deleted", 404, id="deleted-user-not-found"),
])
async def test_user_access_by_status(authenticated_client, status, expected_code):
    ...

複数のフィクスチャをパラメーター化します：

@pytest.mark.parametrize("role,can_delete", [
    ("admin", True),
    ("member", False),
    ("viewer", False),
])
async def test_delete_permission(client, role, can_delete):
    headers = {"Authorization": f"Bearer {create_test_token(role=role)}"}
    response = await client.delete("/api/v1/users/1", headers=headers)
    if can_delete:
        assert response.status_code == 204
    else:
        assert response.status_code == 403

カバレッジ要件

最小しきい値：

• 全体：行カバレッジ 80%
• サービス層：90%（重要なビジネスロジック）
• リポジトリ層：70%（単純な CRUD）
• ルート：80%（すべての成功パスと主要なエラーパス）

pyproject.toml の設定：

[tool.coverage.run]
source = ["app"]
omit = ["app/migrations/*", "app/main.py", "app/__init__.py"]

[tool.coverage.report]
fail_under = 80
show_missing = true
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "if __name__ ==",
    "@overload",
]

カバレッジの実行：

pytest --cov=app --cov-report=term-missing --cov-report=html --cov-fail-under=80

check-test-coverage.sh を使用すると、レポートを出力してカバレッジチェックを自動化できます。

テストのアンチパターン

1. 実装の詳細をテストする：

# BAD: Tests internal method calls
def test_service_calls_repo(mock_repo):
    service.create_user(data)
    mock_repo.insert.assert_called_once_with(...)

# GOOD: Tests observable behavior
async def test_create_user(db_session):
    service = UserService(db_session)
    user = await service.create_user(email="[email protected]", display_name="A")
    fetched = await service.get_user(user.id)
    assert fetched is not None

2. テスト間で変更可能な状態を共有する：

# BAD: Module-level mutable data
users = []

def test_add_user():
    users.append(User(id=1))  # Leaks to other tests

# GOOD: Use fixtures with function scope
@pytest.fixture
def users():
    return [UserFactory()]

3. アサーションの範囲が広すぎる：

# BAD
assert response.status_code == 200  # Only checks status

# GOOD
assert response.status_code == 200
data = response.json()
assert data["email"] == "[email protected]"
assert data["role"] == "member"

4. エラーパスのテストが不足している： 各エンドポイントには、少なくとも成功、見つからない場合、バリデーションエラー、未認証の場合のテストを用意してください。

例

本番環境向けの conftest 設定については、references/conftest-template.py を参照してください。 factory_boy のパターンについては、references/factory-template.py を参照してください。 API 統合テストのパターンについては、references/api-test-template.py を参照してください。 サービスの単体テストのパターンについては、references/service-test-template.py を参照してください。 完全な統合テストのパターンについては、references/integration-test-template.py を参照してください。