pytest

pytest - カスタマーサポート向けの高度な Python ユニットテスト

概要

pytest は Python アプリケーション向けの業界標準テストフレームワークであり、包括的で保守しやすく、拡張性の高いテストスイートを実現する強力な機能を備えています。このスキルでは、カスタマーサポート分野のテクノロジー活用支援に特化し、バックエンドのサポートシステム、チケット管理プラットフォーム、ナレッジベース、顧客データプラットフォームをテストするためのパターンと実践方法を紹介します。

カスタマーサポートシステムは、ミッションクリティカルな性質から、厳格なテストが求められます。停止やバグは、顧客満足度、エージェントの生産性、事業運営に直接影響します。このスキルでは、pytest を使用してサポートシステムのあらゆる側面をテストするための包括的なガイダンスを提供します。

カスタマーサポートシステムで pytest を使う理由

特有の要件

カスタマーサポートアプリケーションには、特有のテスト要件があります。

1. 高い信頼性：サポートシステムはミッションクリティカルであり、障害は顧客体験に直接影響します
2. 複雑なデータ関係：チケット、顧客、エージェント、コメント、添付ファイル、ナレッジ記事
3. 外部連携：メールサービス（SendGrid、AWS SES）、CRM システム（Salesforce、HubSpot）、決済処理サービス
4. 非同期処理：バックグラウンドジョブ、メールキュー、通知サービス、Webhook の配信
5. データ検証：顧客の個人を特定できる情報、チケットの優先度、SLA 要件、エスカレーションルールに対する厳格な検証
6. マルチテナント：異なる顧客組織やワークスペース間の分離
7. API ファースト設計：包括的なエンドポイントカバレッジが必要な RESTful API
8. リアルタイム機能：WebSocket 接続、ライブチャット、チケットのリアルタイム更新
9. コンプライアンス：GDPR、CCPA、データ保持ポリシー、監査ログ

pytest の利点

pytest は、次の機能によってこれらのニーズに対応します。

• 強力なフィクスチャシステム：複雑なデータベースのセットアップ、API クライアント、テストデータを管理できます
• パラメーター化：複数のシナリオ（さまざまなチケット種別、優先度、ユーザーロール）を効率よくテストできます
• 豊富なプラグインエコシステム：非同期テスト用の pytest-asyncio、モック用の pytest-mock、カバレッジ用の pytest-cov
• 優れた統合性：FastAPI、SQLAlchemy、PostgreSQL、Pydantic とシームレスに連携します
• 明確な出力：読みやすいテスト結果と包括的なエラー報告
• 拡張性：並列実行により、小規模なテストスイートから数千件のテストまで対応します
• 柔軟性：1 つのフレームワークで、ユニットテスト、統合テスト、エンドツーエンドテストをサポートします

中核となるスキル

1. フィクスチャと依存性注入

フィクスチャは pytest の強力な機能であり、再利用可能なセットアップ／ティアダウンのロジックと依存性注入を提供します。カスタマーサポートシステムでは、フィクスチャを使ってデータベース、API クライアント、テストデータ、外部サービスのモックを管理します。

基本的なフィクスチャ

import pytest
from app.models import Ticket, Customer, Agent

@pytest.fixture
def support_ticket():
    """Provide a basic support ticket dictionary."""
    return {
        "id": 1,
        "title": "Cannot access account",
        "description": "User unable to login after password reset",
        "status": "open",
        "priority": "high",
        "customer_email": "[email protected]",
        "category": "authentication"
    }

def test_ticket_structure(support_ticket):
    """Test ticket has required fields."""
    assert support_ticket["status"] == "open"
    assert support_ticket["priority"] in ["low", "medium", "high", "critical"]
    assert "@" in support_ticket["customer_email"]

フィクスチャのスコープ

スコープを使って、フィクスチャの作成と破棄のタイミングを制御します。

• function（デフォルト）：テスト関数ごとに 1 回作成され、テスト完了後に破棄されます
• class：テストクラスごとに 1 回作成され、すべてのメソッドで共有されます
• module：モジュールファイルごとに 1 回作成され、そのファイル内のすべてのテストで共有されます
• package：パッケージごとに 1 回作成され、そのパッケージ内のすべてのテストで共有されます
• session：テストセッション全体で 1 回作成され、すべてのテストで共有されます

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from typing import Generator

@pytest.fixture(scope="session")
def database_engine():
    """Create database engine once per session."""
    engine = create_engine(
        "postgresql://support_user:password@localhost/support_test",
        echo=False,
        pool_pre_ping=True,
        pool_size=10,
        max_overflow=20
    )
    # Create all tables
    from app.database import Base
    Base.metadata.create_all(engine)

    yield engine

    # Cleanup: drop all tables and dispose engine
    Base.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture(scope="function")
def db_session(database_engine) -> Generator[Session, None, None]:
    """Create a new database session for each test with automatic rollback."""
    SessionLocal = sessionmaker(bind=database_engine)
    session = SessionLocal()

    try:
        yield session
    finally:
        session.rollback()  # Rollback any changes
        session.close()

自動使用フィクスチャ

明示的に要求しなくても自動的に実行されるフィクスチャです。

@pytest.fixture(autouse=True)
def reset_caches():
    """Clear all caches before each test."""
    from app.cache import ticket_cache, customer_cache, agent_cache
    ticket_cache.clear()
    customer_cache.clear()
    agent_cache.clear()
    yield
    # Optional cleanup after test
    ticket_cache.clear()
    customer_cache.clear()
    agent_cache.clear()

@pytest.fixture(autouse=True, scope="session")
def configure_logging():
    """Configure logging for test session."""
    import logging
    logging.basicConfig(level=logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.ERROR)

フィクスチャの依存関係と合成

フィクスチャはほかのフィクスチャに依存し、依存関係の連鎖を作成できます。

@pytest.fixture
def database_url():
    """Provide database URL for testing."""
    return "postgresql://test:test@localhost:5432/support_test"

@pytest.fixture
def engine(database_url):
    """Create SQLAlchemy engine from database URL."""
    from sqlalchemy import create_engine
    return create_engine(database_url, echo=False)

@pytest.fixture
def session(engine):
    """Create database session from engine."""
    from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()

@pytest.fixture
def sample_customer(session):
    """Create a sample customer using database session."""
    customer = Customer(
        email="[email protected]",
        name="Test Customer",
        tier="premium",
        company="Acme Corp"
    )
    session.add(customer)
    session.commit()
    return customer

@pytest.fixture
def sample_ticket(session, sample_customer):
    """Create a sample ticket linked to sample customer."""
    ticket = Ticket(
        title="Test Ticket",
        description="Test description",
        priority="high",
        status="open",
        customer_id=sample_customer.id
    )
    session.add(ticket)
    session.commit()
    return ticket

ファクトリーフィクスチャ

柔軟なテストデータ作成のために、ファクトリー関数を返すフィクスチャを作成します。

