PythonTestingPatternsSkill python-testing

Python 测试模式

使用 pytest、TDD 方法论和最佳实践的 Python 应用综合测试策略。

何时激活

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

核心测试理念

测试驱动开发（TDD）

始终遵循 TDD 周期：

1. RED：为期望的行为编写失败的测试
2. GREEN：编写最少的代码使测试通过
3. REFACTOR：在保持测试绿色的同时改进代码

# 第 1 步：编写失败的测试（RED）
def test_add_numbers():
    结果 = add(2, 3)
    assert 结果 == 5

# 第 2 步：编写最小实现（GREEN）
def add(a, b):
    返回 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():
    """测试字符串大写。"""
    文本 = "hello"
    assert 文本.upper() == "HELLO"

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

断言

# 相等性
assert 结果 == 预期

# 不相等性
assert 结果 != 意外

# 真实性
assert 结果  # 真实
assert not 结果  # 虚假
assert 结果 is True  # 确切 True
assert 结果 is False  # 确切 False
assert 结果 is None  # 确切 None

# 成员资格
assert 项目 in 集合
assert 项目 not in 集合

# 比较
assert 结果 > 0
assert 0 <= 结果 <= 100

# 类型检查
assert isinstance(结果, str)

# 异常测试（首选方法）
with pytest.raises(ValueError):
    引发 ValueError("错误信息")

# 检查异常消息
with pytest.raises(ValueError, match="无效输入"):
    引发 ValueError("提供的输入无效")

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

固定装置

基本固定装置使用

import pytest

@pytest.fixture
def sample_data():
    """提供样本数据的固定装置。"""
    返回 {"name": "Alice", "age": 30}

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

带设置/拆除的固定装置

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

    产生 db  # 提供给测试

    # 拆除
    db.close()

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

固定装置范围

# 函数范围（默认） - 每个测试运行一次
@pytest.fixture
def temp_file():
    与 open("temp.txt", "w") as f:
        产生 f
    os.remove("temp.txt")

# 模块范围 - 每个模块运行一次
@pytest.fixture(scope="module")
def module_db():
    db = Database(":memory:")
    db.create_tables()
    产生 db
    db.close()

# 会话范围 - 每个测试会话运行一次
@pytest.fixture(scope="session")
def shared_resource():
    资源 = ExpensiveResource()
    产生 资源
    资源.cleanup()

带参数的固定装置

@pytest.fixture(params=[1, 2, 3])
def number(request):
    """参数化固定装置。"""
    返回 request.param

def test_numbers(number):
    """测试运行 3 次，每次使用不同的参数。"""
    assert number > 0

使用多个固定装置

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

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

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

自动使用固定装置

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

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

用于共享固定装置的 Conftest.py

# tests/conftest.py
import pytest

@pytest.fixture
def client():
    """所有测试共享的固定装置。"""
    应用 = create_app(testing=True)
    与应用.test_client() as client:
        产生 client

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

参数化

基本参数化

