跳至主要內容

Python 测试进阶

SunnyFan大约 12 分钟约 3598 字

Python 测试进阶

简介

测试进阶涵盖 pytest 的高级特性、测试组织策略、Mock 与 Patch 技术、参数化测试、异步测试和测试覆盖率管理。在基础单元测试之上，掌握这些能力可以构建高效、可靠且可维护的测试体系，确保代码质量在持续迭代中不退化。

特点

实现

pytest fixture 与参数化测试

"""用户服务测试"""
import pytest
from typing import Generator
from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str
    email: str
    active: bool = True

class UserService:
    def __init__(self):
        self._users: dict[int, User] = {}
        self._next_id = 1

    def create(self, name: str, email: str) -> User:
        if not name or not email:
            raise ValueError("姓名和邮箱不能为空")
        if "@" not in email:
            raise ValueError("邮箱格式不正确")
        user = User(id=self._next_id, name=name, email=email)
        self._users[user.id] = user
        self._next_id += 1
        return user

    def get(self, user_id: int) -> User | None:
        return self._users.get(user_id)

    def deactivate(self, user_id: int) -> bool:
        user = self._users.get(user_id)
        if user:
            user.active = False
            return True
        return False

    def list_active(self) -> list[User]:
        return [u for u in self._users.values() if u.active]

# --- Fixtures ---

@pytest.fixture
def user_service() -> UserService:
    """每个测试获取全新的 UserService 实例"""
    return UserService()

@pytest.fixture
def sample_users(user_service: UserService) -> list[User]:
    """创建示例用户数据"""
    users = [
        user_service.create("张三", "zhang@example.com"),
        user_service.create("李四", "li@example.com"),
        user_service.create("王五", "wang@example.com"),
    ]
    return users

@pytest.fixture(scope="session")
def db_config() -> dict:
    """整个测试会话共享的配置"""
    return {"host": "localhost", "port": 5432, "database": "test_db"}

# --- 参数化测试 ---

@pytest.mark.parametrize("name, email, expected_success", [
    ("张三", "zhang@example.com", True),
    ("李四", "li@example.com", True),
    ("", "empty@example.com", False),
    ("无名", "", False),
    ("测试", "invalid-email", False),
    ("测试", "no-at-sign.com", False),
])
def test_create_user_parametrized(user_service, name, email, expected_success):
    """参数化测试：多种输入验证创建逻辑"""
    if expected_success:
        user = user_service.create(name, email)
        assert user.name == name
        assert user.email == email
        assert user.id > 0
    else:
        with pytest.raises(ValueError):
            user_service.create(name, email)

def test_list_active_users(user_service, sample_users):
    """测试获取活跃用户列表"""
    active = user_service.list_active()
    assert len(active) == 3

    user_service.deactivate(sample_users[0].id)
    active = user_service.list_active()
    assert len(active) == 2
    assert sample_users[0] not in active

def test_deactivate_nonexistent_user(user_service):
    """测试停用不存在的用户"""
    result = user_service.deactivate(999)
    assert result is False

Mock 与 Patch 高级用法

"""外部服务调用测试"""
import pytest
from unittest.mock import Mock, patch, MagicMock, call
from typing import Any
import requests

class NotificationService:
    """通知服务"""
    def __init__(self, api_url: str, api_key: str):
        self.api_url = api_url
        self.api_key = api_key

    def send_email(self, to: str, subject: str, body: str) -> dict:
        response = requests.post(
            f"{self.api_url}/email",
            json={"to": to, "subject": subject, "body": body},
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=10,
        )
        response.raise_for_status()
        return response.json()

    def send_batch(self, recipients: list[str], subject: str, body: str) -> list[dict]:
        results = []
        for recipient in recipients:
            try:
                result = self.send_email(recipient, subject, body)
                results.append({"to": recipient, "status": "success", "data": result})
            except requests.RequestException as e:
                results.append({"to": recipient, "status": "failed", "error": str(e)})
        return results

# --- Mock 测试 ---

