测试与质量保证

一句话总结：测试金字塔从单元测试到端到端层层收敛；结合 pytest、mock、ML 专项校验、持续集成/持续部署 (Continuous Integration / Continuous Deployment, CI/CD) 与代码审查，把 bug 挡在生产之外。

测试就是你知道自己代码能跑的方式，不只是现在能跑，而是每次改动后都能跑。这一节讲测试金字塔、用 pytest 写单元测试、mock 技术、ML 专属代码怎么测、CI/CD 流水线、静态检查与格式化，以及代码审查——这些都是在 bug 进入生产之前把它抓住的工程实践。

• ML 代码是出了名的测试薄弱。"能训练就等于能跑"是这个圈子的普遍心态。结果就是各种隐蔽 bug：顺序打乱错误的数据加载器、符号写反的损失函数、悄悄丢掉 5% 数据的预处理步骤。这些 bug 不会让你的程序崩溃，它们只会让模型悄悄变差，你却要花几周去调试那些"本该更高"的指标。

• 测试不是额外负担，它恰恰是"快而不砸"的最快路径。

测试金字塔

• 测试按层次组织，从快而窄到慢而广:

  • 单元测试 (unit tests) (底层)：隔离测试单个函数与类。快 (毫秒级)、数量多 (几百到几千个)。"normalise_image 输出的值是不是都落在 [0, 1] 区间?"

  • 集成测试 (integration tests) (中层)：测试各组件能否协同工作。稍慢 (秒级)。"数据加载器产出的批次格式，是不是模型期待的那种?"

  • 端到端测试 (end-to-end tests) (顶层)：测试从输入到输出的完整流水线。慢 (分钟级)。"python train.py --config test.yaml 能不能顺利跑完，并输出一个合法的 checkpoint?"

• 金字塔形状意味着：单元测试写很多，集成测试少一些，端到端测试只留一小撮。单元测试能抓住大多数 bug，而且秒级跑完；端到端测试能抓集成问题，但慢且脆。

用 pytest 写单元测试

• pytest 是 Python 事实上的标准测试框架。所谓一个测试，就是一个以 test_ 开头的函数，放在以 test_ 开头的文件里:

# tests/test_utils.py

def test_normalise_image():
    import numpy as np
    image = np.array([0, 128, 255], dtype=np.uint8)
    result = normalise_image(image, mean=128, std=128)
    assert result.min() >= -1.0
    assert result.max() <= 1.0
    assert abs(result[1]) < 1e-6  # 128 用 mean=128 归一化后应约等于 0

def test_normalise_empty():
    import numpy as np
    image = np.array([], dtype=np.uint8)
    result = normalise_image(image, mean=128, std=128)
    assert len(result) == 0

pytest tests/                     # 跑所有测试
pytest tests/test_utils.py        # 只跑一个文件
pytest -v                         # 冗长输出
pytest -x                         # 遇到第一个失败就停
pytest -k "normalise"             # 按名字模式匹配测试
pytest --tb=short                 # 更短的回溯信息

夹具

• 夹具 (fixtures) 为测试提供可复用的准备逻辑。不必在每个测试里都写一遍 setup，定义一次就够:

import pytest

@pytest.fixture
def sample_dataset():
    """构造一个小数据集用于测试。"""
    return {
        "inputs": torch.randn(10, 3, 32, 32),
        "labels": torch.randint(0, 10, (10,))
    }

@pytest.fixture
def trained_model():
    """加载一个小型预训练模型。"""
    model = SmallModel()
    model.load_state_dict(torch.load("tests/fixtures/small_model.pt"))
    return model

def test_model_output_shape(trained_model, sample_dataset):
    output = trained_model(sample_dataset["inputs"])
    assert output.shape == (10, 10)  # batch_size x num_classes

• 夹具有作用域 (scope): scope="function" (默认，每个测试一份新的)、scope="module" (每个文件一次)、scope="session" (整个测试运行一次)。像加载模型这种昂贵的准备工作，用 scope="session"。

参数化测试

• 同一个函数用多组输入测试，但不重复代码:

@pytest.mark.parametrize("input,expected", [
    ([1, 2, 3], 6),
    ([], 0),
    ([-1, 1], 0),
    ([1000000, 1000000], 2000000),
])
def test_sum(input, expected):
    assert sum(input) == expected

Mock 与打桩

• Mock (mocking) 就是在测试时把真实依赖换成一个假的。这样你就能孤立地测某个函数，不需要真去连数据库、调 API 或占用 GPU (Graphics Processing Unit, GPU)。

from unittest.mock import patch, MagicMock

def test_training_logs_metrics():
    mock_logger = MagicMock()

    with patch("my_project.training.trainer.wandb") as mock_wandb:
        trainer = Trainer(logger=mock_logger)
        trainer.train_one_epoch()

        # 验证 trainer 记录了指标
        mock_logger.log.assert_called()
        # 验证它记录了一个 loss 值
        call_args = mock_logger.log.call_args
        assert "loss" in call_args[1]

• 什么时候该 mock：外部服务 (API、数据库、云存储)、昂贵操作 (GPU 计算、大文件 I/O)，以及非确定性行为 (随机数生成器、时间戳)。

• 什么时候不该 mock：你自己的代码。如果你把一切都 mock 掉，你的测试就只是在验证"mock 的行为符合预期"，而不是"你的代码能工作"。Mock 要打在边界上，你自己的逻辑要直接测。

测试 ML 代码

• ML 代码有独特的测试挑战：输出是概率性的、训练很慢、"正确"很模糊。

确定性种子

• 到处设随机种子，让测试可复现:

import random
import numpy as np
import torch

def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

数值容差

