pytest

pytest——面向客户支持的高级 Python 单元测试

概述

pytest 是 Python 应用的行业标准测试框架，提供强大功能，帮助构建全面、易维护且可扩展的测试套件。本技能专注于客户支持技术赋能，提供测试后端支持系统、工单平台、知识库和客户数据平台的模式与实践。

客户支持系统具有关键任务属性，因此需要严格测试。停机或错误会直接影响客户满意度、客服人员的工作效率和业务运营。本技能全面介绍如何使用 pytest 测试支持系统的各个方面。

pytest 为何适用于客户支持系统

独特需求

客户支持应用有特定的测试需求：

1. 高可靠性：支持系统是关键任务系统；故障会直接影响客户体验
2. 复杂的数据关系：工单、客户、客服人员、评论、附件和知识文章
3. 外部集成：电子邮件服务（SendGrid、AWS SES）、CRM 系统（Salesforce、HubSpot）、支付处理服务
4. 异步操作：后台任务、电子邮件队列、通知服务、Webhook 投递
5. 数据验证：严格验证客户个人身份信息（PII）、工单优先级、SLA 要求和升级规则
6. 多租户：不同客户组织或工作区之间的隔离
7. API 优先设计：需要全面覆盖端点的 RESTful API
8. 实时功能：WebSocket 连接、实时聊天、工单实时更新
9. 合规性：GDPR、CCPA、数据保留政策、审计日志记录

pytest 的优势

pytest 通过以下方式满足这些需求：

• 强大的 Fixture 系统：管理复杂的数据库设置、API 客户端和测试数据
• 参数化：高效测试多种场景（不同工单类型、优先级和用户角色）
• 丰富的插件生态：用于异步测试的 pytest-asyncio、用于模拟的 pytest-mock、用于覆盖率的 pytest-cov
• 出色的集成能力：与 FastAPI、SQLAlchemy、PostgreSQL、Pydantic 无缝协作
• 清晰的输出：易读的测试结果和全面的错误报告
• 可扩展性：既能处理小型测试套件，也能通过并行执行处理数千项测试
• 灵活性：在一个框架中支持单元测试、集成测试和端到端测试

核心能力

1. Fixtures 和依赖注入

Fixtures 是 pytest 的杀手级功能，可提供可复用的设置/清理逻辑和依赖注入。对于客户支持系统，Fixtures 用于管理数据库、API 客户端、测试数据和外部服务模拟对象。

基本 Fixtures

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"]

Fixture 作用域

使用作用域控制 Fixture 的创建和销毁时机：

• function（默认）：每个测试函数创建一次，测试完成后销毁
• class：每个测试类创建一次，供所有方法共享
• module：每个模块文件创建一次，供文件中的所有测试共享
• package：每个包创建一次，供包中的所有测试共享
• session：每个完整测试会话创建一次，供所有测试共享

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()

Autouse Fixtures

无需显式请求即可自动运行的 Fixtures：

@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)

Fixture 依赖和组合

Fixtures 可以依赖其他 Fixtures，从而形成依赖链：

@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

工厂 Fixtures

创建返回工厂函数的 Fixtures，以便灵活创建测试数据：

@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

对 Fixtures 进行参数化

@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. 异步测试

现代客户支持系统使用异步操作来提升性能和可扩展性。测试异步代码需要特殊处理，以正确等待协程完成。

基本异步测试

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. 每个测试只测试一件事：

# 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 并等待协程完成。

# 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. 未找到 fixture： 检查 conftest.py 的位置和 fixture 名称拼写
3. 数据库锁定： 并发测试时使用 PostgreSQL，而不是 SQLite
4. 异步测试挂起： 确认已安装 pytest-asyncio，并已配置 asyncio_mode
5. Mock 不起作用： 应在函数被使用的位置打补丁，而不是函数定义的位置

总结

pytest 为客户支持系统提供了全面的测试能力：

• Fixtures：用于可重复使用的设置和依赖注入
• 参数化：用于测试多种场景
• Mock：用于隔离外部依赖
• 异步支持：适用于现代异步应用
• 覆盖率：用于衡量测试完整性
• FastAPI 集成：用于 API 测试
• SQLAlchemy/PostgreSQL：用于数据库测试

这项技能使支持团队能够构建可靠、经过充分测试的后端系统，从而有效地服务客户。