单元测试设计指南

通用指南¶

为每个模块创建单元测试¶

单元测试位于 tests/ludwig/，与 ludwig/ 源码树并行。ludwig/ 中每个具有可测试功能的源文件都应该在 test/ludwig/ 中有一个相应的单元测试文件，文件名与源文件对应，并以 test_ 为前缀。

示例

模块	测试
`ludwig/data/dataset_synthesizer.py`	`tests/ludwig/data/test_dataset_synthesizer.py`
`ludwig/features/audio_feature.py`	`tests/ludwig/features/test_audio_feature.py`
`ludwig/modules/convolutional_modules.py`	`tests/ludwig/modules/test_convolutional_modules.py`

注意

截至本文撰写时，Ludwig 中并非所有模块都拥有适当的单元测试或符合这里的指南。这些指南是我们追求的目标，我们相信渐进式改进好于要求完美。Ludwig 的测试工作正在进行中，任何有助于我们更接近 100% 测试覆盖率目标的更改都受到欢迎！

应该测试什么¶

单元测试通常应该测试模块导出的每个函数或方法，但有一些例外。一个好的规则是，如果一个函数或方法满足某个要求并且将从模块外部调用，那么它就应该有相应的测试。

测试常见情况。这可以在出现问题时提供通知。
测试复杂方法的边界情况，如果你认为它们可能存在错误。
测试失败情况。如果一个方法必须失败（例如，当输入超出范围时），确保它确实失败了。
Bug。当你发现一个 bug 时，在修复它之前，先编写一个测试用例来覆盖它。

参数化测试¶

使用 @pytest.mark.parameterize 来测试参数的组合，以覆盖所有代码路径，确保函数在各种情况下的行为正确。

# test combinations of parameters to exercise all code paths
@pytest.mark.parameterize(
    'num_total_blocks, num_shared_blocks',
    [(4, 2), (6, 4), (3, 1)]
)
@pytest.mark.parameterize('virtual_batch_size', [None, 7])
@pytest.mark.parameterize('size', [4, 12])
@pytest.mark.parameterize('input_size', [2, 6])
def test_feature_transformer(
    input_size: int,
    size: int,
    virtual_batch_size: Optional[int],
    num_total_blocks: int,
    num_shared_blocks: int
) -> None:
    feature_transformer = FeatureTransformer(
        input_size,
        size,
        bn_virtual_bs=virtual_batch_size,
        num_total_blocks=num_total_blocks,
        num_shared_blocks=num_shared_blocks
    )

测试边界情况¶

尽可能测试边界情况。例如，如果一个方法接受多个输入：测试空输入、单个输入和大量输入的情况。

@pytest.mark.parametrize("virtual_batch_size", [None, 7, 64])  # Test with no virtual batch norm, odd size, or large.
@pytest.mark.parametrize("input_size", [1, 8, 256])  # Test with single input feature or many inputs.

张量类型与形状¶

至少，与张量相关的测试应该确认在处理张量时没有引发错误，并且结果张量具有正确的形状和类型。这提供了函数按预期运行的最低保证。

# pass input features through combiner
combiner_output = combiner(input_features)

# check for required attributes in the generated output
assert hasattr(combiner, 'input_dtype')
assert hasattr(combiner, 'output_shape')

# check for correct data type
assert isinstance(combiner_output, dict)

# required key present
assert 'combiner_output' in combiner_output

# check for correct output shape
assert (combiner_output['combiner_output'].shape
       == (batch_size, *combiner.output_shape))

可训练模块¶

测试可训练模块（层、编码器、解码器、Combiner 或模型）时，请确保所有变量/权重在一步训练后得到更新。这将确保计算图中不包含悬空节点。这能捕获不会表现为崩溃的细微问题，这些问题无法通过查看损失分数或将模型训练到收敛（尽管通常损失很差）来发现。更多详情请参见如何对机器学习代码进行单元测试。

根据 torch 输入和输出添加类型检查。确保所有模块输出都具有预期的 torch 数据类型、维度和张量形状。

对于可训练模块，我们建议至少添加一个过拟合测试。确保包含该模块的小型 ECD 模型能够在小型数据集上过拟合。这确保模型能够收敛到合理的目标，并捕获形状、类型或权重更新测试未能捕获的任何范围外问题。

最佳实践¶

网上有很多关于编写优秀测试的建议。以下是我们赞同的 Microsoft 建议中的一些重点内容：

不应该测试什么¶

并非所有函数都需要测试，一个好的规则是，如果一个函数或方法满足某个要求并且将从模块外部调用，那么它就应该有相应的测试。

不需要单元测试的情况

构造函数或属性。仅当它们包含验证时才进行测试。
配置，如常量、只读字段、配置等。
围绕其他框架或库的外观模式或封装器。
私有方法

实现指南¶

使用 pytest.mark.parameterize¶

自动化测试用例的设置。请注意只测试有意义地不同的参数值，因为测试用例的总数会组合式地增长。