• 浮点比较必须带容差 (见第 13 章的 IEEE 754):

# 差的写法: 精确比较, 会被浮点误差搞挂
assert model_output == 0.5

# 好的写法: 近似比较
import numpy as np
assert np.isclose(model_output, 0.5, atol=1e-5)

# 张量版本
assert torch.allclose(output, expected, atol=1e-4)

ML 里到底该测什么

• 形状测试：验证输出维度是不是符合预期。

def test_model_output_shape():
    model = MyModel(d_model=256, n_classes=10)
    x = torch.randn(8, 32, 256)  # batch=8, seq=32, dim=256
    output = model(x)
    assert output.shape == (8, 10)

• 梯度流动：验证可训练参数的梯度非零。

def test_gradients_flow():
    model = MyModel()
    x = torch.randn(4, 3, 32, 32)
    y = torch.randint(0, 10, (4,))

    output = model(x)
    loss = F.cross_entropy(output, y)
    loss.backward()

    for name, param in model.named_parameters():
        assert param.grad is not None, f"No gradient for {name}"
        assert param.grad.abs().sum() > 0, f"Zero gradient for {name}"

• 单批次过拟合：模型应该能背下一个批次。如果连这都做不到，那就一定是有根本问题。

def test_overfit_one_batch():
    model = MyModel()
    optimiser = torch.optim.Adam(model.parameters(), lr=1e-3)
    x, y = get_single_batch()

    for _ in range(100):
        loss = F.cross_entropy(model(x), y)
        loss.backward()
        optimiser.step()
        optimiser.zero_grad()

    assert loss.item() < 0.01, f"Cannot overfit one batch: loss={loss.item()}"

• 数据校验：验证数据加载能产出合法输出。

def test_dataset_basics():
    dataset = MyDataset("tests/fixtures/small_data.csv")
    assert len(dataset) > 0
    x, y = dataset[0]
    assert x.shape == (3, 224, 224)
    assert 0 <= y < 10
    assert not torch.isnan(x).any()
    assert not torch.isinf(x).any()

• 确定性：同样的输入 + 同样的种子 → 同样的输出。

def test_determinism():
    set_seed(42)
    output1 = model(input_data)
    set_seed(42)
    output2 = model(input_data)
    assert torch.allclose(output1, output2)

CI/CD 流水线

• 持续集成 (Continuous Integration, CI)：每次 commit 或 PR 自动跑测试。测试失败，PR 就不能合并。这样能防止坏掉的代码进入 main。

• GitHub Actions 示例 (.github/workflows/ci.yml):

name: CI
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -e ".[dev]"
      - run: ruff check src/
      - run: mypy src/
      - run: pytest tests/ -v --tb=short

• Pre-commit 钩子：在每次 commit 之前 (本地) 跑检查，把问题挡在进入 CI 之前:

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.3.0
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.5.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml

pip install pre-commit
pre-commit install    # 现在每次 git commit 都会跑钩子

静态检查与格式化

• 静态检查 (linting) 不需要运行代码就能抓 bug 和风格问题。格式化 (formatting) 则自动强制统一风格。

• Ruff：一个飞快的 Python linter 兼 formatter (一个工具就顶掉了 flake8、isort 和 black):

ruff check src/          # 静态检查
ruff check --fix src/    # 检查并自动修复
ruff format src/         # 格式化

• mypy: Python 静态类型检查器。在运行之前就抓类型错误:

mypy src/
# src/model.py:42: error: Argument 1 to "forward" has incompatible type "int"; expected "Tensor"

• 类型注解让代码自带文档，顺便抓 bug:

def train(
    model: nn.Module,
    dataloader: DataLoader,
    optimiser: torch.optim.Optimizer,
    num_epochs: int = 10,
) -> float:
    """训练模型并返回最终 loss。"""
    ...

代码审查最佳实践

• 对作者:

  • 请求 review 之前，先自己 review 一遍 diff。明显的问题你自己就能抓到。
  • PR 要小而聚焦。一个 PR 只解决一个关注点。
  • 写清楚描述：做了什么、为什么做、怎么测。
  • 每一条评论都要回 (哪怕只是"已改")。

• 对审查者:

  • 好好说话。批评代码，不是批评人。"这里可以更清楚一点"胜过"这写得让人看不懂"。
  • 区分阻塞问题 (bug、安全) 和建议 (风格、命名)。用标签："nit:"、"suggestion:"、"blocking:"。
  • 多问问题，少下命令。"如果这个 list 是空的会怎样?"比"处理一下空的情况"有用得多。
  • 及时批。一个 PR 等好几天没人 review，既卡住作者，又逼大家攒大 PR (更难 review)。

---

📌 本节要点

• 测试金字塔：单元测试最多、集成测试其次、端到端测试最少。
• pytest 基础：以 test_ 开头的函数即测试，用夹具复用 setup，用参数化避免重复。
• Mock 边界原则：只 mock 外部依赖 (API、DB、GPU)，不 mock 自己的业务代码。
• 数值容差：浮点比较用 np.isclose / torch.allclose，不用 ==。
• ML 六大必测：形状、梯度流动、单批次过拟合、数据校验、确定性、数值容差。
• CI/CD：每次 commit / PR 自动跑 lint、type check、测试，失败即阻断合并。
• Pre-commit 钩子：把 lint 与格式化前置到本地 commit，减少 CI 反馈延迟。
• 工具链: Ruff 做 lint + format, mypy 做静态类型检查，类型注解自带文档。
• 代码审查心法：作者自 review + 小 PR + 清晰描述；审查者友善、区分阻塞与建议、多问问题、及时批。
• 测试哲学：测试不是负担，是"快而不砸"的最快路径。