Back to Details

pytest-patterns

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Pytest Patterns

Pytest 测试模式

When to Use

使用场景

Activate this skill when:
  • Writing unit tests for service or repository classes
  • Writing integration tests for FastAPI endpoints with httpx.AsyncClient
  • Creating or refactoring pytest fixtures and conftest files
  • Setting up factory_boy factories for test data
  • Testing async code with pytest-asyncio
  • Mocking external services (HTTP APIs, email, queues)
  • Adding parametrized tests for input variations
  • Auditing or improving test coverage
Do NOT use this skill for:
  • Frontend React component or hook tests (use 
    react-testing-patterns
    )
  • E2E browser tests with Playwright (use 
    e2e-testing
    )
  • TDD red-green-refactor workflow enforcement (use 
    tdd-workflow
    )
  • Writing application code (use 
    python-backend-expert
    )
激活此技能的场景：
  • 为服务或仓储类编写单元测试
  • 使用httpx.AsyncClient为FastAPI端点编写集成测试
  • 创建或重构pytest fixture与conftest文件
  • 为测试数据设置factory_boy工厂
  • 使用pytest-asyncio测试异步代码
  • Mock外部服务（HTTP API、邮件、队列）
  • 为输入变体添加参数化测试
  • 审核或提升测试覆盖率
请勿将此技能用于：
  • 前端React组件或Hook测试（请使用
    react-testing-patterns
    ）
  • 使用Playwright的端到端浏览器测试（请使用
    e2e-testing
    ）
  • 强制执行TDD的红-绿-重构工作流（请使用
    tdd-workflow
    ）
  • 编写应用代码（请使用
    python-backend-expert
    ）

Instructions

操作指南

Test Organization

测试组织

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
Naming conventions:
  • Test files: 
    test_<module>.py
  • Test classes: 
    Test<Feature>
    (group related tests, no 
    __init__
    )
  • Test functions: 
    test_<action>_<expected_outcome>
    or 
    test_<scenario>
  • Fixtures: descriptive noun (
    db_session
    , 
    authenticated_client
    , 
    sample_user
    )
Marker conventions:
python
undefined
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
    ）
标记规范：
python
undefined

pyproject.toml

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"

Run subsets: `pytest -m unit`, `pytest -m integration`, `pytest -m "not slow"`.
[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 Architecture

Fixture架构

Conftest Hierarchy

Conftest层级结构

Fixtures cascade: root 
conftest.py
provides shared fixtures; subdirectory conftest files add layer-specific fixtures.
Root conftest (tests/conftest.py):
python
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具有层级继承性：根目录
conftest.py
提供共享Fixture；子目录的conftest文件添加特定层级的Fixture。
根目录conftest（tests/conftest.py）：
python
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 Scopes

Fixture作用域

ScopeUse ForExample
function
(default)		Isolated per-test data
db_session
, 
sample_user
class
Shared across test class
service_instance
module
Shared across test file
seeded_database
session
Shared across entire run
engine
, 
anyio_backend
Rules:
  • Default to 
    function
    scope for data isolation
  • Use 
    session
    scope only for expensive, stateless resources (engine, event loop)
  • Never use 
    session
    scope for mutable data -- tests will interfere with each other
  • Fixtures that yield must clean up (rollback, delete, close)
作用域适用场景示例
function
（默认）		每个测试独立的数据
db_session
, 
sample_user
class
测试类内共享
service_instance
module
测试文件内共享
seeded_database
session
整个测试运行期间共享
engine
, 
anyio_backend
规则：
  • 默认使用
    function
    作用域以保证数据隔离
  • 仅对昂贵、无状态的资源（如引擎、事件循环）使用
    session
    作用域
  • 绝不为可变数据使用
    session
    作用域——测试之间会互相干扰
  • 使用yield的Fixture必须清理（回滚、删除、关闭）

Auth Fixtures

认证Fixture

python
@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}"}
python
@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 Pattern

工厂模式

Use 
factory_boy
for consistent, overridable test data.
python
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"
Usage in tests:
python
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 integration (for integration tests that persist to DB):
python
class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = None  # Set per-test via conftest

    # ... fields same as above
