python-testing

Python 测试模式

适用于 Python 应用程序的全面测试策略，使用 pytest、TDD 方法和最佳实践。

何时激活

• 编写新的 Python 代码时（遵循 TDD：红、绿、重构）
• 为 Python 项目设计测试套件时
• 审查 Python 测试覆盖率时
• 搭建测试基础设施时

核心测试理念

测试驱动开发 (TDD)

始终遵循 TDD 周期：

1. RED：为期望的行为编写一个失败的测试
2. GREEN：编写最少的代码使测试通过
3. REFACTOR：在保持测试通过的情况下改进代码

# 步骤 1：编写失败的测试 (RED)
def test_add_numbers():
    result = add(2, 3)
    assert result == 5

# 步骤 2：编写最少的实现 (GREEN)
def add(a, b):
    return a + b

# 步骤 3：如果需要，重构 (REFACTOR)

覆盖率要求

• 目标：80% 以上的代码覆盖率
• 关键路径：需要 100% 的覆盖率
• 使用 pytest --cov 来测量覆盖率

pytest --cov=mypackage --cov-report=term-missing --cov-report=html

pytest 基础

基本测试结构

import pytest

def test_addition():
    """测试基本加法。"""
    assert 2 + 2 == 4

def test_string_uppercase():
    """测试字符串大写。"""
    text = "hello"
    assert text.upper() == "HELLO"

def test_list_append():
    """测试列表添加。"""
    items = [1, 2, 3]
    items.append(4)
    assert 4 in items
    assert len(items) == 4

断言

# 相等
assert result == expected

# 不相等
assert result != unexpected

# 真值
assert result  # 真值
assert not result  # 假值
assert result is True  # 严格为 True
assert result is False  # 严格为 False
assert result is None  # 严格为 None

# 成员运算
assert item in collection
assert item not in collection

# 比较
assert result > 0
assert 0 <= result <= 100

# 类型检查
assert isinstance(result, str)

# 异常测试（推荐方法）
with pytest.raises(ValueError):
    raise ValueError("错误消息")

# 检查异常消息
with pytest.raises(ValueError, match="无效输入"):
    raise ValueError("无效输入")

# 检查异常属性
with pytest.raises(ValueError) as exc_info:
    raise ValueError("错误消息")
assert str(exc_info.value) == "错误消息"

Fixtures

基本 Fixture 使用

import pytest

@pytest.fixture
def sample_data():
    """提供样本数据的 Fixture。"""
    return {"name": "Alice", "age": 30}

def test_sample_data(sample_data):
    """使用 Fixture 进行测试。"""
    assert sample_data["name"] == "Alice"
    assert sample_data["age"] == 30

带有 Setup/Teardown 的 Fixture

@pytest.fixture
def database():
    """带有 setup 和 teardown 的 Fixture。"""
    # Setup
    db = Database(":memory:")
    db.create_tables()
    db.insert_test_data()

    yield db  # 提供给测试

    # Teardown
    db.close()

def test_database_query(database):
    """测试数据库操作。"""
    result = database.query("SELECT * FROM users")
    assert len(result) > 0

Fixture 作用域

# 函数作用域（默认）- 每个测试都运行
@pytest.fixture
def temp_file():
    with open("temp.txt", "w") as f:
        yield f
    os.remove("temp.txt")

# 模块作用域 - 每个模块运行一次
@pytest.fixture(scope="module")
def module_db():
    db = Database(":memory:")
    db.create_tables()
    yield db
    db.close()

# 会话作用域 - 每个测试会话运行一次
@pytest.fixture(scope="session")
def shared_resource():
    resource = ExpensiveResource()
    yield resource
    resource.cleanup()

带参数的 Fixture

@pytest.fixture(params=[1, 2, 3])
def number(request):
    """参数化的 Fixture。"""
    return request.param

def test_numbers(number):
    """测试运行 3 次，每个参数一次。"""
    assert number > 0

使用多个 Fixture

@pytest.fixture
def user():
    return User(id=1, name="Alice")

@pytest.fixture
def admin():
    return User(id=2, name="Admin", role="admin")

def test_user_admin_interaction(user, admin):
    """使用多个 Fixture 进行测试。"""
    assert admin.can_manage(user)

自动使用的 Fixture