class TestNotificationService:
    @patch("requests.post")
    def test_send_email_success(self, mock_post):
        """测试成功发送邮件"""
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {"message_id": "msg-123", "status": "sent"}
        mock_response.raise_for_status = Mock()
        mock_post.return_value = mock_response

        svc = NotificationService("https://api.example.com", "test-key")
        result = svc.send_email("user@example.com", "测试主题", "测试内容")

        assert result["status"] == "sent"
        mock_post.assert_called_once_with(
            "https://api.example.com/email",
            json={"to": "user@example.com", "subject": "测试主题", "body": "测试内容"},
            headers={"Authorization": "Bearer test-key"},
            timeout=10,
        )

    @patch("requests.post")
    def test_send_email_failure(self, mock_post):
        """测试发送邮件失败"""
        mock_post.side_effect = requests.ConnectionError("网络不可达")

        svc = NotificationService("https://api.example.com", "test-key")
        with pytest.raises(requests.ConnectionError):
            svc.send_email("user@example.com", "测试", "内容")

    @patch("requests.post")
    def test_send_batch_partial_failure(self, mock_post):
        """测试批量发送部分失败"""
        responses = [
            Mock(json=Mock(return_value={"id": 1}), raise_for_status=Mock()),
            requests.Timeout("超时"),
            Mock(json=Mock(return_value={"id": 3}), raise_for_status=Mock()),
        ]
        mock_post.side_effect = responses

        svc = NotificationService("https://api.example.com", "test-key")
        results = svc.send_batch(["a@test.com", "b@test.com", "c@test.com"], "群发", "内容")

        assert results[0]["status"] == "success"
        assert results[1]["status"] == "failed"
        assert results[2]["status"] == "success"
        assert mock_post.call_count == 3

    def test_send_batch_with_magic_mock(self):
        """使用 MagicMock 测试批量发送"""
        svc = NotificationService("https://api.example.com", "test-key")
        svc.send_email = MagicMock(side_effect=[
            {"id": 1},
            Exception("发送失败"),
            {"id": 3},
        ])

        results = svc.send_batch(["a@t.com", "b@t.com", "c@t.com"], "测试", "内容")
        assert len(results) == 3

属性测试与假数据生成

"""属性测试（Property-based Testing）与假数据生成"""
import pytest
from hypothesis import given, strategies as st, settings, assume

# 使用 hypothesis 进行属性测试
# 不同于参数化测试指定具体值，属性测试自动生成大量随机输入

class Calculator:
    def add(self, a: int, b: int) -> int:
        return a + b

    def multiply(self, a: int, b: int) -> int:
        return a * b

    def divide(self, a: float, b: float) -> float:
        if b == 0:
            raise ValueError("除数不能为零")
        return a / b

# 属性测试 1：加法交换律
@given(st.integers(), st.integers())
def test_add_commutative(a, b):
    """加法满足交换律: a + b == b + a"""
    calc = Calculator()
    assert calc.add(a, b) == calc.add(b, a)

# 属性测试 2：乘法对加法的分配律
@given(st.integers(), st.integers(), st.integers())
def test_multiply_distributes(a, b, c):
    """乘法分配律: a * (b + c) == a * b + a * c"""
    calc = Calculator()
    assert calc.multiply(a, calc.add(b, c)) == calc.add(
        calc.multiply(a, b), calc.multiply(a, c))

# 属性测试 3：除法是乘法的逆运算
@given(st.floats(min_value=-1e6, max_value=1e6, allow_nan=False, allow_infinity=False),
       st.floats(min_value=-1e6, max_value=1e6, max_value=0, allow_nan=False, allow_infinity=False))
def test_divide_inverse(a, b):
    """除法是乘法的逆运算: (a * b) / b == a"""
    assume(abs(b) > 1e-10)  # 排除接近零的除数
    calc = Calculator()
    product = calc.multiply(a, b)
    result = calc.divide(product, b)
    assert abs(result - a) < 1e-6  # 浮点数比较

# 使用 Faker 生成假数据
from faker import Faker
fake = Faker("zh_CN")

@pytest.fixture
def fake_user_data():
    """生成假用户数据"""
    return {
        "name": fake.name(),
        "email": fake.email(),
        "phone": fake.phone_number(),
        "address": fake.address(),
        "company": fake.company(),
        "date_of_birth": fake.date_of_birth(minimum_age=18, maximum_age=65),
    }

def test_user_creation_with_fake_data(user_service, fake_user_data):
    """使用假数据测试用户创建"""
    user = user_service.create(
        name=fake_user_data["name"],
        email=fake_user_data["email"]
    )
    assert user.name == fake_user_data["name"]
    assert user.email == fake_user_data["email"]

# 参数化 + Faker 组合
@pytest.mark.parametrize("locale", ["zh_CN", "en_US", "ja_JP"])
def test_international_names(user_service, locale):
    """测试不同语言环境的姓名"""
    local_fake = Faker(locale)
    name = local_fake.name()
    user = user_service.create(name=name, email=f"test@example.com")
    assert user.name == name

