pytest-patterns

Pytest 测试模式

适用场景

在以下场景中启用此技能：

• 为服务类或仓库类编写单元测试
• 使用 httpx.AsyncClient 为 FastAPI 端点编写集成测试
• 创建或重构 pytest fixture 和 conftest 文件
• 设置 factory_boy 工厂来创建测试数据
• 使用 pytest-asyncio 测试异步代码
• 模拟外部服务（HTTP API、邮件、队列）
• 为不同输入编写参数化测试
• 审查或提升测试覆盖率

以下场景不要使用此技能：

• 测试前端 React 组件或 Hook（使用 react-testing-patterns）
• 使用 Playwright 编写端到端浏览器测试（使用 e2e-testing）
• 强制执行 TDD 红—绿—重构工作流（使用 tdd-workflow）
• 编写应用代码（使用 python-backend-expert）

说明

测试组织

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

命名约定：

• 测试文件：test_<module>.py
• 测试类：Test<Feature>（归类相关测试，不要定义 __init__）
• 测试函数：test_<action>_<expected_outcome> 或 test_<scenario>
• Fixture：使用描述性名词（db_session、authenticated_client、sample_user）

标记约定：

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

运行部分测试：pytest -m unit、pytest -m integration、pytest -m "not slow"。

Fixture 架构

Conftest 层级

Fixture 会逐层级联：根目录下的 conftest.py 提供共享 fixture；子目录中的 conftest 文件则添加特定层级的 fixture。

根目录 conftest（tests/conftest.py）：

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

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

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

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

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

Fixture 作用域

作用域	用途	示例
function（默认）	每个测试隔离的数据	db_session、sample_user
class	在测试类内共享	service_instance
module	在测试文件内共享	seeded_database
session	在整个测试运行期间共享	engine、anyio_backend

规则：

• 默认使用 function 作用域以隔离数据
• 仅对开销较大且无状态的资源（engine、事件循环）使用 session 作用域
• 切勿对可变数据使用 session 作用域——测试之间会相互干扰
• 使用 yield 的 fixture 必须执行清理（回滚、删除、关闭）

身份验证 Fixture

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

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

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

工厂模式

使用 factory_boy 创建一致且可覆盖的测试数据。

import factory
from app.models import User, Order

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

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

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

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

在测试中的用法：

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

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

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

SQLAlchemy 集成（用于将数据持久化到数据库的集成测试）：

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

    # ... fields same as above

在 conftest 中设置 session：

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

API 集成测试

使用 httpx.AsyncClient 针对真实应用测试 FastAPI 端点，但使用测试数据库。

import pytest
from httpx import AsyncClient

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

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

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

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

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

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

关键模式：

• 受保护的端点使用 authenticated_client，测试身份验证时使用普通的 client
• 先断言状态码，再断言响应正文
• 测试错误路径：404、409、422、401、403
• 测试分页参数
• 切勿断言精确的时间戳或自动生成的 ID（使用 "id" in data）

异步测试

在 pyproject.toml 中设置 asyncio_mode = "auto" 后，所有 async def test_* 函数都会自动运行。

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

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

常见问题：

• 不要随意混用 sync 和 async fixture——异步 fixture 只能用于异步测试
• 始终使用 pytest-asyncio（不要直接使用 anyio），以保持一致
• 如果测试挂起，请检查是否有未等待的协程或缺少 async 关键字

模拟策略

模拟外部服务——是：

from unittest.mock import AsyncMock, patch

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

模拟数据库——否：

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

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

模拟对象：

模拟	不要模拟
HTTP API（使用 respx 或 unittest.mock）	数据库查询
邮件/SMS 服务	SQLAlchemy session
文件存储（S3、GCS）	仓库方法（在集成测试中）
消息队列（Redis、RabbitMQ）	Pydantic 验证
时间/datetime（freezegun）	FastAPI 依赖注入
随机数/UUID 生成	ORM 关系

使用 respx 模拟 HTTP：

import respx
from httpx import Response

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

参数化测试

使用 @pytest.mark.parametrize 测试多种输入/输出组合：

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

使用 ID 进行参数化，以便清晰地查看输出：

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

对多个 fixture 进行参数化：

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

覆盖率要求

最低阈值：

• 整体：行覆盖率达到 80%
• 服务层：达到 90%（关键业务逻辑）
• 仓储层：达到 70%（简单的 CRUD）
• 路由：达到 80%（所有成功路径和主要错误路径）

pyproject.toml 配置：

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

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

运行覆盖率检查：

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

使用 scripts/check-test-coverage.sh 自动执行覆盖率检查并输出报告。

测试反模式

1. 测试实现细节：

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

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

2. 测试之间共享可变状态：

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

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

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

3. 断言范围过宽：

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

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

4. 缺少错误路径测试： 每个端点都应至少包含以下情况的测试：成功、未找到、验证错误和未授权。

示例

有关生产环境的 conftest 设置，请参阅 references/conftest-template.py。 有关 factory_boy 模式，请参阅 references/factory-template.py。 有关 API 集成测试模式，请参阅 references/api-test-template.py。 有关服务单元测试模式，请参阅 references/service-test-template.py。 有关完整集成测试模式，请参阅 references/integration-test-template.py。