@pytest.fixture
def ticket_factory(db_session):
    """Factory for creating tickets with custom attributes."""
    created_tickets = []

    def _create_ticket(
        title="Default Ticket",
        description="Default description",
        priority="medium",
        status="open",
        **kwargs
    ):
        ticket = Ticket(
            title=title,
            description=description,
            priority=priority,
            status=status,
            **kwargs
        )
        db_session.add(ticket)
        db_session.commit()
        created_tickets.append(ticket)
        return ticket

    yield _create_ticket

    # Cleanup: delete all created tickets
    for ticket in created_tickets:
        db_session.delete(ticket)
    db_session.commit()

def test_multiple_tickets(ticket_factory):
    """Test using factory to create multiple tickets."""
    high_priority = ticket_factory(priority="high", title="Urgent Issue")
    low_priority = ticket_factory(priority="low", title="Minor Request")

    assert high_priority.priority == "high"
    assert low_priority.priority == "low"

2. パラメーター化

パラメーター化を使うと、同じテストを異なる入力で実行できます。さまざまな優先度、ステータス、ユーザーロール、エッジケースを持つカスタマーサポートシステムの包括的なテストに不可欠です。

基本的なパラメーター化

@pytest.mark.parametrize("priority,expected_sla_hours", [
    ("critical", 1),
    ("high", 4),
    ("medium", 24),
    ("low", 72)
])
def test_sla_calculation(priority, expected_sla_hours):
    """Test SLA hours calculation for different priorities."""
    from app.services.sla import calculate_sla_hours
    sla = calculate_sla_hours(priority)
    assert sla == expected_sla_hours

@pytest.mark.parametrize("email,is_valid", [
    ("[email protected]", True),
    ("[email protected]", True),
    ("user@localhost", True),
    ("invalid.email", False),
    ("@example.com", False),
    ("user@", False),
    ("", False),
    ("user @example.com", False),
    ("user@example .com", False),
])
def test_email_validation(email, is_valid):
    """Test email validation with various valid and invalid inputs."""
    from app.validators import is_valid_email
    assert is_valid_email(email) == is_valid

複数のパラメーター

@pytest.mark.parametrize("status,priority,assigned,expected_urgent", [
    ("open", "critical", False, True),   # Unassigned critical = urgent
    ("open", "high", False, True),       # Unassigned high = urgent
    ("open", "medium", False, False),    # Unassigned medium = not urgent
    ("open", "low", False, False),       # Unassigned low = not urgent
    ("in_progress", "critical", True, False),  # Assigned critical = not urgent
    ("in_progress", "high", True, False),      # Assigned high = not urgent
    ("closed", "critical", True, False),       # Closed tickets never urgent
    ("resolved", "high", True, False),         # Resolved tickets never urgent
])
def test_urgent_ticket_identification(status, priority, assigned, expected_urgent):
    """Test identification of urgent tickets requiring immediate attention."""
    from app.models import Ticket
    ticket = Ticket(status=status, priority=priority)
    if assigned:
        ticket.assigned_agent_id = 123

    assert ticket.is_urgent() == expected_urgent

フィクスチャのパラメーター化

@pytest.fixture(params=["sqlite", "postgresql"])
def database_type(request):
    """Parametrized fixture for different database types."""
    return request.param

def test_database_connection(database_type):
    """Test runs twice: once with SQLite, once with PostgreSQL."""
    from app.database import get_connection
    connection = get_connection(database_type)
    assert connection.is_connected()
    connection.close()

@pytest.fixture(params=[
    {"role": "admin", "can_delete": True, "can_reassign": True},
    {"role": "agent", "can_delete": False, "can_reassign": True},
    {"role": "viewer", "can_delete": False, "can_reassign": False},
])
def user_permissions(request):
    """Parametrized fixture for different user roles and permissions."""
    return request.param

def test_permission_checks(user_permissions):
    """Test permission validation for different roles."""
    from app.auth import check_permission

    can_delete = check_permission(user_permissions["role"], "delete_ticket")
    can_reassign = check_permission(user_permissions["role"], "reassign_ticket")

    assert can_delete == user_permissions["can_delete"]
    assert can_reassign == user_permissions["can_reassign"]

複雑なケースでの pytest.param の使用

@pytest.mark.parametrize("email,expected_valid,reason", [
    pytest.param("[email protected]", True, "valid", id="valid-standard-email"),
    pytest.param("[email protected]", True, "valid", id="valid-with-plus"),
    pytest.param("invalid", False, "missing-at", id="no-at-symbol"),
    pytest.param("@example.com", False, "no-user", id="missing-username"),
    pytest.param("user@", False, "no-domain", id="missing-domain"),
    pytest.param("[email protected]", False, "invalid-domain", id="dot-at-domain-start"),
])
def test_email_validation_detailed(email, expected_valid, reason):
    """Test email validation with detailed reasoning."""
    from app.validators import validate_email_detailed
    result = validate_email_detailed(email)
    assert result.is_valid == expected_valid
    if not expected_valid:
        assert reason in result.error_code.lower()

ID によるパラメーター化

@pytest.mark.parametrize("ticket_data,expected_status", [
    ({"priority": "critical", "auto_assign": True}, "assigned"),
    ({"priority": "high", "auto_assign": True}, "assigned"),
    ({"priority": "medium", "auto_assign": False}, "open"),
    ({"priority": "low", "auto_assign": False}, "open"),
], ids=["critical-auto", "high-auto", "medium-manual", "low-manual"])
def test_auto_assignment(ticket_data, expected_status):
    """Test automatic ticket assignment based on priority."""
    from app.services.assignment import process_new_ticket
    ticket = process_new_ticket(**ticket_data)
    assert ticket.status == expected_status

3. pytest-mock によるモック

モックは、メールサービス、決済ゲートウェイ、CRM システム、サードパーティ API などの外部依存関係からコードの各単位を分離するために不可欠です。pytest-mock プラグインは、Python の unittest.mock に使いやすいインターフェースを提供します。

基本的なモック

def test_ticket_notification(mocker, db_session):
    """Test that creating a ticket sends email notification."""
    # Mock the email service
    mock_send = mocker.patch('app.services.email.EmailService.send_email')
    mock_send.return_value = {"status": "sent", "message_id": "msg_123"}

    # Create ticket
    from app.services.ticket import TicketService
    service = TicketService(db_session)
    ticket = service.create_ticket(
        title="Login Issue",
        description="Cannot access account",
        customer_email="[email protected]",
        priority="high"
    )

    # Verify email was sent
    mock_send.assert_called_once()
    call_args = mock_send.call_args
    assert call_args.kwargs['to'] == "[email protected]"
    assert "Login Issue" in call_args.kwargs['subject']
    assert ticket.id is not None

戻り値のモック

def test_customer_tier_lookup(mocker, api_client):
    """Test customer tier lookup from external CRM."""
    # Mock external CRM API response
    mock_response = mocker.Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {
        "customer_id": "CRM_12345",
        "tier": "enterprise",
        "support_plan": "24/7 premium",
        "account_manager": "[email protected]"
    }

    mocker.patch('requests.get', return_value=mock_response)

    # Test endpoint that uses CRM data
    response = api_client.get("/api/customers/[email protected]/tier")

    assert response.status_code == 200
    data = response.json()
    assert data["tier"] == "enterprise"
    assert data["support_plan"] == "24/7 premium"

副作用のモック