Set session in conftest:
python
@pytest.fixture(autouse=True)
def set_factory_session(db_session):
    UserFactory._meta.sqlalchemy_session = db_session
    OrderFactory._meta.sqlalchemy_session = db_session
使用
factory_boy
生成一致且可覆盖的测试数据。
python
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"
测试中的用法：
python
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集成（用于持久化到数据库的集成测试）：
python
class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = None  # Set per-test via conftest

    # ... fields same as above
在conftest中设置会话：
python
@pytest.fixture(autouse=True)
def set_factory_session(db_session):
    UserFactory._meta.sqlalchemy_session = db_session
    OrderFactory._meta.sqlalchemy_session = db_session

API Integration Tests

API集成测试

Test FastAPI endpoints with 
httpx.AsyncClient
against the real app, but with a test database.
python
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": "new@example.com",
            "display_name": "New User",
        })
        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "new@example.com"
        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
Key patterns:
  • Use 
    authenticated_client
    for protected endpoints, plain 
    client
    for auth testing
  • Assert status code first, then response body
  • Test error paths: 404, 409, 422, 401, 403
  • Test pagination parameters
  • Never assert on exact timestamps or auto-generated IDs (use 
    "id" in data
    )
使用
httpx.AsyncClient
针对真实应用测试FastAPI端点，但使用测试数据库。
python
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": "new@example.com",
            "display_name": "New User",
        })
        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "new@example.com"
        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
    ）

Async Tests

异步测试