@pytest.fixture(autouse=True)
def reset_config():
    """在每个测试前自动运行。"""
    Config.reset()
    yield
    Config.cleanup()

def test_without_fixture_call():
    # reset_config 自动运行
    assert Config.get_setting("debug") is False

用于共享 Fixture 的 Conftest.py

# tests/conftest.py
import pytest

@pytest.fixture
def client():
    """所有测试的共享 Fixture。"""
    app = create_app(testing=True)
    with app.test_client() as client:
        yield client

@pytest.fixture
def auth_headers(client):
    """为 API 测试生成认证头。"""
    response = client.post("/api/login", json={
        "username": "test",
        "password": "test"
    })
    token = response.json["token"]
    return {"Authorization": f"Bearer {token}"}

参数化

基本参数化

@pytest.mark.parametrize("input,expected", [
    ("hello", "HELLO"),
    ("world", "WORLD"),
    ("PyThOn", "PYTHON"),
])
def test_uppercase(input, expected):
    """测试运行 3 次，使用不同的输入。"""
    assert input.upper() == expected

多个参数

@pytest.mark.parametrize("a,b,expected", [
    (2, 3, 5),
    (0, 0, 0),
    (-1, 1, 0),
    (100, 200, 300),
])
def test_add(a, b, expected):
    """使用多个输入测试加法。"""
    assert add(a, b) == expected

带 ID 的参数化

@pytest.mark.parametrize("input,expected", [
    ("[email protected]", True),
    ("invalid", False),
    ("@no-domain.com", False),
], ids=["valid-email", "missing-at", "missing-domain"])
def test_email_validation(input, expected):
    """使用可读的测试 ID 测试电子邮件验证。"""
    assert is_valid_email(input) is expected

参数化的 Fixture

@pytest.fixture(params=["sqlite", "postgresql", "mysql"])
def db(request):
    """针对多个数据库后端进行测试。"""
    if request.param == "sqlite":
        return Database(":memory:")
    elif request.param == "postgresql":
        return Database("postgresql://localhost/test")
    elif request.param == "mysql":
        return Database("mysql://localhost/test")

def test_database_operations(db):
    """测试运行 3 次，每个数据库一次。"""
    result = db.query("SELECT 1")
    assert result is not None

标记和测试选择

自定义标记

# 标记慢速测试
@pytest.mark.slow
def test_slow_operation():
    time.sleep(5)

# 标记集成测试
@pytest.mark.integration
def test_api_integration():
    response = requests.get("https://api.example.com")
    assert response.status_code == 200

# 标记单元测试
@pytest.mark.unit
def test_unit_logic():
    assert calculate(2, 3) == 5

运行特定测试

# 只运行快速测试
pytest -m "not slow"

# 只运行集成测试
pytest -m integration

# 运行集成或慢速测试
pytest -m "integration or slow"

# 运行标记为单元但非慢速的测试
pytest -m "unit and not slow"

在 pytest.ini 中配置标记

[pytest]
markers =
    slow: 将测试标记为慢速
    integration: 将测试标记为集成测试
    unit: 将测试标记为单元测试
    django: 将测试标记为需要 Django

模拟和修补

模拟函数

from unittest.mock import patch, Mock

@patch("mypackage.external_api_call")
def test_with_mock(api_call_mock):
    """使用模拟的外部 API 进行测试。"""
    api_call_mock.return_value = {"status": "success"}

    result = my_function()

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

模拟返回值

@patch("mypackage.Database.connect")
def test_database_connection(connect_mock):
    """使用模拟的数据库连接进行测试。"""
    connect_mock.return_value = MockConnection()

    db = Database()
    db.connect()

    connect_mock.assert_called_once_with("localhost")

模拟异常

@patch("mypackage.api_call")
def test_api_error_handling(api_call_mock):
    """使用模拟的异常测试错误处理。"""
    api_call_mock.side_effect = ConnectionError("网络错误")

    with pytest.raises(ConnectionError):
        api_call()

    api_call_mock.assert_called_once()

模拟上下文管理器

@patch("builtins.open", new_callable=mock_open)
def test_file_reading(mock_file):
    """使用模拟的 open 测试文件读取。"""
    mock_file.return_value.read.return_value = "文件内容"

    result = read_file("test.txt")

    mock_file.assert_called_once_with("test.txt", "r")
    assert result == "文件内容"

使用 Autospec

