python-testing

快速参考

pytest 命令	用途
pytest	运行所有测试
pytest -v	输出详细信息
pytest -k "name"	运行与模式匹配的测试
pytest -x	首次失败时停止
pytest --lf	运行上次失败的测试
pytest -n auto	并行执行

Fixture 作用域	持续时间
function	每个测试（默认）
class	每个测试类
module	每个测试文件
session	整个测试运行期间

Mock 模式	代码
Patch 函数	mocker.patch("module.func")
返回值	mock.return_value = {...}
副作用	mock.side_effect = [a, b, exc]
断言已调用	mock.assert_called_once()

标记	使用场景
@pytest.mark.asyncio	异步测试
@pytest.mark.parametrize	多个输入
@pytest.mark.skip	跳过测试
@pytest.mark.xfail	预期失败

何时使用此技能

用于测试 Python 代码：

• 使用 fixture 编写 pytest 测试
• Mock 外部依赖
• 测试异步代码
• 设置代码覆盖率
• 测试 FastAPI 应用

相关技能：

• 关于 FastAPI：参见 python-fastapi
• 关于异步模式：参见 python-asyncio
• 关于 CI/CD：参见 python-github-actions

---

Python 测试最佳实践（2025）

概述

现代 Python 测试以 pytest 为事实标准，并配合使用覆盖率、Mock 和异步测试工具。

pytest 基础

安装

# With uv
uv add --dev pytest pytest-cov pytest-asyncio pytest-xdist

# With pip
pip install pytest pytest-cov pytest-asyncio pytest-xdist

基本测试结构

# tests/test_calculator.py
import pytest
from mypackage.calculator import add, divide

def test_add_positive_numbers():
    assert add(2, 3) == 5

def test_add_negative_numbers():
    assert add(-2, -3) == -5

def test_add_mixed_numbers():
    assert add(-2, 3) == 1

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

def test_divide_result():
    result = divide(10, 2)
    assert result == pytest.approx(5.0)

运行测试

# Run all tests
pytest

# Verbose output
pytest -v

# Run specific file
pytest tests/test_calculator.py

# Run specific test
pytest tests/test_calculator.py::test_add_positive_numbers

# Run tests matching pattern
pytest -k "add"

# Stop on first failure
pytest -x

# Run last failed tests
pytest --lf

# Parallel execution
pytest -n auto

Fixture

基本 Fixture

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def sample_user():
    """Provides a sample user dictionary."""
    return {
        "name": "John Doe",
        "email": "[email protected]",
        "age": 30,
    }

@pytest.fixture
def db_session():
    """Provides a database session that rolls back after test."""
    engine = create_engine("sqlite:///:memory:")
    Session = sessionmaker(bind=engine)
    session = Session()

    yield session

    session.rollback()
    session.close()

Fixture 作用域

@pytest.fixture(scope="function")  # Default: new for each test
def per_test_fixture():
    return create_resource()

@pytest.fixture(scope="class")  # Shared within test class
def per_class_fixture():
    return create_expensive_resource()

@pytest.fixture(scope="module")  # Shared within test module
def per_module_fixture():
    return create_very_expensive_resource()

@pytest.fixture(scope="session")  # Shared across entire test session
def per_session_fixture():
    resource = create_global_resource()
    yield resource
    cleanup_global_resource(resource)

参数化 Fixture

@pytest.fixture(params=["sqlite", "postgresql", "mysql"])
def database_type(request):
    return request.param

def test_with_multiple_databases(database_type):
    # This test runs 3 times, once for each database type
    db = create_connection(database_type)
    assert db.is_connected()

参数化测试

基本参数化

import pytest

@pytest.mark.parametrize("input,expected", [
    (1, 1),
    (2, 4),
    (3, 9),
    (4, 16),
])
def test_square(input, expected):
    assert input ** 2 == expected

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3),
    (-1, 1, 0),
    (0, 0, 0),
    (100, 200, 300),
])
def test_add(a, b, expected):
    assert add(a, b) == expected

使用 ID 参数化