def test_retry_logic_on_api_failure(mocker):
    """Test retry mechanism when external API fails temporarily."""
    from app.services.external import call_with_retry

    mock_api = mocker.patch('app.services.external.call_external_api')
    # First two calls fail, third succeeds
    mock_api.side_effect = [
        ConnectionError("Connection timeout"),
        ConnectionError("Connection timeout"),
        {"success": True, "data": "response"}
    ]

    result = call_with_retry(max_attempts=3, backoff_seconds=0)

    assert result["success"] is True
    assert result["data"] == "response"
    assert mock_api.call_count == 3

def test_email_fallback_on_primary_failure(mocker):
    """Test email service fallback when primary provider fails."""
    from app.services.email import send_email_with_fallback

    mock_primary = mocker.patch('app.services.email.sendgrid_send')
    mock_fallback = mocker.patch('app.services.email.ses_send')

    # Primary fails, fallback succeeds
    mock_primary.side_effect = Exception("SendGrid API error")
    mock_fallback.return_value = {"status": "sent", "provider": "ses"}

    result = send_email_with_fallback(
        to="[email protected]",
        subject="Ticket Update",
        body="Your ticket has been updated"
    )

    assert result["status"] == "sent"
    assert result["provider"] == "ses"
    mock_primary.assert_called_once()
    mock_fallback.assert_called_once()

実際のメソッドのスパイ

def test_cache_utilization(mocker, db_session):
    """Spy on cache to verify it's being used correctly."""
    from app.cache import ticket_cache

    spy_get = mocker.spy(ticket_cache, 'get')
    spy_set = mocker.spy(ticket_cache, 'set')

    from app.services.ticket import get_ticket

    # First call: cache miss, should query database
    ticket1 = get_ticket(123)
    assert spy_get.call_count == 1
    assert spy_set.call_count == 1

    # Second call: cache hit, should not query database
    ticket2 = get_ticket(123)
    assert spy_get.call_count == 2
    assert spy_set.call_count == 1  # Not called again

    assert ticket1.id == ticket2.id

データベース操作のモック

def test_ticket_creation_business_logic(mocker):
    """Test ticket creation logic without database."""
    mock_session = mocker.Mock()
    mock_session.add = mocker.Mock()
    mock_session.commit = mocker.Mock()
    mock_session.refresh = mocker.Mock(side_effect=lambda obj: setattr(obj, 'id', 123))

    from app.services.ticket import create_ticket_entity
    ticket = create_ticket_entity(
        mock_session,
        title="Test Ticket",
        priority="high"
    )

    assert ticket.id == 123
    assert ticket.priority == "high"
    mock_session.add.assert_called_once()
    mock_session.commit.assert_called_once()

非同期モック

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_async_notification_service(mocker):
    """Test async notification service with proper async mocking."""
    mock_send = mocker.patch(
        'app.services.notification.send_push_notification',
        new_callable=AsyncMock
    )
    mock_send.return_value = {
        "delivered": True,
        "notification_id": "notif_789"
    }

    from app.services.notification import NotificationService
    service = NotificationService()
    result = await service.notify_agent(
        agent_id=456,
        message="New high-priority ticket assigned to you",
        priority="high"
    )

    assert result["delivered"] is True
    assert result["notification_id"] is not None
    mock_send.assert_awaited_once()

    # Verify call arguments
    call_kwargs = mock_send.call_args.kwargs
    assert call_kwargs["agent_id"] == 456
    assert "high-priority" in call_kwargs["message"]

4. カバレッジレポート

コードカバレッジは、テスト中にコードの何パーセントが実行されたかを測定します。カスタマーサポートシステムでは、包括的なカバレッジによって信頼性を確保し、テストされていないエッジケースの特定に役立ちます。

基本的なカバレッジ設定

# Install pytest-cov
pip install pytest-cov

# Run tests with coverage
pytest --cov=app tests/

# Generate HTML report
pytest --cov=app --cov-report=html tests/

# Show lines not covered
pytest --cov=app --cov-report=term-missing tests/

# Fail if coverage below threshold
pytest --cov=app --cov-fail-under=80 tests/

pytest.ini での設定

[pytest]
addopts =
    --cov=app
    --cov-report=html
    --cov-report=term-missing
    --cov-report=xml
    --cov-fail-under=80
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
markers =
    unit: Unit tests
    integration: Integration tests
    slow: Slow running tests

カバレッジ設定（.coveragerc）

