python-testing

Python 测试模式

使用 pytest、TDD 方法论和最佳实践为 Python 应用程序提供的全面测试策略。

何时激活

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

核心测试理念

测试驱动开发 (TDD)

始终遵循 TDD 循环：

1. 红 (RED)：为期望的行为编写一个失败的测试
2. 绿 (GREEN)：编写最少的代码使测试通过
3. 重构 (REFACTOR)：保持测试通过的同时改进代码

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

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

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

覆盖率要求

• 目标：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("error message")

# 检查异常消息
with pytest.raises(ValueError, match="invalid input"):
    raise ValueError("invalid input provided")

# 检查异常属性
with pytest.raises(ValueError) as exc_info:
    raise ValueError("error message")
assert str(exc_info.value) == "error message"

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

带设置/拆卸的 Fixture

@pytest.fixture
def database():
    """带设置和拆卸的 Fixture。"""
    # 设置
    db = Database(":memory:")
    db.create_tables()
    db.insert_test_data()

    yield db  # 提供给测试

    # 拆卸
    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

使用多个 Fixtures

@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):
    """使用多个 Fixtures 进行测试。"""
    assert admin.can_manage(user)

自动使用 Fixtures

@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

Conftest.py 用于共享 Fixtures

# 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

参数化 Fixtures

@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("Network error")

    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 = "file content"

    result = read_file("test.txt")

    mock_file.assert_called_once_with("test.txt", "r")
    assert result == "file content"

使用 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="invalid input"):
        validate_input("invalid")

测试异常属性

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

    assert exc_info.value.code == 400
    assert "error" 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("test content")
        temp_path = f.name

    try:
        result = process_file(temp_path)
        assert result == "processed: test content"
    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("hello world")

    result = process_file(str(test_file))
    assert result == "hello world"
    # tmp_path 自动清理

使用 tmpdir Fixture

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

    result = process_file(str(test_file))
    assert result == "data"

测试组织

目录结构

tests/
├── conftest.py                 # 共享 Fixtures
├── __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):
        """此设置在此类的每个测试之前运行。"""
        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("Bob")
        assert result is True

配置

命令行

# 运行所有测试
pytest

# 运行带详细输出的测试
pytest -v

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

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

# 在第一个失败时停止
pytest -x

# 运行并显示 print 语句输出
pytest -s

# 运行并生成覆盖率报告
pytest --cov=myproject --cov-report=html

# 并行运行测试（需要 pytest-xdist）
pytest -n auto

# 运行带有特定关键字表达式的测试
pytest -k "addition or subtraction"

pytest.ini 配置

[pytest]
minversion = 7.0
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -v --strict-markers --tb=short
markers =
    slow: 慢速测试
    integration: 集成测试
    unit: 单元测试
    e2e: 端到端测试

持续集成中的测试

CI 管道中的测试执行

# .github/workflows/tests.yml
name: Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: 设置 Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.11'
    - name: 安装依赖
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov
    - name: 用覆盖率运行测试
      run: |
        pytest --cov=myproject --cov-report=xml
    - name: 上传覆盖率到 Codecov
      uses: codecov/codecov-action@v1

最佳实践

1. 一致性：在代码库中始终使用 pytest 风格
2. 有意义的名称：测试名称应以 test_ 开头并清晰描述
3. 单一断言：通常每个测试一个断言的关注点
4. 使用 Fixtures：使用 Fixtures 管理测试设置和拆卸
5. 利用 parametrize：使用参数化简化测试矩阵
6. 模拟外部依赖：模拟外部 API 和数据库以避免外部故障
7. 测试边界条件：包括无效输入、空值和边缘情况
8. 保持测试轻量：优先使用内存数据库和模拟
9. 组织结构：将测试组织成单元测试、集成测试和端到端测试
10. 自动化：在 CI 管道中运行测试

故障排除

常见问题和解决方案

问题	解决方案
测试查找问题	确保文件名以 test_ 开头，函数以 test_ 开头
Fixture 未找到	确保 Fixture 在 conftest.py 中或测试文件中定义
覆盖率低	使用 pytest --cov-report=term-missing 查看未覆盖的行
异步测试失败	确保安装了 pytest-asyncio 并使用 @pytest.mark.asyncio
模拟无效	确保 patching 使用正确的对象路径（字符串）
测试速度慢	使用 -x 在第一个失败时停止，或使用 -n auto 并行运行