@pytest.mark.parametrize("input,expected", [
    pytest.param(1, 1, id="one"),
    pytest.param(2, 4, id="two"),
    pytest.param(3, 9, id="three"),
])
def test_square_with_ids(input, expected):
    assert input ** 2 == expected

# Run with: pytest -v
# Shows: test_square_with_ids[one] PASSED

组合参数化

@pytest.mark.parametrize("x", [1, 2])
@pytest.mark.parametrize("y", [3, 4])
def test_combinations(x, y):
    # Runs 4 times: (1,3), (1,4), (2,3), (2,4)
    assert x + y in [4, 5, 6]

Mock

使用 pytest-mock

from unittest.mock import MagicMock, patch

def test_with_mock(mocker):
    # Mock a method
    mock_api = mocker.patch("mypackage.api.fetch_data")
    mock_api.return_value = {"status": "success"}

    result = process_data()
    assert result["status"] == "success"
    mock_api.assert_called_once()

def test_mock_return_values(mocker):
    # Different return values for successive calls
    mock_func = mocker.patch("mypackage.service.get_item")
    mock_func.side_effect = [
        {"id": 1},
        {"id": 2},
        ValueError("Not found"),
    ]

    assert get_item(1) == {"id": 1}
    assert get_item(2) == {"id": 2}

    with pytest.raises(ValueError):
        get_item(3)

上下文管理器 Mock

def test_mock_context_manager(mocker):
    mock_open = mocker.patch("builtins.open", mocker.mock_open(read_data="file content"))

    with open("test.txt") as f:
        content = f.read()

    assert content == "file content"
    mock_open.assert_called_once_with("test.txt")

Mock 类

def test_mock_class(mocker):
    MockUser = mocker.patch("mypackage.models.User")
    mock_instance = MockUser.return_value
    mock_instance.name = "Test User"
    mock_instance.save.return_value = True

    user = create_user("Test User")
    assert user.name == "Test User"
    mock_instance.save.assert_called_once()

异步测试

pytest-asyncio

import pytest
import asyncio
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_function():
    result = await async_fetch_data()
    assert result is not None

@pytest.mark.asyncio
async def test_async_with_fixture(async_client: AsyncClient):
    response = await async_client.get("/api/users")
    assert response.status_code == 200

# Fixture for async client
@pytest.fixture
async def async_client():
    async with AsyncClient(base_url="http://test") as client:
        yield client

异步 Fixture

@pytest.fixture
async def async_db_session():
    engine = create_async_engine("postgresql+asyncpg://...")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async_session = sessionmaker(engine, class_=AsyncSession)

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

测试 FastAPI

import pytest
from httpx import AsyncClient, ASGITransport
from myapp.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        yield client

@pytest.mark.asyncio
async def test_read_main(client: AsyncClient):
    response = await client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

@pytest.mark.asyncio
async def test_create_item(client: AsyncClient):
    response = await client.post(
        "/items/",
        json={"name": "Test Item", "price": 10.0},
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Test Item"

覆盖率

配置

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = [
    "-ra",
    "-q",
    "--strict-markers",
    "--cov=src",
    "--cov-report=term-missing",
    "--cov-report=html",
    "--cov-fail-under=80",
]

[tool.coverage.run]
branch = true
source = ["src"]
omit = ["*/tests/*", "*/__init__.py"]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise NotImplementedError",
    "if TYPE_CHECKING:",
    "if __name__ == .__main__.:",
]

运行并收集覆盖率

# Generate coverage report
pytest --cov=src --cov-report=html

# Check coverage threshold
pytest --cov=src --cov-fail-under=80

# Coverage for specific paths
pytest --cov=src/mypackage --cov-report=term-missing

测试组织

推荐结构

tests/
├── conftest.py          # Shared fixtures
├── unit/                # Unit tests
│   ├── conftest.py      # Unit test fixtures
│   ├── test_models.py
│   └── test_utils.py
├── integration/         # Integration tests
│   ├── conftest.py
│   └── test_api.py
├── e2e/                 # End-to-end tests
│   └── test_workflows.py
└── fixtures/            # Test data
    ├── users.json
    └── products.json