[run]
source = app
omit =
    */tests/*
    */migrations/*
    */__init__.py
    */config.py
    */venv/*
    */.venv/*

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    def __str__
    raise AssertionError
    raise NotImplementedError
    if __name__ == .__main__.:
    if TYPE_CHECKING:
    @abstractmethod
    @abc.abstractmethod
    except ImportError
precision = 2
show_missing = True

[html]
directory = htmlcov
title = Customer Support API Coverage Report

分岐カバレッジ

[run]
source = app
branch = True  # Enable branch coverage

[report]
show_missing = True
skip_covered = False

# Run with branch coverage
pytest --cov=app --cov-branch --cov-report=html

5. 非同期テスト

最新のカスタマーサポートシステムでは、パフォーマンスとスケーラビリティを向上させるために非同期操作を使用します。非同期コードのテストでは、コルーチンを適切に await するための特別な処理が必要です。

基本的な非同期テスト

import pytest

@pytest.mark.asyncio
async def test_async_ticket_creation():
    """Test asynchronous ticket creation service."""
    from app.services.ticket import create_ticket_async

    ticket = await create_ticket_async(
        title="Async Test Ticket",
        description="Testing async operations",
        priority="high",
        customer_email="[email protected]"
    )

    assert ticket.id is not None
    assert ticket.status == "open"
    assert ticket.priority == "high"

非同期フィクスチャ

import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

@pytest.fixture
async def async_db_session():
    """Provide async database session."""
    engine = create_async_engine(
        "postgresql+asyncpg://user:pass@localhost/support_test",
        echo=False,
        future=True
    )

    async_session_maker = sessionmaker(
        engine,
        class_=AsyncSession,
        expire_on_commit=False
    )

    async with async_session_maker() as session:
        yield session
        await session.rollback()

    await engine.dispose()

@pytest.mark.asyncio
async def test_async_database_operation(async_db_session):
    """Test async database operations."""
    from app.models import Ticket
    from sqlalchemy import select

    # Create ticket
    ticket = Ticket(title="Async DB Test", priority="medium")
    async_db_session.add(ticket)
    await async_db_session.commit()

    # Query ticket
    result = await async_db_session.execute(
        select(Ticket).where(Ticket.title == "Async DB Test")
    )
    found_ticket = result.scalar_one()

    assert found_ticket.id is not None
    assert found_ticket.priority == "medium"

非同期バックグラウンドタスクのテスト

@pytest.mark.asyncio
async def test_background_email_task(mocker):
    """Test async background task for sending emails."""
    from app.tasks.email import process_ticket_notifications

    mock_send = mocker.patch(
        'app.tasks.email.send_email_async',
        new_callable=AsyncMock
    )
    mock_send.return_value = {"status": "sent", "message_id": "msg_456"}

    await process_ticket_notifications(ticket_id=123)

    mock_send.assert_awaited_once()
    call_kwargs = mock_send.call_args.kwargs
    assert "ticket_id" in call_kwargs or call_kwargs.get("ticket_id") == 123

並行操作のテスト

import asyncio

@pytest.mark.asyncio
async def test_concurrent_ticket_updates():
    """Test handling of concurrent ticket updates."""
    from app.services.ticket import update_ticket_async, add_comment_async

    ticket_id = 1

    async def update_status():
        return await update_ticket_async(ticket_id, status="in_progress")

    async def add_comment():
        return await add_comment_async(ticket_id, "Working on this issue")

    async def assign_agent():
        return await update_ticket_async(ticket_id, assigned_agent_id=42)

    # Execute concurrently
    results = await asyncio.gather(
        update_status(),
        add_comment(),
        assign_agent()
    )

    assert results[0]["status"] == "in_progress"
    assert results[1]["comment_count"] >= 1
    assert results[2]["assigned_agent_id"] == 42

非同期テストの設定

# pytest.ini or pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # Automatically detect and run async tests

# Or in pytest.ini
[pytest]
asyncio_mode = auto

6. FastAPI のテスト

FastAPI はカスタマーサポート API で広く使用されています。エンドポイントをテストすることで、リクエストとレスポンスの処理、検証、認証、ビジネスロジックが適切に機能することを確認できます。

FastAPI の基本的なテスト

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

@pytest.fixture
def client():
    """Provide FastAPI test client."""
    return TestClient(app)

def test_create_ticket_endpoint(client):
    """Test ticket creation via POST endpoint."""
    response = client.post(
        "/api/v1/tickets",
        json={
            "title": "Cannot login to application",
            "description": "User receives 'invalid credentials' error",
            "priority": "high",
            "customer_email": "[email protected]",
            "category": "authentication"
        }
    )

    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Cannot login to application"
    assert data["priority"] == "high"
    assert "id" in data
    assert "created_at" in data

def test_get_ticket_by_id(client, sample_tickets):
    """Test retrieving specific ticket by ID."""
    ticket_id = sample_tickets[0].id

    response = client.get(f"/api/v1/tickets/{ticket_id}")

    assert response.status_code == 200
    data = response.json()
    assert data["id"] == ticket_id
    assert data["title"] == sample_tickets[0].title

def test_update_ticket_status(client, sample_tickets):
    """Test updating ticket status via PATCH."""
    ticket_id = sample_tickets[0].id

    response = client.patch(
        f"/api/v1/tickets/{ticket_id}",
        json={"status": "in_progress"}
    )

    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "in_progress"
    assert data["id"] == ticket_id

依存関係のオーバーライド

FastAPI の依存性注入は、テスト用にオーバーライドできます。

from app.dependencies import get_db, get_current_user
from app.models import User

@pytest.fixture
def authenticated_client(client, mocker):
    """Provide authenticated test client."""
    mock_user = User(
        id=1,
        email="[email protected]",
        role="agent",
        name="Test Agent"
    )

    async def override_get_current_user():
        return mock_user

    app.dependency_overrides[get_current_user] = override_get_current_user
    yield client
    app.dependency_overrides.clear()

def test_protected_endpoint(authenticated_client):
    """Test endpoint requiring authentication."""
    response = authenticated_client.get("/api/v1/admin/users")
    assert response.status_code == 200
    data = response.json()
    assert isinstance(data, list)

@pytest.fixture
def admin_client(client, mocker):
    """Provide client authenticated as admin."""
    mock_admin = User(
        id=1,
        email="[email protected]",
        role="admin",
        permissions=["read", "write", "delete", "admin"]
    )

    async def override_get_current_user():
        return mock_admin

    app.dependency_overrides[get_current_user] = override_get_current_user
    yield client
    app.dependency_overrides.clear()

def test_admin_only_endpoint(admin_client):
    """Test endpoint requiring admin privileges."""
    response = admin_client.delete("/api/v1/tickets/123")
    assert response.status_code in [200, 204, 404]  # Success or not found

データベースを使ったテスト

@pytest.fixture
def client_with_db(db_session):
    """Provide FastAPI client with database session override."""
    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()

def test_ticket_lifecycle(client_with_db, db_session):
    """Test complete ticket lifecycle with database persistence."""
    # Create ticket
    create_response = client_with_db.post(
        "/api/v1/tickets",
        json={
            "title": "Lifecycle Test",
            "description": "Testing full CRUD cycle",
            "priority": "medium"
        }
    )
    assert create_response.status_code == 201
    ticket_id = create_response.json()["id"]

    # Read ticket
    get_response = client_with_db.get(f"/api/v1/tickets/{ticket_id}")
    assert get_response.status_code == 200
    assert get_response.json()["title"] == "Lifecycle Test"

    # Update ticket
    update_response = client_with_db.patch(
        f"/api/v1/tickets/{ticket_id}",
        json={"status": "resolved", "priority": "low"}
    )
    assert update_response.status_code == 200
    assert update_response.json()["status"] == "resolved"

    # Verify in database
    from app.models import Ticket
    ticket = db_session.query(Ticket).filter_by(id=ticket_id).first()
    assert ticket is not None
    assert ticket.status == "resolved"
    assert ticket.priority == "low"

    # Delete ticket
    delete_response = client_with_db.delete(f"/api/v1/tickets/{ticket_id}")
    assert delete_response.status_code in [200, 204]

    # Verify deletion
    get_deleted = client_with_db.get(f"/api/v1/tickets/{ticket_id}")
    assert get_deleted.status_code == 404

リクエスト検証のテスト

def test_invalid_ticket_creation(client):
    """Test that invalid ticket data returns proper validation errors."""
    # Missing required field
    response = client.post(
        "/api/v1/tickets",
        json={"description": "Missing title"}
    )
    assert response.status_code == 422

    # Invalid priority
    response = client.post(
        "/api/v1/tickets",
        json={
            "title": "Test",
            "description": "Test",
            "priority": "invalid_priority"
        }
    )
    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any("priority" in str(e).lower() for e in errors)

    # Invalid email format
    response = client.post(
        "/api/v1/tickets",
        json={
            "title": "Test",
            "description": "Test",
            "customer_email": "not_an_email"
        }
    )
    assert response.status_code == 422

7. SQLAlchemy のテスト

データベースモデル、クエリ、リレーションシップ、操作をテストすることで、カスタマーサポートシステムのデータ整合性を確保できます。

モデルのテスト

from app.models import Ticket, Comment, User, Customer

def test_ticket_model_creation(db_session):
    """Test creating and persisting ticket model."""
    customer = Customer(
        email="[email protected]",
        name="John Doe",
        tier="enterprise"
    )
    db_session.add(customer)
    db_session.flush()

    ticket = Ticket(
        title="Database Connection Error",
        description="Cannot connect to production database",
        priority="critical",
        status="open",
        customer_id=customer.id,
        category="technical"
    )
    db_session.add(ticket)
    db_session.commit()

    assert ticket.id is not None
    assert ticket.created_at is not None
    assert ticket.customer.email == "[email protected]"

def test_ticket_model_defaults(db_session):
    """Test default values on ticket model."""
    ticket = Ticket(title="Test", description="Test")
    db_session.add(ticket)
    db_session.commit()

    assert ticket.status == "open"  # Default status
    assert ticket.priority == "medium"  # Default priority
    assert ticket.created_at is not None
    assert ticket.updated_at is not None

リレーションシップのテスト

def test_ticket_comments_relationship(db_session):
    """Test one-to-many relationship between tickets and comments."""
    ticket = Ticket(
        title="Test Ticket",
        description="Testing relationships",
        priority="low"
    )
    db_session.add(ticket)
    db_session.flush()

    # Add multiple comments
    comment1 = Comment(
        ticket_id=ticket.id,
        content="First comment",
        author_email="[email protected]"
    )
    comment2 = Comment(
        ticket_id=ticket.id,
        content="Second comment",
        author_email="[email protected]"
    )
    db_session.add_all([comment1, comment2])
    db_session.commit()

    # Reload ticket and verify relationship
    db_session.expire(ticket)
    assert len(ticket.comments) == 2
    assert ticket.comments[0].content == "First comment"
    assert ticket.comments[1].content == "Second comment"

def test_customer_tickets_relationship(db_session):
    """Test one-to-many relationship between customers and tickets."""
    customer = Customer(
        email="[email protected]",
        name="Loyal Customer"
    )
    db_session.add(customer)
    db_session.flush()

    # Create multiple tickets for same customer
    for i in range(3):
        ticket = Ticket(
            title=f"Issue #{i+1}",
            description=f"Description {i+1}",
            customer_id=customer.id
        )
        db_session.add(ticket)

    db_session.commit()

    # Verify customer has all tickets
    db_session.expire(customer)
    assert len(customer.tickets) == 3
    assert all(t.customer_id == customer.id for t in customer.tickets)

クエリのテスト

def test_filter_tickets_by_status(db_session):
    """Test filtering tickets by status."""
    # Create tickets with different statuses
    tickets = [
        Ticket(title=f"Ticket {i}", status="open" if i % 2 == 0 else "closed")
        for i in range(10)
    ]
    db_session.add_all(tickets)
    db_session.commit()

    # Query open tickets
    open_tickets = db_session.query(Ticket).filter_by(status="open").all()
    assert len(open_tickets) == 5
    assert all(t.status == "open" for t in open_tickets)

    # Query closed tickets
    closed_tickets = db_session.query(Ticket).filter_by(status="closed").all()
    assert len(closed_tickets) == 5
    assert all(t.status == "closed" for t in closed_tickets)

def test_complex_ticket_query(db_session):
    """Test complex multi-filter ticket query."""
    from sqlalchemy import and_, or_
    from datetime import datetime, timedelta

    # Create diverse set of tickets
    old_critical = Ticket(
        title="Old Critical",
        priority="critical",
        status="open",
        created_at=datetime.utcnow() - timedelta(days=7)
    )
    recent_high = Ticket(
        title="Recent High",
        priority="high",
        status="open",
        created_at=datetime.utcnow() - timedelta(hours=2)
    )
    old_low = Ticket(
        title="Old Low",
        priority="low",
        status="open",
        created_at=datetime.utcnow() - timedelta(days=30)
    )

    db_session.add_all([old_critical, recent_high, old_low])
    db_session.commit()

    # Query: open tickets that are either critical OR created in last 24 hours
    cutoff = datetime.utcnow() - timedelta(days=1)
    urgent_tickets = db_session.query(Ticket).filter(
        and_(
            Ticket.status == "open",
            or_(
                Ticket.priority == "critical",
                Ticket.created_at >= cutoff
            )
        )
    ).all()

    assert len(urgent_tickets) == 2
    ticket_titles = [t.title for t in urgent_tickets]
    assert "Old Critical" in ticket_titles
    assert "Recent High" in ticket_titles
    assert "Old Low" not in ticket_titles

トランザクションのテスト

def test_transaction_rollback_on_error(db_session):
    """Test that failed transactions are properly rolled back."""
    from sqlalchemy.exc import IntegrityError

    # Create initial ticket
    ticket1 = Ticket(title="First Ticket", description="Test")
    db_session.add(ticket1)
    db_session.commit()

    initial_count = db_session.query(Ticket).count()

    # Attempt to create invalid ticket (will fail)
    try:
        ticket2 = Ticket(title="Second Ticket", description="Test")
        db_session.add(ticket2)
        db_session.flush()

        # Simulate constraint violation
        ticket3 = Ticket(
            title="Third Ticket",
            external_id="DUPLICATE_ID"  # Assuming unique constraint
        )
        db_session.add(ticket3)

        ticket4 = Ticket(
            title="Fourth Ticket",
            external_id="DUPLICATE_ID"  # Will violate unique constraint
        )
        db_session.add(ticket4)
        db_session.commit()
    except IntegrityError:
        db_session.rollback()

    # Verify count hasn't changed
    final_count = db_session.query(Ticket).count()
    assert final_count == initial_count

8. データベースフィクスチャ

適切に設計されたデータベースフィクスチャは、複雑なデータ関係を持つカスタマーサポートシステムのテストに不可欠です。

包括的なデータベースフィクスチャ

import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from app.database import Base

@pytest.fixture(scope="session")
def db_engine():
    """Create test database engine for entire session."""
    engine = create_engine(
        "postgresql://test:test@localhost:5432/support_test",
        echo=False,
        pool_pre_ping=True
    )

    # Create all tables
    Base.metadata.create_all(engine)

    yield engine

    # Cleanup: drop all tables
    Base.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture
def db_session(db_engine):
    """Provide clean database session with automatic rollback."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()

    # Enable savepoints for nested transactions
    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(session, transaction):
        if transaction.nested and not transaction._parent.nested:
            session.expire_all()
            session.begin_nested()

    session.begin_nested()

    yield session

    # Cleanup
    session.close()
    transaction.rollback()
    connection.close()

サンプルデータフィクスチャ

@pytest.fixture
def sample_customers(db_session):
    """Create sample customers with various tiers."""
    customers = [
        Customer(email="[email protected]", name="Free User", tier="free"),
        Customer(email="[email protected]", name="Basic User", tier="basic"),
        Customer(email="[email protected]", name="Premium User", tier="premium"),
        Customer(email="[email protected]", name="Enterprise User", tier="enterprise"),
    ]
    db_session.add_all(customers)
    db_session.commit()
    return customers

@pytest.fixture
def sample_agents(db_session):
    """Create sample support agents."""
    agents = [
        User(email="[email protected]", name="Agent One", role="agent"),
        User(email="[email protected]", name="Agent Two", role="agent"),
        User(email="[email protected]", name="Senior Agent", role="senior_agent"),
    ]
    db_session.add_all(agents)
    db_session.commit()
    return agents

@pytest.fixture
def sample_tickets(db_session, sample_customers):
    """Create diverse set of sample tickets."""
    tickets = []
    priorities = ["low", "medium", "high", "critical"]
    statuses = ["open", "in_progress", "resolved", "closed"]

    for i in range(12):
        ticket = Ticket(
            title=f"Sample Ticket #{i+1}",
            description=f"Description for ticket {i+1}",
            priority=priorities[i % len(priorities)],
            status=statuses[i % len(statuses)],
            customer_id=sample_customers[i % len(sample_customers)].id,
            category=["technical", "billing", "account"][i % 3]
        )
        tickets.append(ticket)

    db_session.add_all(tickets)
    db_session.commit()
    return tickets

ファクトリーパターンのフィクスチャ

@pytest.fixture
def customer_factory(db_session):
    """Factory for creating customers with custom attributes."""
    created = []

    def _create_customer(
        email=None,
        name=None,
        tier="basic",
        **kwargs
    ):
        import uuid
        customer = Customer(
            email=email or f"customer{uuid.uuid4().hex[:8]}@example.com",
            name=name or f"Customer {uuid.uuid4().hex[:8]}",
            tier=tier,
            **kwargs
        )
        db_session.add(customer)
        db_session.commit()
        created.append(customer)
        return customer

    yield _create_customer

    # Cleanup
    for customer in created:
        db_session.delete(customer)
    db_session.commit()

@pytest.fixture
def ticket_factory(db_session, customer_factory):
    """Factory for creating tickets with automatic customer creation."""
    created = []

    def _create_ticket(
        title=None,
        customer=None,
        priority="medium",
        status="open",
        **kwargs
    ):
        import uuid

        if customer is None:
            customer = customer_factory()

        ticket = Ticket(
            title=title or f"Ticket {uuid.uuid4().hex[:8]}",
            description=kwargs.pop("description", "Test ticket description"),
            customer_id=customer.id,
            priority=priority,
            status=status,
            **kwargs
        )
        db_session.add(ticket)
        db_session.commit()
        created.append(ticket)
        return ticket

    yield _create_ticket

    # Cleanup
    for ticket in created:
        db_session.delete(ticket)
    db_session.commit()

# Usage in tests
def test_with_factories(customer_factory, ticket_factory):
    """Test using factory fixtures for flexible test data creation."""
    # Create premium customer
    premium_customer = customer_factory(tier="enterprise", name="Big Corp")

    # Create multiple tickets for same customer
    critical_ticket = ticket_factory(
        customer=premium_customer,
        priority="critical",
        title="Production outage"
    )
    normal_ticket = ticket_factory(
        customer=premium_customer,
        priority="low",
        title="Feature request"
    )

    assert critical_ticket.customer_id == premium_customer.id
    assert normal_ticket.customer_id == premium_customer.id
    assert premium_customer.tier == "enterprise"

9. カスタマーサポートのテストシナリオ

カスタマーサポートシステムに特化した、実際のユースケースに基づくテストシナリオです。

チケット割り当てロジックのテスト

def test_round_robin_assignment(db_session, sample_agents):
    """Test round-robin ticket assignment to available agents."""
    from app.services.assignment import assign_ticket_round_robin

    tickets = []
    for i in range(9):
        ticket = Ticket(
            title=f"Assignment Test {i}",
            priority="medium",
            status="open"
        )
        db_session.add(ticket)
        db_session.flush()

        assign_ticket_round_robin(ticket, db_session)
        tickets.append(ticket)

    db_session.commit()

    # Verify even distribution
    for agent in sample_agents:
        assigned_count = len([t for t in tickets if t.assigned_agent_id == agent.id])
        assert assigned_count == 3  # 9 tickets / 3 agents = 3 each

def test_priority_based_assignment(db_session, sample_agents):
    """Test that critical tickets go to senior agents."""
    from app.services.assignment import assign_ticket_by_priority

    critical_ticket = Ticket(
        title="Critical Issue",
        priority="critical",
        status="open"
    )
    db_session.add(critical_ticket)
    db_session.flush()

    assign_ticket_by_priority(critical_ticket, db_session)
    db_session.commit()

    # Find senior agent
    senior_agent = next(a for a in sample_agents if a.role == "senior_agent")
    assert critical_ticket.assigned_agent_id == senior_agent.id

SLA 遵守のテスト

from datetime import datetime, timedelta

def test_sla_breach_detection(db_session):
    """Test detection of SLA breaches for different priorities."""
    from app.services.sla import find_sla_breaches

    now = datetime.utcnow()

    # Critical ticket created 2 hours ago (SLA: 1 hour) - BREACHED
    breached_critical = Ticket(
        title="Critical - Breached",
        priority="critical",
        status="open",
        created_at=now - timedelta(hours=2)
    )

    # High priority created 5 hours ago (SLA: 4 hours) - BREACHED
    breached_high = Ticket(
        title="High - Breached",
        priority="high",
        status="open",
        created_at=now - timedelta(hours=5)
    )

    # Medium priority created 2 hours ago (SLA: 24 hours) - OK
    ok_medium = Ticket(
        title="Medium - OK",
        priority="medium",
        status="open",
        created_at=now - timedelta(hours=2)
    )

    # Critical ticket but already assigned (not breached)
    assigned_critical = Ticket(
        title="Critical - Assigned",
        priority="critical",
        status="in_progress",
        created_at=now - timedelta(hours=2),
        assigned_agent_id=1
    )

    db_session.add_all([breached_critical, breached_high, ok_medium, assigned_critical])
    db_session.commit()

    breaches = find_sla_breaches(db_session)
    breach_titles = [b.title for b in breaches]

    assert len(breaches) == 2
    assert "Critical - Breached" in breach_titles
    assert "High - Breached" in breach_titles
    assert "Medium - OK" not in breach_titles
    assert "Critical - Assigned" not in breach_titles

def test_sla_time_remaining_calculation(db_session):
    """Test calculation of remaining SLA time."""
    from app.services.sla import calculate_sla_remaining

    now = datetime.utcnow()

    # High priority ticket created 1 hour ago (SLA: 4 hours)
    ticket = Ticket(
        title="SLA Test",
        priority="high",
        status="open",
        created_at=now - timedelta(hours=1)
    )
    db_session.add(ticket)
    db_session.commit()

    remaining = calculate_sla_remaining(ticket)

    # Should have ~3 hours remaining (4 hour SLA - 1 hour elapsed)
    assert 2.9 <= remaining.total_seconds() / 3600 <= 3.1

エスカレーションルールのテスト

def test_automatic_escalation(db_session):
    """Test automatic ticket escalation for overdue tickets."""
    from app.services.escalation import escalate_overdue_tickets

    now = datetime.utcnow()

    # Medium priority ticket open for 3 days without assignment
    overdue_ticket = Ticket(
        title="Overdue Ticket",
        priority="medium",
        status="open",
        created_at=now - timedelta(days=3)
    )

    # Recent ticket (should not escalate)
    recent_ticket = Ticket(
        title="Recent Ticket",
        priority="medium",
        status="open",
        created_at=now - timedelta(hours=2)
    )

    db_session.add_all([overdue_ticket, recent_ticket])
    db_session.commit()

    # Run escalation logic
    escalated_count = escalate_overdue_tickets(db_session)

    db_session.refresh(overdue_ticket)
    db_session.refresh(recent_ticket)

    assert escalated_count == 1
    assert overdue_ticket.priority == "high"  # Escalated from medium
    assert overdue_ticket.escalated is True
    assert recent_ticket.priority == "medium"  # Unchanged
    assert recent_ticket.escalated is False

顧客通知のテスト

def test_status_change_notification(mocker, db_session):
    """Test email notification sent when ticket status changes."""
    from app.services.ticket import update_ticket_status

    mock_send_email = mocker.patch('app.services.email.send_email')
    mock_send_email.return_value = {"status": "sent"}

    ticket = Ticket(
        title="Notification Test",
        customer_email="[email protected]",
        status="open"
    )
    db_session.add(ticket)
    db_session.commit()

    # Update status
    update_ticket_status(ticket.id, "resolved", db_session)

    # Verify notification sent
    mock_send_email.assert_called_once()
    call_kwargs = mock_send_email.call_args.kwargs
    assert call_kwargs["to"] == "[email protected]"
    assert "resolved" in call_kwargs["template_name"].lower()
    assert call_kwargs["template_data"]["ticket_id"] == ticket.id

def test_assignment_notification(mocker, db_session, sample_agents):
    """Test notification sent to agent when ticket assigned."""
    from app.services.ticket import assign_ticket

    mock_notify = mocker.patch('app.services.notification.notify_agent')

    ticket = Ticket(title="Assignment Notification Test", status="open")
    db_session.add(ticket)
    db_session.commit()

    agent = sample_agents[0]
    assign_ticket(ticket.id, agent.id, db_session)

    mock_notify.assert_called_once()
    call_args = mock_notify.call_args
    assert call_args[0][0] == agent.id  # First arg is agent_id
    assert "assigned" in call_args[0][1].lower()  # Second arg is message

10. 統合パターン

外部サービスやシステムとの統合をテストします。

外部 API 統合のテスト

@pytest.mark.integration
def test_crm_sync_integration(mocker, db_session):
    """Test synchronization with external CRM system."""
    from app.services.crm import sync_customer_to_crm

    # Mock external CRM API
    mock_response = mocker.Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {
        "crm_customer_id": "CRM_98765",
        "sync_status": "success",
        "last_sync": "2025-01-15T10:30:00Z"
    }

    mocker.patch('requests.post', return_value=mock_response)

    customer = Customer(
        email="[email protected]",
        name="CRM Test Customer",
        tier="enterprise"
    )
    db_session.add(customer)
    db_session.commit()

    result = sync_customer_to_crm(customer.id, db_session)

    assert result["sync_status"] == "success"
    assert result["crm_customer_id"] == "CRM_98765"

    # Verify customer record updated
    db_session.refresh(customer)
    assert customer.crm_id == "CRM_98765"

Webhook 配信のテスト

def test_webhook_delivery_on_ticket_created(mocker, db_session):
    """Test webhook delivery when new ticket is created."""
    from app.services.webhooks import deliver_webhook

    # Setup webhook subscription
    webhook = WebhookSubscription(
        url="https://customer-app.example.com/webhooks/support",
        events=["ticket.created", "ticket.updated"],
        active=True
    )
    db_session.add(webhook)
    db_session.commit()

    mock_post = mocker.patch('requests.post')
    mock_post.return_value.status_code = 200

    ticket = Ticket(
        title="Webhook Test Ticket",
        priority="high"
    )
    db_session.add(ticket)
    db_session.commit()

    # Trigger webhook delivery
    from app.services.webhooks import trigger_webhooks
    trigger_webhooks("ticket.created", ticket.to_dict(), db_session)

    # Verify webhook called
    assert mock_post.call_count == 1
    call_kwargs = mock_post.call_args.kwargs
    assert call_kwargs["url"] == "https://customer-app.example.com/webhooks/support"
    assert call_kwargs["json"]["event"] == "ticket.created"
    assert call_kwargs["json"]["data"]["id"] == ticket.id

バックグラウンドジョブ処理のテスト

@pytest.mark.asyncio
async def test_background_export_job(db_session, sample_tickets):
    """Test asynchronous ticket export background job."""
    from app.tasks.export import export_tickets_job

    job = Job(
        type="export_tickets",
        status="pending",
        params={
            "format": "csv",
            "date_range": "last_7_days",
            "filters": {"priority": ["high", "critical"]}
        }
    )
    db_session.add(job)
    db_session.commit()

    # Process job
    result = await export_tickets_job(job.id, db_session)

    db_session.refresh(job)

    assert job.status == "completed"
    assert job.result is not None
    assert "file_url" in job.result
    assert "row_count" in job.result
    assert result["success"] is True

11. ベストプラクティス

テストの構成

アプリケーションのアーキテクチャを反映するようにテストを構成します。

tests/
├── unit/                   # Fast, isolated unit tests
│   ├── test_models.py
│   ├── test_validators.py
│   ├── test_utils.py
│   └── services/
│       ├── test_ticket_service.py
│       ├── test_email_service.py
│       └── test_sla_service.py
├── integration/           # Tests with database/external deps
│   ├── test_api_tickets.py
│   ├── test_api_customers.py
│   ├── test_database_ops.py
│   └── test_external_apis.py
├── e2e/                   # End-to-end workflow tests
│   ├── test_ticket_lifecycle.py
│   └── test_customer_journey.py
├── performance/           # Performance/load tests
│   └── test_api_performance.py
└── conftest.py           # Shared fixtures

説明的なテスト名

# Good - clear, specific test names
def test_high_priority_ticket_sends_immediate_notification_to_assigned_agent():
    pass

def test_ticket_auto_closes_after_7_days_without_customer_response():
    pass

def test_sla_breach_alert_sent_to_team_lead_when_critical_ticket_unassigned_for_1_hour():
    pass

# Bad - vague test names
def test_ticket_notification():
    pass

def test_auto_close():
    pass

def test_sla():
    pass

AAA パターン（Arrange-Act-Assert）

def test_ticket_assignment():
    # Arrange: Set up test data and conditions
    agent = create_agent(name="Test Agent", role="agent")
    ticket = create_ticket(title="Test", priority="high", status="open")

    # Act: Perform the action being tested
    result = assign_ticket(ticket.id, agent.id)

    # Assert: Verify expected outcomes
    assert result.assigned_agent_id == agent.id
    assert result.status == "assigned"
    assert result.assigned_at is not None

テストの分離

# Each test should be independent and not rely on others

# Bad - tests depend on execution order
def test_create_user():
    create_user("[email protected]")

def test_get_user():
    user = get_user("[email protected]")  # Relies on test_create_user
    assert user is not None

# Good - each test is self-contained
@pytest.fixture
def test_user(db_session):
    user = User(email="[email protected]")
    db_session.add(user)
    db_session.commit()
    return user

def test_get_user_by_email(db_session, test_user):
    found_user = get_user(test_user.email, db_session)
    assert found_user.email == test_user.email

テストのガイドライン

1. テストごとに1つのことをテストします。

# Good
def test_ticket_creation_sets_default_status():
    ticket = create_ticket(title="Test")
    assert ticket.status == "open"

def test_ticket_creation_sets_created_timestamp():
    ticket = create_ticket(title="Test")
    assert ticket.created_at is not None

# Bad
def test_ticket_creation():
    ticket = create_ticket(title="Test")
    assert ticket.status == "open"
    assert ticket.created_at is not None
    assert ticket.priority == "medium"
    assert ticket.category is None

2. セットアップにはフィクスチャを使用します。

# Good
@pytest.fixture
def authenticated_user(db_session):
    user = User(email="[email protected]", role="agent")
    db_session.add(user)
    db_session.commit()
    return user

def test_authenticated_endpoint(authenticated_user, api_client):
    response = api_client.get("/api/tickets", user=authenticated_user)
    assert response.status_code == 200

# Less ideal - setup in test
def test_authenticated_endpoint(api_client):
    user = User(email="[email protected]", role="agent")
    # ... more setup ...
    response = api_client.get("/api/tickets", user=user)
    assert response.status_code == 200

3. 外部依存関係をモックします。

def test_email_notification_on_ticket_creation(mocker):
    """Always mock external services like email."""
    mock_send = mocker.patch('app.services.email.send_email')
    mock_send.return_value = {"status": "sent"}

    ticket = create_ticket(title="Test", customer_email="[email protected]")

    mock_send.assert_called_once()
    assert ticket.id is not None

12. よくある落とし穴と解決策

問題：データベース状態の漏洩

データベースの状態が共有されているため、テスト同士が影響し合います。

解決策： フィクスチャでトランザクションをロールバックします。

@pytest.fixture
def db_session(db_engine):
    """Isolate database state per test with rollback."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()

    yield session

    session.close()
    transaction.rollback()  # Rollback all changes
    connection.close()

問題：非同期テストが停止する

非同期テストのマークを忘れたり、イベントループの処理が適切でなかったりします。

解決策： 常に @pytest.mark.asyncio を使用し、コルーチンを await します。

# Wrong - test will hang
async def test_async_operation():
    result = await async_function()
    assert result is not None

# Correct
@pytest.mark.asyncio
async def test_async_operation():
    result = await async_function()
    assert result is not None

問題：フィクスチャのスコープに関する問題

変更可能な状態を持つセッションスコープのフィクスチャによって、テストが汚染されます。

解決策： 適切なスコープを使用し、状態をリセットします。

# Problematic
@pytest.fixture(scope="session")
def user_cache():
    return {}  # Shared dict across all tests

# Better
@pytest.fixture(scope="function")
def user_cache():
    return {}  # New dict per test

# Or reset state
@pytest.fixture(scope="session")
def user_cache():
    cache = {}
    yield cache
    cache.clear()  # Clean up after each test

問題：過剰なモック

モックを使いすぎると、テストに意味がなくなります。

解決策： 外部との境界のみをモックします。

# Over-mocked - testing nothing real
def test_ticket_creation(mocker):
    mocker.patch('app.services.create_ticket', return_value=Ticket(id=1))
    ticket = create_ticket(title="Test")
    assert ticket.id == 1  # Just testing the mock

# Better - test actual logic
def test_ticket_creation(db_session):
    ticket = create_ticket_service(db_session, title="Test", priority="high")
    assert ticket.id is not None
    assert ticket.status == "open"
    assert ticket.priority == "high"

問題：テストスイートの実行が遅い

テストの実行に時間がかかります。

解決策： 適切なマーカーを使用し、並列実行します。

# Run only fast tests
pytest -m "not slow"

# Run tests in parallel
pytest -n auto  # Requires pytest-xdist

# Run specific test file
pytest tests/unit/test_validators.py -v

13. PostgreSQL の統合

実際の PostgreSQL を使用したテスト

@pytest.fixture(scope="session")
def postgres_engine():
    """Create PostgreSQL engine with test database."""
    from sqlalchemy import create_engine

    # Connect to default database to create test database
    admin_url = "postgresql://postgres:password@localhost/postgres"
    admin_engine = create_engine(admin_url, isolation_level="AUTOCOMMIT")

    with admin_engine.connect() as conn:
        # Drop and recreate test database
        conn.execute("DROP DATABASE IF EXISTS support_test")
        conn.execute("CREATE DATABASE support_test")

    # Connect to test database
    test_url = "postgresql://postgres:password@localhost/support_test"
    test_engine = create_engine(test_url)

    from app.database import Base
    Base.metadata.create_all(test_engine)

    yield test_engine

    # Cleanup
    test_engine.dispose()
    with admin_engine.connect() as conn:
        conn.execute("DROP DATABASE support_test")
    admin_engine.dispose()

PostgreSQL 固有の機能のテスト

def test_full_text_search(db_session):
    """Test PostgreSQL full-text search on tickets."""
    from sqlalchemy import func

    tickets = [
        Ticket(title="Cannot login", description="Password reset not working"),
        Ticket(title="Slow performance", description="Dashboard loading slowly"),
        Ticket(title="Email issues", description="Not receiving notifications"),
    ]
    db_session.add_all(tickets)
    db_session.commit()

    # Search using PostgreSQL full-text search
    search_term = "login password"
    results = db_session.query(Ticket).filter(
        func.to_tsvector('english', Ticket.title + ' ' + Ticket.description).match(search_term)
    ).all()

    assert len(results) >= 1
    assert any("login" in t.title.lower() for t in results)

def test_jsonb_field_operations(db_session):
    """Test JSONB field operations in PostgreSQL."""
    from sqlalchemy.dialects.postgresql import JSONB

    ticket = Ticket(
        title="JSONB Test",
        metadata_={"tags": ["urgent", "billing"], "source": "email"}
    )
    db_session.add(ticket)
    db_session.commit()

    # Query using JSONB operations
    from sqlalchemy import cast
    results = db_session.query(Ticket).filter(
        Ticket.metadata_['tags'].astext.contains('urgent')
    ).all()

    assert len(results) >= 1
    assert results[0].metadata_["tags"] == ["urgent", "billing"]

14. CI/CD の統合

GitHub Actions の設定

name: Test Suite

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_password
          POSTGRES_DB: support_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install dependencies
        run: |
          pip install --upgrade pip
          pip install -r requirements.txt
          pip install pytest pytest-cov pytest-asyncio pytest-mock

      - name: Run migrations
        env:
          DATABASE_URL: postgresql://test_user:test_password@localhost:5432/support_test
        run: |
          alembic upgrade head

      - name: Run tests
        env:
          DATABASE_URL: postgresql://test_user:test_password@localhost:5432/support_test
        run: |
          pytest -v \
            --cov=app \
            --cov-report=xml \
            --cov-report=term-missing \
            --cov-fail-under=80

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml
          fail_ci_if_error: true

トラブルシューティング

よくある問題

1. インポートエラー： PYTHONPATH にプロジェクトルートが含まれていることを確認するか、pip install -e . を使用してください
2. フィクスチャが見つからない： conftest.py の場所とフィクスチャ名のスペルを確認してください
3. データベースがロックされる： 並行テストには SQLite ではなく PostgreSQL を使用してください
4. 非同期テストがハングする： pytest-asyncio がインストールされ、asyncio_mode が設定されていることを確認してください
5. モックが機能しない： 関数が定義されている場所ではなく、関数が使用されている場所をパッチしてください

まとめ

pytest は、カスタマーサポートシステム向けに包括的なテスト機能を提供します。

• フィクスチャ：再利用可能なセットアップと依存性注入
• パラメーター化：複数のシナリオのテスト
• モック：外部依存関係の分離
• 非同期サポート：最新の非同期アプリケーションへの対応
• カバレッジ：テストの網羅性の測定
• FastAPI の統合：API のテスト
• SQLAlchemy/PostgreSQL：データベースのテスト

このスキルにより、サポートチームは顧客に効果的なサービスを提供する、信頼性の高い、十分にテストされたバックエンドシステムを構築できます。