@patch("mypackage.DBConnection", autospec=True)
def test_autospec(db_mock):
    """使用 autospec 捕获 API 误用。"""
    db = db_mock.return_value
    db.query("SELECT * FROM users")

    # 如果 DBConnection 没有 query 方法，这将失败
    db_mock.assert_called_once()

模拟类实例

class TestUserService:
    @patch("mypackage.UserRepository")
    def test_create_user(self, repo_mock):
        """使用模拟的仓库测试用户创建。"""
        repo_mock.return_value.save.return_value = User(id=1, name="Alice")

        service = UserService(repo_mock.return_value)
        user = service.create_user(name="Alice")

        assert user.name == "Alice"
        repo_mock.return_value.save.assert_called_once()

模拟属性

@pytest.fixture
def mock_config():
    """创建一个带有属性的模拟。"""
    config = Mock()
    type(config).debug = PropertyMock(return_value=True)
    type(config).api_key = PropertyMock(return_value="test-key")
    return config

def test_with_mock_config(mock_config):
    """使用模拟的配置属性进行测试。"""
    assert mock_config.debug is True
    assert mock_config.api_key == "test-key"

测试异步代码

使用 pytest-asyncio 进行异步测试

import pytest

@pytest.mark.asyncio
async def test_async_function():
    """测试异步函数。"""
    result = await async_add(2, 3)
    assert result == 5

@pytest.mark.asyncio
async def test_async_with_fixture(async_client):
    """使用异步 fixture 进行异步测试。"""
    response = await async_client.get("/api/users")
    assert response.status_code == 200

异步 Fixture

@pytest.fixture
async def async_client():
    """提供异步测试客户端的异步 Fixture。"""
    app = create_app()
    async with app.test_client() as client:
        yield client

@pytest.mark.asyncio
async def test_api_endpoint(async_client):
    """使用异步 Fixture 进行测试。"""
    response = await async_client.get("/api/data")
    assert response.status_code == 200

模拟异步函数

@pytest.mark.asyncio
@patch("mypackage.async_api_call")
async def test_async_mock(api_call_mock):
    """使用模拟测试异步函数。"""
    api_call_mock.return_value = {"status": "ok"}

    result = await my_async_function()

    api_call_mock.assert_awaited_once()
    assert result["status"] == "ok"

测试异常

测试预期异常

def test_divide_by_zero():
    """测试除以零会引发 ZeroDivisionError。"""
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

def test_custom_exception():
    """测试带有消息的自定义异常。"""
    with pytest.raises(ValueError, match="无效输入"):
        validate_input("无效")

测试异常属性

def test_exception_with_details():
    """测试带有自定义属性的异常。"""
    with pytest.raises(CustomError) as exc_info:
        raise CustomError("错误", code=400)

    assert exc_info.value.code == 400
    assert "错误" in str(exc_info.value)

测试副作用

测试文件操作

import tempfile
import os

def test_file_processing():
    """使用临时文件测试文件处理。"""
    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f:
        f.write("测试内容")
        temp_path = f.name

    try:
        result = process_file(temp_path)
        assert result == "processed: 测试内容"
    finally:
        os.unlink(temp_path)

使用 pytest 的 tmp_path Fixture 进行测试

def test_with_tmp_path(tmp_path):
    """使用 pytest 内置临时路径 Fixture 进行测试。"""
    test_file = tmp_path / "test.txt"
    test_file.write_text("你好世界")

    result = process_file(str(test_file))
    assert result == "你好世界"
    # tmp_path 自动清理

使用 tmpdir Fixture 进行测试

def test_with_tmpdir(tmpdir):
    """使用 pytest 的 tmpdir Fixture 进行测试。"""
    test_file = tmpdir.join("test.txt")
    test_file.write("数据")

    result = process_file(str(test_file))
    assert result == "数据"

测试组织

目录结构

tests/
├── conftest.py                 # 共享的 Fixture
├── __init__.py
├── unit/                       # 单元测试
│   ├── __init__.py
│   ├── test_models.py
│   ├── test_utils.py
│   └── test_services.py
├── integration/                # 集成测试
│   ├── __init__.py
│   ├── test_api.py
│   └── test_database.py
└── e2e/                        # 端到端测试
    ├── __init__.py
    └── test_user_flow.py

测试类