测试夹具的高级用法

"""fixture 高级模式：工厂 fixture、嵌套 fixture、fixture 组合"""
import pytest
from typing import Generator

# 模式 1：工厂 fixture — 每次调用创建不同的实例
@pytest.fixture
def user_factory(user_service: UserService):
    """工厂 fixture：按需创建用户"""
    created_users = []

    def _create(name: str = None, email: str = None):
        name = name or f"测试用户_{len(created_users)}"
        email = email or f"user{len(created_users)}@test.com"
        user = user_service.create(name, email)
        created_users.append(user)
        return user

    yield _create

    # 清理：停用所有创建的用户
    for user in created_users:
        user_service.deactivate(user.id)

def test_with_factory(user_factory):
    """使用工厂 fixture 创建多个独立用户"""
    user1 = user_factory("张三", "zhang@test.com")
    user2 = user_factory("李四", "li@test.com")
    user3 = user_factory()  # 使用默认值

    assert user1.name == "张三"
    assert user2.name == "李四"
    assert user3.id > 0

# 模式 2：参数化 fixture — 自动生成多组 fixture
@pytest.fixture(params=[
    ("张三", "zhang@test.com", True),
    ("李四", "li@test.com", True),
    ("", "empty@test.com", False),
    ("王五", "", False),
])
def user_creation_params(request):
    """参数化 fixture：测试不同输入"""
    return request.param

def test_user_creation_parametrized_fixture(user_service, user_creation_params):
    name, email, should_succeed = user_creation_params
    if should_succeed:
        user = user_service.create(name, email)
        assert user.name == name
    else:
        with pytest.raises(ValueError):
            user_service.create(name, email)

# 模式 3：conftest.py 分层 fixture
# tests/
# ├── conftest.py          <- 全局 fixture（db_config, event_loop）
# ├── unit/
# │   └── conftest.py      <- 单元测试 fixture（mock services）
# ├── integration/
# │   └── conftest.py      <- 集成测试 fixture（real db, test client）
# └── e2e/
#     └── conftest.py      <- E2E fixture（browser, base url）

异步测试与 conftest.py

# conftest.py - 测试配置和共享 fixture
import pytest
import asyncio
from typing import AsyncGenerator

@pytest.fixture(scope="session")
def event_loop():
    """为所有异步测试提供事件循环"""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

@pytest.fixture
async def async_db_connection() -> AsyncGenerator:
    """异步数据库连接 fixture"""
    conn = {"connected": True, "data": []}
    yield conn
    conn["connected"] = False

# --- 异步测试 ---

import asyncio

class AsyncDataService:
    """异步数据服务"""
    async def fetch_items(self, category: str) -> list[dict]:
        await asyncio.sleep(0.01)
        items = [
            {"id": 1, "name": "项目A", "category": "tech"},
            {"id": 2, "name": "项目B", "category": "tech"},
            {"id": 3, "name": "项目C", "category": "finance"},
        ]
        return [i for i in items if i["category"] == category]

    async def process_items(self, items: list[dict]) -> list[dict]:
        results = []
        for item in items:
            await asyncio.sleep(0.01)
            results.append({**item, "processed": True})
        return results

@pytest.mark.asyncio
async def test_fetch_items():
    """测试异步获取数据"""
    svc = AsyncDataService()
    tech_items = await svc.fetch_items("tech")
    assert len(tech_items) == 2
    assert all(i["category"] == "tech" for i in tech_items)

@pytest.mark.asyncio
async def test_process_items():
    """测试异步处理数据"""
    svc = AsyncDataService()
    items = [{"id": 1, "name": "测试"}, {"id": 2, "name": "数据"}]
    results = await svc.process_items(items)
    assert len(results) == 2
    assert all(r["processed"] for r in results)

@pytest.mark.asyncio
async def test_fetch_and_process():
    """测试异步组合操作"""
    svc = AsyncDataService()
    items = await svc.fetch_items("tech")
    results = await svc.process_items(items)
    assert len(results) == 2

# --- 标记与跳过 ---

@pytest.mark.slow
def test_large_dataset():
    """标记为慢速测试，CI 中可选择性跳过"""
    data = list(range(100_000))
    assert len(data) == 100_000

@pytest.mark.skipif(sys.platform == "win32", reason="Linux 专用测试")
def test_linux_feature():
    assert True