With 
asyncio_mode = "auto"
in pyproject.toml, all 
async def test_*
functions run automatically.
python
async def test_async_service_call(db_session):
    service = UserService(db_session)
    user = await service.create_user(email="test@example.com", 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
Common pitfalls:
  • Do NOT mix 
    sync
    and 
    async
    fixtures carelessly -- an async fixture can only be used by async tests
  • Always use 
    pytest-asyncio
    (not 
    anyio
    directly) for consistency
  • If a test hangs, check for un-awaited coroutines or missing 
    async
    keywords
在pyproject.toml中设置
asyncio_mode = "auto"
后，所有
async def test_*
函数会自动运行。
python
async def test_async_service_call(db_session):
    service = UserService(db_session)
    user = await service.create_user(email="test@example.com", 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
    ）以保证一致性
  • 如果测试挂起，检查是否有未await的协程或遗漏的
    async
    关键字

Mocking Strategy

Mock策略

Mock external services -- YES:
python
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()
Mock the database -- NO:
python
undefined
建议Mock外部服务：
python
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()
不建议Mock数据库：
python
undefined

BAD: Mocking the database hides real query issues

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

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="test@example.com", display_name="Test") fetched = await service.get_user(user.id) assert fetched.email == "test@example.com"

**What to mock:**
| Mock | Do Not Mock |
|------|-------------|
| HTTP APIs (use `respx` or `unittest.mock`) | Database queries |
| Email/SMS services | SQLAlchemy sessions |
| File storage (S3, GCS) | Repository methods (in integration tests) |
| Message queues (Redis, RabbitMQ) | Pydantic validation |
| Time/datetime (`freezegun`) | FastAPI dependency injection |
| Random/UUID generation | ORM relationships |

**`respx` for HTTP mocking:**
```python
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"
async def test_user_service(db_session): service = UserService(db_session) user = await service.create_user(email="test@example.com", display_name="Test") fetched = await service.get_user(user.id) assert fetched.email == "test@example.com"

**Mock范围参考：**
| 建议Mock | 不建议Mock |
|------|-------------|
| HTTP API（使用`respx`或`unittest.mock`） | 数据库查询 |
| 邮件/SMS服务 | SQLAlchemy会话 |
| 文件存储（S3、GCS） | 仓储层方法（集成测试中） |
| 消息队列（Redis、RabbitMQ） | Pydantic验证 |
| 时间/日期（`freezegun`） | FastAPI依赖注入 |
| 随机/UUID生成 | ORM关联关系 |

**使用`respx`进行HTTP Mock：**
```python
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"

Parametrized Tests

参数化测试

Use 
@pytest.mark.parametrize
for testing multiple input/output combinations:
python
@pytest.mark.parametrize("email,is_valid", [
    ("user@example.com", True),
    ("user@sub.domain.com", True),
    ("user+tag@example.com", 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)
Parametrize with IDs for readable output:
python
@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):
    ...
Parametrize multiple fixtures:
python
@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
使用
@pytest.mark.parametrize
测试多组输入/输出组合：
python
@pytest.mark.parametrize("email,is_valid", [
    ("user@example.com", True),
    ("user@sub.domain.com", True),
    ("user+tag@example.com", 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的参数化（提升输出可读性）：
python
@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参数化：
python
@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

Coverage Requirements

覆盖率要求

Minimum thresholds:
  • Overall: 80% line coverage
  • Service layer: 90% (critical business logic)
  • Repository layer: 70% (straightforward CRUD)
  • Routes: 80% (all success + primary error paths)
pyproject.toml configuration:
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",
]
Running coverage:
bash
pytest --cov=app --cov-report=term-missing --cov-report=html --cov-fail-under=80
Use 
scripts/check-test-coverage.sh
to automate coverage checks with report output.
最低阈值：
  • 整体覆盖率：80%行覆盖率
  • 服务层：90%（核心业务逻辑）
  • 仓储层：70%（简单CRUD操作）
  • 路由层：80%（所有成功路径+主要错误路径）
pyproject.toml配置：
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",
]
运行覆盖率检查：
bash
pytest --cov=app --cov-report=term-missing --cov-report=html --cov-fail-under=80
使用
scripts/check-test-coverage.sh
自动化覆盖率检查并生成报告。

Test Anti-Patterns

测试反模式

1. Testing implementation details:
python
undefined
1. 测试实现细节：
python
undefined

BAD: Tests internal method calls

BAD: Tests internal method calls

def test_service_calls_repo(mock_repo): service.create_user(data) mock_repo.insert.assert_called_once_with(...)
def test_service_calls_repo(mock_repo): service.create_user(data) mock_repo.insert.assert_called_once_with(...)

GOOD: Tests observable behavior

GOOD: Tests observable behavior

async def test_create_user(db_session): service = UserService(db_session) user = await service.create_user(email="a@b.com", display_name="A") fetched = await service.get_user(user.id) assert fetched is not None

**2. Shared mutable state between tests:**
```python
async def test_create_user(db_session): service = UserService(db_session) user = await service.create_user(email="a@b.com", display_name="A") fetched = await service.get_user(user.id) assert fetched is not None

**2. 测试间共享可变状态：**
```python

BAD: Module-level mutable data

BAD: Module-level mutable data

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

GOOD: Use fixtures with function scope

GOOD: Use fixtures with function scope

@pytest.fixture def users(): return [UserFactory()]

**3. Overly broad assertions:**
```python
@pytest.fixture def users(): return [UserFactory()]

**3. 断言范围过宽：**
```python

BAD

BAD

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

GOOD

GOOD

assert response.status_code == 200 data = response.json() assert data["email"] == "test@example.com" assert data["role"] == "member"

**4. Missing error path tests:**
Every endpoint should have tests for at least: success, not found, validation error, and unauthorized.
assert response.status_code == 200 data = response.json() assert data["email"] == "test@example.com" assert data["role"] == "member"

**4. 遗漏错误路径测试：**
每个端点至少应测试：成功路径、资源不存在、验证错误、未授权这几种场景。

Examples

示例

See 
references/conftest-template.py
for production conftest setup. See 
references/factory-template.py
for factory_boy patterns. See 
references/api-test-template.py
for API integration test patterns. See 
references/service-test-template.py
for service unit test patterns. See 
references/integration-test-template.py
for full integration test patterns.
查看
references/conftest-template.py
获取生产环境级别的conftest配置。 查看
references/factory-template.py
获取factory_boy使用模式。 查看
references/api-test-template.py
获取API集成测试模式。 查看
references/service-test-template.py
获取服务层单元测试模式。 查看
references/integration-test-template.py
获取完整集成测试模式。