Conftest 层级结构

# tests/conftest.py - Shared across all tests
@pytest.fixture
def app_config():
    return {"debug": True, "testing": True}

# tests/unit/conftest.py - Unit test specific
@pytest.fixture
def mock_database(mocker):
    return mocker.MagicMock()

# tests/integration/conftest.py - Integration test specific
@pytest.fixture
def real_database():
    db = create_test_database()
    yield db
    db.cleanup()

标记

内置标记

import pytest

@pytest.mark.skip(reason="Not implemented yet")
def test_future_feature():
    pass

@pytest.mark.skipif(sys.platform == "win32", reason="Unix only")
def test_unix_specific():
    pass

@pytest.mark.xfail(reason="Known bug, see issue #123")
def test_known_failure():
    assert False

@pytest.mark.slow
def test_slow_operation():
    time.sleep(10)

自定义标记

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

# Run only slow tests
pytest -m slow

# Skip slow tests
pytest -m "not slow"

# Combine markers
pytest -m "integration and not slow"

基于属性的测试

Hypothesis

from hypothesis import given, strategies as st

@given(st.integers(), st.integers())
def test_addition_commutative(a, b):
    assert add(a, b) == add(b, a)

@given(st.lists(st.integers()))
def test_sort_idempotent(items):
    sorted_once = sorted(items)
    sorted_twice = sorted(sorted_once)
    assert sorted_once == sorted_twice

@given(st.text(min_size=1))
def test_string_reverse_twice(s):
    assert s[::-1][::-1] == s

# Custom strategy
@given(st.builds(
    User,
    name=st.text(min_size=1, max_size=100),
    email=st.emails(),
    age=st.integers(min_value=0, max_value=150),
))
def test_user_validation(user):
    assert user.is_valid()

快照测试

syrupy

from syrupy.assertion import SnapshotAssertion

def test_api_response(snapshot: SnapshotAssertion):
    response = get_api_response()
    assert response == snapshot

def test_html_output(snapshot: SnapshotAssertion):
    html = render_template("user.html", user=sample_user)
    assert html == snapshot(extension_class=HTMLSnapshotExtension)

最佳实践

1. 测试命名

# Clear, descriptive names
def test_user_creation_with_valid_email_succeeds():
    ...

def test_user_creation_with_invalid_email_raises_validation_error():
    ...

def test_empty_cart_returns_zero_total():
    ...

2. 准备-执行-断言

def test_order_total():
    # Arrange
    order = Order()
    order.add_item(Item(price=10.00))
    order.add_item(Item(price=20.00))

    # Act
    total = order.calculate_total()

    # Assert
    assert total == 30.00

3. 每个测试只做一项断言（通常）

# Prefer multiple focused tests
def test_user_has_correct_name():
    user = create_user("John")
    assert user.name == "John"

def test_user_has_default_role():
    user = create_user("John")
    assert user.role == "member"

# Over one test with many assertions
def test_user_creation():  # Less focused
    user = create_user("John")
    assert user.name == "John"
    assert user.role == "member"
    assert user.is_active
    assert user.created_at is not None

4. 使用 Fixture 进行设置

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

def test_user_can_post(authenticated_user):
    post = authenticated_user.create_post("Hello")
    assert post.author == authenticated_user

# Avoid: Repeated setup in each test
def test_user_can_post_bad():
    user = User(name="Test", email="[email protected]")
    db_session.add(user)  # Repeated in every test
    db_session.commit()
    post = user.create_post("Hello")
    assert post.author == user

更多参考资料

如需了解本指南之外适用于生产环境的 Fixture 模式，请参阅：

• Pytest Fixtures Cookbook - SQLAlchemy 异步会话、使用 Docker 的 PostgreSQL、用户工厂、通用模型工厂、身份验证 Fixture、外部 API Mock、冻结时间、环境 Fixture、文件 Fixture、参数化 Fixture、清理模式、Redis Fixture