# pytest -m "not slow"  # 跳过慢速测试
# pytest -m "slow"      # 只运行慢速测试

测试覆盖率与 CI 集成配置

# pytest.ini 或 pyproject.toml 中的 pytest 配置

# pyproject.toml:
# [tool.pytest.ini_options]
# testpaths = ["tests"]
# python_files = ["test_*.py"]
# python_classes = ["Test*"]
# python_functions = ["test_*"]
# addopts = "-v --tb=short --strict-markers"
# markers = [
#     "slow: 慢速测试",
#     "integration: 集成测试",
#     "unit: 单元测试",
# ]
#
# [tool.coverage.run]
# source = ["src"]
# omit = ["tests/*"]
#
# [tool.coverage.report]
# fail_under = 80
# exclude_lines = [
#     "pragma: no cover",
#     "if TYPE_CHECKING:",
#     "raise NotImplementedError",
# ]
# tests/__init__.py
# 测试包初始化（可以为空）

# tests/conftest.py
import pytest
from typing import Generator
import tempfile
import os
from pathlib import Path

@pytest.fixture
def temp_dir() -> Generator[Path, None, None]:
    """临时目录 fixture"""
    with tempfile.TemporaryDirectory() as tmpdir:
        yield Path(tmpdir)

@pytest.fixture
def env_vars() -> Generator[dict, None, None]:
    """临时环境变量 fixture"""
    original = os.environ.copy()
    test_env = {
        "APP_ENV": "test",
        "DATABASE_URL": "sqlite:///test.db",
        "DEBUG": "true",
    }
    os.environ.update(test_env)
    yield test_env
    os.environ.clear()
    os.environ.update(original)

# 运行命令:
# pytest                          # 运行所有测试
# pytest -v                       # 详细输出
# pytest --cov=src --cov-report=html  # 覆盖率报告
# pytest -m "not slow"            # 排除慢速测试
# pytest tests/unit/              # 只运行单元测试
# pytest -x                       # 第一个失败就停止
# pytest --lf                     # 只运行上次失败的测试

快照测试（Snapshot Testing）

"""快照测试 — 自动捕获和比较输出"""
import json
import pytest
from pathlib import Path

class SnapshotTester:
    """轻量级快照测试实现"""
    def __init__(self, snapshot_dir: Path):
        self.snapshot_dir = snapshot_dir
        self.snapshot_dir.mkdir(parents=True, exist_ok=True)
        self.update_mode = False  # 设置为 True 时更新快照

    def assert_match(self, data: dict, snapshot_name: str):
        """比较实际输出与快照"""
        snapshot_path = self.snapshot_dir / f"{snapshot_name}.json"
        actual_json = json.dumps(data, indent=2, ensure_ascii=False, sort_keys=True)

        if not snapshot_path.exists() or self.update_mode:
            snapshot_path.write_text(actual_json, encoding="utf-8")
            if not self.update_mode:
                pytest.fail(f"新快照已创建: {snapshot_name}")
            return

        expected_json = snapshot_path.read_text(encoding="utf-8")

        if actual_json != expected_json:
            # 输出差异
            print(f"\n快照不匹配: {snapshot_name}")
            print(f"期望:\n{expected_json}")
            print(f"实际:\n{actual_json}")
            pytest.fail(f"快照不匹配: {snapshot_name}（使用 update_mode=True 更新）")

@pytest.fixture
def snapshot_tester(tmp_path):
    return SnapshotTester(tmp_path / "__snapshots__")

# 使用快照测试 API 响应
def test_api_response_snapshot(snapshot_tester):
    """测试 API 响应是否与快照匹配"""
    response_data = {
        "users": [
            {"id": 1, "name": "张三", "role": "admin"},
            {"id": 2, "name": "李四", "role": "user"},
        ],
        "total": 2,
        "page": 1
    }
    snapshot_tester.assert_match(response_data, "users_list_api")

测试反模式与最佳实践

"""
测试反模式 — 避免以下写法
"""

# 反模式 1：测试私有方法
# 错误：直接测试 _internal_method
# 正确：通过公共方法间接测试私有逻辑
class UserService:
    def _validate_email(self, email: str) -> bool:
        return "@" in email and "." in email

    def create(self, name: str, email: str):
        if not self._validate_email(email):
            raise ValueError("邮箱无效")
        # ...

# 测试公共方法即可覆盖私有方法
def test_create_user_invalid_email(user_service):
    with pytest.raises(ValueError, match="邮箱无效"):
        user_service.create("测试", "invalid")