@pytest.mark.parameterize('enc_should_embed', [True, False])
@pytest.mark.parameterize('enc_reduce_output', [None, 'sum'])
@pytest.mark.parameterize('enc_norm', [None, 'batch', 'layer'])
@pytest.mark.parameterize('enc_num_layers', [1, 2])
@pytest.mark.parameterize('enc_dropout', [0, 0.2])
@pytest.mark.parameterize('enc_cell_type', ['rnn', 'gru', 'lstm'])
@pytest.mark.parameterize('enc_encoder', ENCODERS + ['passthrough'])
def test_sequence_encoders(
        enc_encoder: str,
        enc_cell_type: str,
        enc_dropout: float,
        enc_num_layers: int,
        enc_norm: Union[None, str],
        enc_reduce_output: Union[None, str],
        enc_should_embed: bool,
        input_sequence: torch.Tensor
):

为生成的数据使用 temp_path 或 tmpdir¶

为任何生成的数据使用临时目录。PyTest 提供了用于临时目录的 fixture，这些目录在每次测试运行时都是唯一的，并且会自动清理。我们推荐使用 tmpdir，它提供一个兼容 os.path 方法的 py.path.local 对象。如果你使用 pathlib，PyTest 也提供了 tmp_path，它是一个 pathlib.Path。

更多详情，请参阅 PyTest 文档中的临时目录和文件。

示例

@pytest.mark.parametrize("skip_save_processed_input", [True, False])
@pytest.mark.parametrize("in_memory", [True, False])
@pytest.mark.parametrize("image_source", ["file", "tensor"])
@pytest.mark.parametrize("num_channels", [1, 3])
def test_basic_image_feature(
        num_channels, image_source, in_memory, skip_save_processed_input, tmpdir
):
    # Image Inputs
    image_dest_folder = os.path.join(tmpdir, "generated_images")

合并需要设置的测试¶

例如，多个测试可能依赖于加载需要时间的相同训练/测试数据集。如果多个测试依赖于相同的通用资源，请将这些测试分组到一个模块中，并使用适当范围的 @pytest.fixture 来减少重复执行设置的开销。

可在 tests/conftest.py 中找到可重用测试 fixture 的示例。该模块包含适用于许多测试的可重用 @pytest.fixtures。

确定性测试¶

在可能的情况下，使用相同参数运行的每个测试都应产生相同的结果。使用随机数生成器时，始终指定一个种子。如果一个测试在不同运行中产生不同的输出，将很难调试。

import torch

RANDOM_SEED = 1919

# setup synthetic tensor, ensure reproducibility by setting seed
torch.manual_seed(RANDOM_SEED)
input_tensor = torch.randn([BATCH_SIZE, input_size], dtype=torch.float32)

测试参数更新¶

tests.integration_tests.parameter_utils 模块中的实用函数 check_module_parameters_updated() 可用于测试 Ludwig 模块（例如 encoder、combiner、decoder 和相关的子组件）在前向传播-反向传播-优化步骤序列中是否更新参数。

实现参数更新测试的指南

对于非常简单的模块（如全连接层）或知名的预训练模块（如 Huggingface 文本编码器）来说不是必需的。
在实现参数更新测试之前，请确保测试不会生成运行时异常，生成的输出符合预期的数据结构，并且输出的形状正确。

check_module_parameters_updated(module, input, target) 函数需要三个位置参数：

module 是要测试的 Ludwig 组件，即 encoder、combiner 或 decoder
input 是一个元组，作为 Ludwig 组件前向方法的输入
target 是一个合成张量，表示用于在前向传播结束时计算损失的目标值。

module 和 input 参数可以与测试早期用于确保没有运行时异常和输出正确的部分相同。

建议分两步来实现测试：

步骤 1:

设置随机种子以确保可重复性。部分实现参数更新测试以打印参数计数，例如：

    # check for parameter updating
    target = torch.randn(output.shape)
    fpc, tpc, upc, not_updated = check_module_parameters_updated(sequence_rnn_decoder, (combiner_outputs, None), target)
    print(fpc, tpc, upc, not_updated)

target 是一个包含合成数据的张量，用于在前向传播后计算损失。

fpc 是冻结参数的计数。tpc 是可训练参数的计数。upc 是更新参数的计数，即在前向传播-反向传播-优化步骤循环中更新的参数数量。检查这些值是否正确，例如：

fpc + tpc 应该等于模块中的参数总数
除了像 Huggingface 文本编码器这样的预训练模块外，fpc 应该为零。
upc <= tpc

如果某些参数未更新，not_updated 是一个包含未更新参数名称的 Python 列表。

在理想情况下，upc == tpc，即可训练参数的数量等于更新参数的数量。然而，在某些情况下可能会出现 upc < tpc。这可能在使用 dropout、或对单个训练样本使用批量归一化、或在 forward() 方法中进行条件处理时发生。上述并非详尽列表。无论何时出现 upc < tpc，开发者都应确认该计数对于当前情况是正确的。