@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", [
    ("valid@email.com", 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

参数化固定装置

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

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

标记和测试选择

自定义标记

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

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

# 标记单元测试
@pytest.mark.unit
def test_unit_logic():
    assert 计算(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 的测试

模拟和打补丁

模拟函数

从 unittest.mock 导入 patch, Mock

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

    结果 = 我的函数()

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

模拟返回值

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

    db = Database()
    db.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("网络错误")

    与 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 = "文件内容"

    结果 = 读取文件("test.txt")

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

使用 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()

模拟类实例

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

        服务 = UserService(仓库_mock.return_value)
        用户 = 服务.create_user(name="Alice")

        assert 用户.name == "Alice"
        仓库_mock.return_value.save.assert_called_once()

模拟属性

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

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

测试异步代码

使用 pytest-asyncio 进行异步测试

导入 pytest

@pytest.mark.asyncio
异步 def test_async_function():
    """测试异步函数。"""
    结果 = 等待 async_add(2, 3)
    assert 结果 == 5

@pytest.mark.asyncio
异步 def test_async_with_fixture(async_client):
    """使用异步固定装置进行测试。"""
    响应 = 等待 async_client.get("/api/users")
    assert 响应.status_code == 200

异步固定装置

@pytest.fixture
异步 def async_client():
    """提供异步测试客户端的异步固定装置。"""
    应用 = create_app()
    异步与 应用.test_client() as client:
        产生 client

@pytest.mark.asyncio
异步 def test_api_endpoint(async_client):
    """使用异步固定装置进行测试。"""
    响应 = 等待 async_client.get("/api/data")
    assert 响应.status_code == 200

模拟异步函数

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

    结果 = 等待 我的异步函数()

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

测试异常

测试预期异常

def test_divide_by_zero():
    """测试除以零引发的 ZeroDivisionError。"""
    与 pytest.raises(ZeroDivisionError):
        分割(10, 0)

def test_custom_exception():
    """测试带有消息的自定义异常。"""
    与 pytest.raises(ValueError, match="invalid input"):
        验证输入("invalid")

测试异常属性

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

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

测试副作用

测试文件操作

导入 tempfile
导入 os

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

    尝试：
        结果 = 处理文件(临时路径)
        assert 结果 == "processed: 测试内容"
    最后：
        os.unlink(临时路径)

使用 pytest 的 tmp_path 固定装置进行测试

def test_with_tmp_path(tmp_path):
    """使用 pytest 内置的临时路径固定装置进行测试。"""
    测试文件 = tmp_path / "test.txt"
    测试文件.write_text("hello world")

    结果 = 处理文件(str(测试文件))
    assert 结果 == "hello world"
    # tmp_path 自动清理

使用 tmpdir 固定装置进行测试

def test_with_tmpdir(tmpdir):
    """使用 pytest 的 tmpdir 固定装置进行测试。"""
    测试文件 = tmpdir.join("test.txt")
    测试文件.write("数据")

    结果 = 处理文件(str(测试文件))
    assert 结果 == "数据"

测试组织

目录结构

tests/
├── conftest.py                 # 共享固定装置
├── __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

测试类

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

    @pytest.fixture(autouse=True)
def setup(self):
        """在此类别的每个测试前运行的设置。"""
        自我.service = UserService()

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

    def test_delete_user(self):
        """测试用户删除。"""
        用户 = User(id=1, name="Bob")
        自我.service.delete_user(用户)
        assert not 自我.service.user_exists(1)

最佳实践

应该做

• 遵循 TDD：在代码之前编写测试（红-绿-重构）
• 测试一件事：每个测试应该验证一个单一行为
• 使用描述性名称：test_user_login_with_invalid_credentials_fails
• 使用固定装置：使用固定装置消除重复
• 模拟外部依赖项：不要依赖外部服务
• 测试边缘情况：空输入，None 值，边界条件
• 目标 80%+ 覆盖率：专注于关键路径
• 保持测试快速：使用标记分离慢速测试

不应该做

• 不要测试实现：测试行为，而不是内部
• 不要在测试中使用复杂的条件：保持测试简单
• 不要忽略测试失败：所有测试必须通过
• 不要测试第三方代码：相信库的工作
• 不要在测试之间共享状态：测试应该是独立的
• 不要在测试中捕获异常：使用 pytest.raises
• 不要使用打印语句：使用断言和 pytest 输出
• 不要编写过于脆弱的测试：避免过度特定的模拟

常见模式

测试 API 端点（FastAPI/Flask）

@pytest.fixture
def client():
    应用 = create_app(testing=True)
    返回 应用.test_client()

def test_get_user(client):
    响应 = 客户端.get("/api/users/1")
    assert 响应.status_code == 200
    assert 响应.json["id"] == 1

def test_create_user(client):
    响应 = 客户端.post("/api/users", json={
        "name": "Alice",
        "email": "alice@example.com"
    })
    assert 响应.status_code == 201
    assert 响应.json["name"] == "Alice"

测试数据库操作

@pytest.fixture
def db_session():
    """创建测试数据库会话。"""
    会话 = Session(bind=engine)
    会话.begin_nested()
    产生 会话
    会话.rollback()
    会话.close()

def test_create_user(db_session):
    用户 = User(name="Alice", email="alice@example.com")
    db_session.add(用户)
    db_session.commit()

    检索到的 = db_session.query(User).filter_by(name="Alice").first()
    assert 检索到的.email == "alice@example.com"

测试类方法

类 TestCalculator:
    @pytest.fixture
    def calculator(self):
        返回 Calculator()

    def test_add(self, calculator):
        assert calculator.add(2, 3) == 5

    def test_divide_by_zero(self, calculator):
        与 pytest.raises(ZeroDivisionError):
            计算器.divide(10, 0)

pytest 配置

pytest.ini

[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
    --strict-markers
    --disable-warnings
    --cov=mypackage
    --cov-report=term-missing
    --cov-report=html
markers =
    slow: 标记慢速测试
    integration: 标记集成测试
    unit: 标记单元测试

pyproject.toml

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
    "--strict-markers",
    "--cov=mypackage",
    "--cov-report=term-missing",
    "--cov-report=html",
]
markers = [
    "slow: 标记慢速测试",
    "integration: 标记集成测试",
    "unit: 标记单元测试",
]

运行测试

# 运行所有测试
pytest

# 运行特定文件
pytest tests/test_utils.py

# 运行特定测试
pytest tests/test_utils.py::test_function

# 运行详细输出
pytest -v

# 运行覆盖
pytest --cov=mypackage --cov-report=html

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

# 运行直到第一次失败
pytest -x

# 运行并在 N 次失败后停止
pytest --maxfail=3

# 运行上次失败的测试
pytest --lf

# 运行带有模式的测试
pytest -k "test_user"

# 运行并在失败时使用调试器
pytest --pdb

快速参考

模式	使用
pytest.raises()	测试预期异常
@pytest.fixture()	创建可重用的测试固定装置
@pytest.mark.parametrize()	使用多个输入运行测试
@pytest.mark.slow	标记慢速测试
pytest -m "not slow"	跳过慢速测试
@patch()	模拟函数和类
tmp_path 固定装置	自动临时目录
pytest --cov	生成覆盖报告
assert	简单且易读的断言

记住：测试也是代码。保持它们干净、易读和可维护。好的测试可以捕获错误；伟大的测试可以预防它们。