class TestUserService:
    """在类中组织相关的测试。"""

    @pytest.fixture(autouse=True)
    def setup(self):
        """Setup 在此类中的每个测试之前运行。"""
        self.service = UserService()

    def test_create_user(self):
        """测试用户创建。"""
        user = self.service.create_user("Alice")
        assert user.name == "Alice"

    def test_delete_user(self):
        """测试用户删除。"""
        self.service.create_user("Bob")
        result = self.service.delete_user(1)
        assert result is True

插件集成

# 在 conftest.py 或 pytest.ini 中
# pytest.ini
[pytest]
plugins =
    pytest-cov
    pytest-mock
    pytest-asyncio

# conftest.py
pytest_plugins = [
    "pytest_mock",
    "pytest_asyncio",
]

运行测试

常用命令

# 运行所有测试
pytest

# 运行详细模式
pytest -v

# 运行特定测试文件
pytest tests/test_models.py

# 运行特定测试函数
pytest tests/test_models.py::test_create_user

# 运行特定测试类
pytest tests/test_models.py::TestUserService

# 运行匹配名称的测试
pytest -k "user or admin"

# 激活断点调试
pytest --pdb

# 显示前 3 个失败回溯
pytest --tb=short

# 在失败时运行测试
pytest --lf

# 隐藏输出
pytest --no-header -q

配置设置

# setup.cfg
[tool:pytest]
minversion = 7.0
testpaths =
    tests
python_files =
    test_*.py
python_classes =
    Test*
python_functions =
    test_*
    *test
markers =
    slow: 慢速测试
    integration: 集成测试
    e2e: 端到端测试
    smoke: 冒烟测试
filterwarnings =
    ignore::DeprecationWarning
addopts =
    --strict-markers

# pyproject.toml
[tool.pytest.ini_options]
minversion = "7.0"
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*", "*test"]
markers = {
    "slow": "慢速测试",
    "integration": "集成测试",
    "e2e": "端到端测试",
    "smoke": "冒烟测试",
}
filterwarnings = ["ignore::DeprecationWarning"]
addopts = ["--strict-markers"]

高级模式

阶梯式 Fixture

@pytest.fixture
def user():
    return User(id=1, name="Alice")

@pytest.fixture
def authenticated_user(user):
    return login(user)

@pytest.fixture
def admin_user(authenticated_user):
    authenticated_user.role = "admin"
    return authenticated_user

def test_admin_panel_access(admin_user):
    """测试管理面板访问。"""
    assert can_access_admin(admin_user)
    # 无法直接访问其他测试中的 admin_user

条件测试

import sys
import pytest

@pytest.mark.skipif(sys.version_info < (3, 8), reason="需要 Python 3.8 或更高版本")
def test_new_feature():
    """仅在 Python 3.8+ 上运行。"""
    assert feature_available()

@pytest.mark.skipif(
    not database_available(),
    reason="数据库不可用"
)
def test_database_feature():
    """仅当数据库可用时运行。"""
    assert query_database() is not None

预期失败

@pytest.mark.xfail
def test_known_bug():
    """已知失败的测试。"""
    assert buggy_function() == expected_value

@pytest.mark.xfail(reason="等待修复完成")
def test_pending_feature():
    """标记为预期失败的测试。"""
    assert pending_feature() is True

性能测试

测试性能

import time

def test_performance():
    """测试函数性能。"""
    start = time.time()
    result = slow_function()
    duration = time.time() - start

    assert duration < 1.0  # 应在 1 秒内完成

pytest-benchmark

# 需要安装：pip install pytest-benchmark

def test_slow_function(benchmark):
    """对函数进行基准测试。"""
    result = benchmark(slow_function)
    assert result is not None

测试数据

工厂和构建器

class UserFactory:
    """在测试中创建用户实例的工厂。"""
    @staticmethod
    def create(name="默认名称", email="[email protected]"):
        return User(id=randint(1000, 9999), name=name, email=email)

def test_user_registration():
    """测试用户注册。"""
    user = UserFactory.create(name="测试用户", email="[email protected]")
    assert register_user(user) is True

结论

这些模式为 Python 测试提供了坚实的方法，利用了 pytest 的强大功能和灵活性。遵循这些实践可以确保测试套件可维护、可靠，并且能随着代码库的增长而扩展。

请记住：先写测试，保持测试快速，测试应该具有确定性。