理解 Ludwig 模块参数结构的技巧

在进行步骤 1 时，可以暂时使用这些代码片段来深入了解 Ludwig 模块的结构和参数。

print() 将显示构成 Ludwig 模块的层。开发者可以从输出中确认正确的结构和配置值。

encoder = create_encoder(
    Stacked2DCNN, num_channels=image_size[0], height=image_size[1], width=image_size[2], **encoder_kwargs
)
print(encoder)

# output
Stacked2DCNN(
  (conv_stack_2d): Conv2DStack(
    (stack): ModuleList(
      (0): Conv2DLayer(
        (layers): ModuleList(
          (0): Conv2d(3, 32, kernel_size=(3, 3), stride=(1, 1), padding=valid)
          (1): ReLU()
          (2): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
        )
      )
      (1): Conv2DLayer(
        (layers): ModuleList(
          (0): Conv2d(32, 32, kernel_size=(3, 3), stride=(1, 1), padding=valid)
          (1): ReLU()
          (2): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
        )
      )
    )
  )
  (flatten): Flatten(start_dim=1, end_dim=-1)
  (fc_stack): FCStack(
    (stack): ModuleList(
      (0): FCLayer(
        (layers): ModuleList(
          (0): Linear(in_features=32, out_features=28, bias=True)
          (1): ReLU()
        )
      )
    )
  )
)

使用 named_parameters() 方法来理解 Ludwig 模块中包含的参数。该方法返回一个元组列表。元组中的第一个元素是参数的名称。第二个元素是 PyTorch Parameter 对象的实例。在下面的示例中，有 6 个参数，所有参数都是可训练的。

encoder = create_encoder(
    Stacked2DCNN, num_channels=image_size[0], height=image_size[1], width=image_size[2], **encoder_kwargs
)
for p in encoder.named_parameters():
    print(f"name: {p[0]}, shape: {p[1].shape}, trainable: {p[1].requires_grad}")

# output
name: conv_stack_2d.stack.0.layers.0.weight, shape: torch.Size([32, 3, 3, 3]), trainable: True
name: conv_stack_2d.stack.0.layers.0.bias, shape: torch.Size([32]), trainable: True
name: conv_stack_2d.stack.1.layers.0.weight, shape: torch.Size([32, 32, 3, 3]), trainable: True
name: conv_stack_2d.stack.1.layers.0.bias, shape: torch.Size([32]), trainable: True
name: fc_stack.stack.0.layers.0.weight, shape: torch.Size([28, 32]), trainable: True
name: fc_stack.stack.0.layers.0.bias, shape: torch.Size([28]), trainable: True

步骤 2:

一旦 tpc 和 upc 之间的所有差异都得到解释，就可以用适当的 assert 语句替换 print() 语句。以下是一些示例：

    # check for parameter updating
    target = torch.randn(output.shape)
    fpc, tpc, upc, not_updated = check_module_parameters_updated(sequence_rnn_decoder, (combiner_outputs, None), target)
    assert upc == tpc, f"Failed to update parameters.  Parameters not update: {not_updated}"

    target = torch.randn(conv1_stack.output_shape)
    fpc, tpc, upc, not_updated = check_module_parameters_updated(conv1_stack, (input,), target)
    if dropout == 0:
        # all trainable parameters should be updated
        assert tpc == upc, (
            f"All parameter not updated. Parameters not updated: {not_updated}" f"\nModule structure:\n{conv1_stack}"
        )
    else:
        # with specified config and random seed, non-zero dropout update parameter count could take different values
        assert (tpc == upc) or (upc == 1), (
            f"All parameter not updated. Parameters not updated: {not_updated}" f"\nModule structure:\n{conv1_stack}"
        )

包含参数更新检查的完整测试示例:

@pytest.mark.parametrize("cell_type", ["rnn", "gru"])
@pytest.mark.parametrize("num_layers", [1, 2])
@pytest.mark.parametrize("batch_size", [20, 1])
def test_sequence_rnn_decoder(cell_type, num_layers, batch_size):
    hidden_size = 256
    vocab_size = 50
    max_sequence_length = 10

    # make repeatable
    set_random_seed(RANDOM_SEED)

    combiner_outputs = {HIDDEN: torch.rand([batch_size, hidden_size])}
    sequence_rnn_decoder = SequenceRNNDecoder(
        hidden_size, vocab_size, max_sequence_length, cell_type, num_layers=num_layers
    )

    output = sequence_rnn_decoder(combiner_outputs, target=None)

    assert list(output.size()) == [batch_size, max_sequence_length, vocab_size]

    # check for parameter updating
    target = torch.randn(output.shape)
    fpc, tpc, upc, not_updated = check_module_parameters_updated(sequence_rnn_decoder, (combiner_outputs, None), target)
    assert upc == tpc, f"Failed to update parameters.  Parameters not update: {not_updated}"