# 反模式 2：过度 Mock 导致测试脆弱
# 错误：Mock 了内部实现细节
def test_fragile_mock():
    svc = UserService()
    svc._validator = Mock(return_value=True)  # Mock 私有属性
    # 一旦内部实现变化，测试就失败

# 正确：只 Mock 外部依赖
def test_stable_mock():
    with patch("requests.post") as mock_post:
        mock_post.return_value = Mock(status_code=200)
        # 测试行为，不测试实现

# 反模式 3：测试之间有隐式依赖
# 错误：test_b 依赖 test_a 的执行结果
def test_a(user_service):
    user_service.create("张三", "zhang@test.com")

def test_b(user_service):
    # 假设 test_a 已经创建了用户 — 隐式依赖
    users = user_service.list_active()
    assert len(users) > 0  # 如果 test_a 未执行，这里会失败

# 正确：每个测试自己准备数据
def test_list_active_users_independent(user_service):
    user_service.create("张三", "zhang@test.com")
    user_service.create("李四", "li@test.com")
    users = user_service.list_active()
    assert len(users) == 2

# 反模式 4：忽略异常路径
# 只测试正常路径是不够的
def test_create_user_happy_path(user_service):
    user = user_service.create("张三", "zhang@test.com")
    assert user.name == "张三"

# 正确：同时覆盖异常路径
@pytest.mark.parametrize("name,email,expected_error", [
    ("", "test@example.com", "姓名和邮箱不能为空"),
    (None, "test@example.com", "姓名和邮箱不能为空"),
    ("张三", "", "姓名和邮箱不能为空"),
    ("张三", "invalid", "邮箱格式不正确"),
    ("张三", "no-at.com", "邮箱格式不正确"),
])
def test_create_user_edge_cases(user_service, name, email, expected_error):
    with pytest.raises(ValueError, match=expected_error):
        user_service.create(name, email)

优点

缺点

总结

测试进阶的核心是掌握 pytest 的 fixture 体系、参数化能力和 Mock 技术，构建高效可靠的测试体系。测试应分层组织（单元/集成/端到端），核心业务逻辑追求高覆盖率，外部依赖通过 Mock 隔离。在 CI 中运行测试并监控覆盖率趋势，确保代码质量在持续迭代中不退化。

关键知识点

  • pytest fixture 通过依赖注入管理测试数据的创建和清理
  • @pytest.mark.parametrize 实现一组逻辑覆盖多组输入输出
  • unittest.mock 的 patch 替换外部依赖，Mock 模拟返回值和行为
  • conftest.py 文件中的 fixture 对同目录及子目录的测试自动可见

项目落地视角

  • 测试按 unit/integration/e2e 分层组织，CI 中分层执行
  • 核心业务逻辑测试覆盖率不低于 80%，关键路径追求 90%+
  • fixture 和 Mock 的使用在团队内统一规范，避免过度 Mock
  • 测试失败的告警集成到团队通知渠道，确保问题及时处理

常见误区

  • 过度追求覆盖率数字，编写无意义的测试（如测试 getter/setter）
  • Mock 了过多内部依赖，导致测试与实现强耦合
  • 测试之间有隐式依赖，执行顺序影响测试结果
  • 忽略异步代码的测试，异步逻辑缺少测试覆盖

进阶路线

  • 学习 pytest-bdd 实现 BDD 风格的测试
  • 研究 snapshot testing 和 approval testing 模式
  • 了解 property-based testing（hypothesis 库）
  • 探索测试在微服务契约测试和混沌工程中的应用

适用场景

  • 中大型项目需要持续保证代码质量
  • 团队协作项目需要防止代码变更引入回归
  • 公共库和 API 需要保证接口兼容性

落地建议

  • 从核心业务逻辑开始补测试，优先覆盖失败路径和边界条件
  • CI 流水线中强制运行测试，未通过的 PR 不允许合并
  • 定期审查测试质量，清理脆弱和冗余的测试

排错清单

  • 检查 fixture 的作用域（function/class/module/session）是否正确
  • 确认 Mock 的 patch 路径是否指向实际使用位置而非定义位置
  • 排查测试间的状态泄漏，确保每个测试独立运行

复盘问题

  • 你的项目测试覆盖率是多少？哪些模块最需要补测试？
  • 测试失败时能否快速定位原因？是否需要改善测试的错误信息？
  • 团队是否有测试编写规范？新功能的测试是否与代码同步提交？

延伸阅读

Copyright © 2026 SunnyFan