测试与质量保证

测试与质量保证

测试是让你确信代码不仅在当下能用、而且在每次改动之后都能用的方式。本文件覆盖测试金字塔、用 pytest 写单元测试、mock、测试 ML 特有的代码、CI/CD 管道、lint、格式化和代码评审——这些实践能在 bug 到达生产之前就抓住它们。

• ML 代码是出了名的缺乏测试。"它能训练，所以它能用"是主流态度。这会导致隐蔽的 bug：一个 shuffle 错误的数据加载器、一个符号写反的损失函数、一个悄悄丢掉 5% 数据的预处理步骤。这些 bug 不会让你的程序崩溃。它们只是让你的模型悄悄地变差，然后你在"本该更高"的指标上浪费几周去调试。

• 测试不是负担。它是在不搞坏东西的前提下快速迭代的最快方式。

测试金字塔

• 测试按层组织，从又快又窄到又慢又宽：

  • 单元测试（unit tests，底层）：在隔离状态下测试单个函数和类。快（毫秒级），数量多（几百到几千个）。"normalise_image 的输出是否落在 [0, 1] 区间？"

  • 集成测试（integration tests，中层）：测试多个组件能否协同工作。较慢（秒级）。"数据加载器产出的 batch 是不是模型期望的格式？"

  • 端到端测试（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）

• **夹具（fixture）**为测试提供可复用的初始化。与其在每个测试里重复写初始化代码，不如定义一次：

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
​

模拟与打补丁

• **模拟（mocking）**在测试期间用一个假的依赖替换真的。这让你能在隔离状态下测试一个函数，而不需要数据库、API 或 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()

        # 验证训练器记录了指标
        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)

# 对 tensor
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}"
​

• 在一个 batch 上过拟合：一个模型应该能够记住单个 batch。如果做不到，那就有根本性的问题。

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）：在每次提交或 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 钩子：在每次提交之前（本地）运行检查，在问题到达 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 都会跑这些钩子
​

Lint 与格式化

• Lint 在不运行代码的情况下抓住 bug 和风格问题。**格式化（formatting）**自动地强制执行一致的风格。

• Ruff：一个快速的 Python linter 和格式化工具（一个工具替代 flake8、isort 和 black）：

ruff check src/          # lint
ruff check --fix src/    # lint 并自动修复
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。"""
    ...
​

代码评审的最佳实践

• 对作者来说：

  • 在请求评审之前先自己评审一遍 diff。你会抓住明显的问题。
  • 让 PR 小而聚焦。一个 PR 只关注一件事。
  • 写一个清晰的描述：改了什么、为什么、怎么测试。
  • 回应每一条评论（哪怕只是"done"）。

• 对评审者来说：

  • 友善。批评代码，不批评人。说"这里可以更清晰"而不是"这里让人困惑"。
  • 区分阻塞性问题（bug、安全）和建议（风格、命名）。用标签："nit:"、"suggestion:"、"blocking:"。
  • 多提问，少命令。"如果这个列表是空的会怎么样？"比"处理一下空的情况"更有帮助。
  • 及时批准。一个 PR 等几天才有评审，会卡住作者，并鼓励那种又大又批量化的 PR（它们更难评审）。