pytest-coder

Pytest 编码器

核心理念

原则	应用
AAA 模式	每个测试都遵循 Arrange-Act-Assert
行为优于实现	测试代码的行为，而非实现方式
隔离性	测试必须彼此独立
快速测试	模拟 I/O，尽量减少数据库访问
描述性名称	测试名称应说明测试场景
覆盖率	测试正常路径和边缘情况

项目结构

tests/
├── conftest.py          # Shared fixtures
├── unit/                # Unit tests (fast, isolated)
│   ├── test_models.py
│   └── test_services.py
├── integration/         # Integration tests (real dependencies)
│   └── test_api.py
└── fixtures/            # Test data files
    └── sample_data.json

基本模式

基本测试结构

import pytest
from myapp.services import UserService

class TestUserService:
    """Tests for UserService."""

    def test_create_user_with_valid_data(self, user_service):
        # Arrange
        user_data = {"email": "[email protected]", "name": "Test User"}

        # Act
        result = user_service.create(user_data)

        # Assert
        assert result.email == "[email protected]"
        assert result.id is not None

    def test_create_user_with_duplicate_email_raises_error(self, user_service, existing_user):
        # Arrange
        user_data = {"email": existing_user.email, "name": "Another User"}

        # Act & Assert
        with pytest.raises(ValueError, match="Email already exists"):
            user_service.create(user_data)

Fixtures

# conftest.py
import pytest
from myapp.database import get_db
from myapp.services import UserService

@pytest.fixture
def db():
    """Provide a clean database session."""
    session = get_db()
    yield session
    session.rollback()

@pytest.fixture
def user_service(db):
    """Provide UserService instance."""
    return UserService(db)

@pytest.fixture
def sample_user():
    """Provide sample user data."""
    return {"email": "[email protected]", "name": "Test User", "password": "secret123"}

@pytest.fixture
def existing_user(db, sample_user):
    """Create and return an existing user."""
    from myapp.models import User
    user = User(**sample_user)
    db.add(user)
    db.commit()
    return user

参数化测试

import pytest

@pytest.mark.parametrize("input_email,expected_valid", [
    ("[email protected]", True),
    ("[email protected]", True),
    ("invalid-email", False),
    ("missing@domain", False),
    ("", False),
])
def test_email_validation(input_email, expected_valid):
    from myapp.validators import is_valid_email
    assert is_valid_email(input_email) == expected_valid

@pytest.mark.parametrize("status,expected_message", [
    ("pending", "Order is being processed"),
    ("shipped", "Order has been shipped"),
    ("delivered", "Order has been delivered"),
], ids=["pending-status", "shipped-status", "delivered-status"])
def test_order_status_message(status, expected_message):
    from myapp.orders import get_status_message
    assert get_status_message(status) == expected_message

模拟

from unittest.mock import Mock, patch, AsyncMock

def test_send_email_calls_smtp(user_service):
    # Mock external dependency
    with patch("myapp.services.smtp_client") as mock_smtp:
        mock_smtp.send.return_value = True

        user_service.send_welcome_email("[email protected]")

        mock_smtp.send.assert_called_once_with(
            to="[email protected]",
            subject="Welcome!",
        )

def test_payment_processing_handles_failure():
    mock_gateway = Mock()
    mock_gateway.charge.side_effect = PaymentError("Card declined")

    service = PaymentService(gateway=mock_gateway)

    with pytest.raises(PaymentError):
        service.process_payment(amount=100)

异步测试

import pytest

@pytest.mark.asyncio
async def test_async_fetch_user(user_service):
    # Arrange
    user_id = 1

    # Act
    user = await user_service.get_by_id(user_id)

    # Assert
    assert user.id == user_id

@pytest.fixture
async def async_db():
    """Async database session fixture."""
    from myapp.database import async_session
    async with async_session() as session:
        yield session
        await session.rollback()

# Mock async functions
@pytest.mark.asyncio
async def test_async_external_api():
    with patch("myapp.client.fetch_data", new_callable=AsyncMock) as mock_fetch:
        mock_fetch.return_value = {"status": "ok"}

        result = await fetch_and_process()

        assert result["status"] == "ok"

测试异常

import pytest

def test_divide_by_zero_raises_error():
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

def test_invalid_input_raises_with_message():
    with pytest.raises(ValueError, match="must be positive"):
        process_amount(-100)

def test_exception_attributes():
    with pytest.raises(CustomError) as exc_info:
        risky_operation()

    assert exc_info.value.code == "E001"
    assert "failed" in str(exc_info.value)

Fixture 作用域

作用域	生命周期	使用场景
function	每个测试（默认）	大多数 fixtures
class	每个测试类	在类中共享设置
module	每个模块	模块间共享的耗时设置
session	整个测试运行期间	数据库连接、服务器

@pytest.fixture(scope="session")
def database_engine():
    """Create engine once for entire test session."""
    engine = create_engine(TEST_DATABASE_URL)
    yield engine
    engine.dispose()

@pytest.fixture(scope="function")
def db_session(database_engine):
    """Create fresh session per test."""
    connection = database_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    yield session

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

标记

# pytest.ini or pyproject.toml
[tool.pytest.ini_options]
markers = [
    "slow: marks tests as slow",
    "integration: marks integration tests",
    "unit: marks unit tests",
]

# Usage
@pytest.mark.slow
def test_complex_calculation():
    ...

@pytest.mark.integration
def test_database_connection():
    ...

# Run specific markers
# pytest -m "not slow"
# pytest -m "unit"

质量检查清单

• 每个测试都采用 AAA 模式（Arrange-Act-Assert）
• 测试名称具有描述性，能说明测试场景
• 使用 fixtures 处理常见设置
• 使用参数化测试覆盖多种输入
• 使用模拟处理外部依赖
• 测试正常路径
• 测试错误情况
• 覆盖边缘情况
• 异步测试使用 @pytest.mark.asyncio
• 测试之间没有依赖关系
• 覆盖率 >90%

反模式

反模式	不妥之处	修正方法
测试依赖执行顺序	不稳定，难以调试	使用 fixtures，隔离测试
测试实现方式	测试脆弱	测试行为
断言过多	难以定位失败原因	每个测试只做一个断言
不测试错误情况	覆盖不足	明确测试异常
单元测试运行缓慢	反馈缓慢	模拟 I/O，使用内存数据库