Advanced Testing & QA
OGGM 的测试套件是项目质量保障策略的关键组成部分。项目包含着数千行数值计算代码、 复杂的物理参数化方案以及不断增长的用户群,自动化测试能够确保代码变更不会 悄无声息地破坏现有功能。本章将介绍测试架构、测试类别、如何运行测试,以及 如何按照 OGGM 的约定编写新测试。
OGGM 使用 pytest 作为其测试框架。测试基础设施位于
oggm/tests/ 目录中,围绕共享 fixture、辅助函数和模块化的测试文件套件
进行组织。
oggm/tests/
├── __init__.py # Test package marker
├── conftest.py # Shared fixtures and configuration
├── funcs.py # Test helper functions and factories
├── test_workflow.py # End-to-end workflow tests
├── test_models.py # Flowline model correctness tests
├── test_prepro.py # Preprocessing pipeline tests
├── test_numerics.py # Numerical accuracy and verification tests
├── test_utils.py # Utility function tests
├── test_graphics.py # Visualization tests
├── test_benchmarks.py # Performance benchmarks
├── test_shop.py # Shop / data provider tests
├── test_sandbox.py # Sandbox module tests
├── test_minimal.py # Minimal functional (smoke) tests
├── test_workflow.py # High-level workflow integration tests
├── test_glathida.py # GlaThiDa dataset tests
└── test_rgi.py # RGI inventory handling tests
conftest.py 文件提供了自动对所有测试文件可用的 fixture。主要的 fixture 包括:
| Fixture | 作用域 | 用途 |
|---|---|---|
test_dir | session | 用于测试输出的临时目录,自动清理 |
mini_gdir | session | 用于快速单元测试的最小 GlacierDirectory |
hef_gdir | session | 已完成完整预处理的 Hintereisferner GlacierDirectory |
class_gdirs | session | 多个测试冰川的预处理 GlacierDirectory |
cfg_init | function | 确保在每个测试之前调用 cfg.initialize() |
funcs.py 包含可复用的测试工具,包括:
这些测试验证从原始 RGI 轮廓到最终模型输出的完整处理流程。 它们是覆盖范围最广、同时计算代价也最高的测试。主要场景包括:
execute_entity_task。这些测试验证 flowline 演变模型的数值正确性,包括:
def test_flux_based_model_vs_analytical():
"""Compare FluxBasedModel against an analytical solution.
For a constant-width slab with constant mass balance, the SIA
has a known equilibrium profile. This test verifies that the
numerical model converges to within tolerance of that profile.
"""
from oggm.tests.funcs import idealised_gdir
# ...idealised glacier setup...
# Compare numerical equilibrium vs. analytical profile
np.testing.assert_allclose(
modeled.surface_h, analytical.surface_h,
atol=1.0, rtol=0.01
)
class MassConservationChecker:
"""Utility class for verifying mass conservation in flowline models.
Checks:
1. Volume change = integrated mass balance * dt
2. No mass creation or destruction at domain boundaries
3. Mass flux continuity across flowline junctions
"""
def __init__(self, model):
self.model = model
self.initial_volume = self._compute_total_volume()
def check_step(self, dt, mb_model):
"""Verify conservation over one time step."""
v_before = self._compute_total_volume()
self.model.step(dt)
v_after = self._compute_total_volume()
mb_int = self._integrate_mass_balance(dt, mb_model)
# Volume change should match integrated mass balance
np.testing.assert_allclose(
v_after - v_before, mb_int * dt,
rtol=0.1 # allow some tolerance for numerical diffusion
)
def _compute_total_volume(self):
return sum(fl.volume_km3 for fl in self.model.fls) * 1e9
def _integrate_mass_balance(self, dt, mb_model):
total = 0.0
for fl in self.model.fls:
mb = mb_model.get_annual_mb(fl.surface_h)
area = fl.widths_m * fl.dx_meter
total += (mb * area).sum()
return total
def test_cfl_compliance():
"""Verify adaptive time-stepping respects the CFL condition.
The CFL number (u * dt / dx) must be ≤ 1 for explicit schemes
and ≤ ~5 for semi-implicit schemes with the production model.
"""
GIS 和中心线处理流程的测试:
process_dem() 的测试,
涵盖不同的 DEM 来源、分辨率和投影。这些测试验证 OGGM 中使用的各种数值算法的正确性:
通用工具的测试:
monthly_timeseries DataFrame 的
创建、重采样和聚合。pack_config、unpack_config、
参数校验。可视化测试使用 matplotlib 的测试基础设施来验证绘图函数能够无错误地执行 并生成预期的视觉元素:
基准测试确保性能关键的操作保持在可接受的时间范围内。关键指标:
最小功能测试(冒烟测试),能够在 60 秒内运行完毕,验证核心 API 是否可用:
def test_import_oggm():
import oggm
assert oggm.__version__ is not None
def test_cfg_initialize():
from oggm import cfg
cfg.initialize()
assert cfg.PARAMS['continue_on_error'] is False
def test_get_demo_file():
from oggm.utils import get_demo_file
f = get_demo_file('Hintereisferner_RGI5.shp')
assert os.path.exists(f)
创建测试用 GlacierDirectory 是一个常见需求。OGGM 在 funcs.py
中提供了工厂函数:
# Create a minimal GlacierDirectory from a demo shapefile
def get_test_gdir(minimal=False, border=None):
"""Factory for test GlacierDirectory objects.
Parameters
----------
minimal : bool
If True, create a bare GlacierDirectory with just the RGI outline.
If False, run the full preprocessing pipeline.
border : int, optional
Map border in pixels.
Returns
-------
GlacierDirectory
"""
from oggm.utils import get_demo_file, GlacierDirectory
from oggm import workflow
cfg.initialize()
rgi_file = get_demo_file('Hintereisferner_RGI5.shp')
gdir = GlacierDirectory(rgi_file)
if not minimal:
workflow.gis_prepro_tasks(gdir)
return gdir
OGGM 维护一组用于回归测试的参考输出。这些输出由已知正确的代码版本生成, 并以压缩形式存储在测试套件中。当代码发生变更时,回归测试会验证输出结果是否 保持一致(或在容差范围内一致)。
def test_inversion_regression():
"""Verify that the mass-conservation inversion produces consistent results."""
from oggm.core.inversion import mass_conservation_inversion
from oggm.tests.funcs import get_benchmark_data
gdir = get_test_gdir()
# Run inversion
mass_conservation_inversion(gdir)
result = gdir.read_pickle('inversion_output')
# Compare against benchmark
benchmark = get_benchmark_data('inversion_hintereis')
for vn in ['volume', 'thickness']:
v_new = result[vn]
v_ref = benchmark[vn]
np.testing.assert_allclose(
v_new, v_ref, rtol=1e-3,
err_msg=f"{vn} regression failure"
)
OGGM 使用 GitHub Actions 进行持续集成。CI 工作流定义在
.github/workflows/ 目录中,通常包括以下作业:
| 作业 | 触发条件 | 用途 |
|---|---|---|
tests | Push、PR | 在多个 Python 版本(3.9、3.10、3.11)上运行完整测试套件 |
minimal | Push、PR | 冒烟测试;作为快速门禁最先运行 |
docs | Push 到 master | 构建并部署 Sphinx 文档 |
lint | Push、PR | 代码风格检查(flake8、black) |
benchmarks | 定时调度、PR 标签 | 性能回归检测 |
tox 在本地运行与 GitHub Actions 执行的相同的测试矩阵:
# Run all test environments
tox
# Run just the tests for Python 3.10
tox -e py310
# Run linting only
tox -e lint
OGGM 维护一个独立的仓库 OGGM/oggm-sample-data,
其中包含轻量级的测试数据集:
oggm.utils 中的 get_demo_file() 函数负责透明地
下载并缓存这些文件。
oggm/ext/sia_fluxlim.py 模块包含了 SIA 通量限制器(flux limiter)算法的独立
参考实现。这段外部代码不在业务化运行环境中使用,而是作为验证靶标:
# Run the entire test suite
pytest
# Run a specific test file
pytest oggm/tests/test_models.py
# Run a specific test function
pytest oggm/tests/test_models.py::test_flux_based_model_init
# Run tests matching a keyword expression
pytest -k "inversion"
# Run tests with verbose output
pytest -v
# Run tests and stop on first failure
pytest -x
# Run tests in parallel (requires pytest-xdist)
pytest -n 4
OGGM 使用 pytest 标记对测试进行分类:
# Run only slow tests
pytest -m slow
# Skip graphics tests (no display available)
pytest -m "not test_graphics"
# Run only the workflow integration tests
pytest -m workflow
# See all available markers
pytest --markers
| 标记 | 描述 | 典型耗时 |
|---|---|---|
@pytest.mark.slow | 需要大量计算资源的测试 | > 30 秒 |
@pytest.mark.workflow | 端到端工作流集成测试 | > 60 秒 |
@pytest.mark.test_graphics | 需要显示环境 / matplotlib 的测试 | < 10 秒 |
@pytest.mark.test_benchmarks | 性能基准测试 | 不固定 |
import pytest
import numpy as np
import oggm
from oggm import workflow
from oggm.utils import get_demo_file, GlacierDirectory
from oggm.tests.funcs import get_test_gdir
class TestMyNewTask:
"""Test suite for my_new_custom_task."""
@classmethod
def setup_class(cls):
"""Create shared test resources once per class."""
oggm.cfg.initialize()
cls.gdir = get_test_gdir(minimal=True)
def test_task_returns_expected_type(self):
"""The task should return a dict with required keys."""
from my_module import my_new_task
result = my_new_task(self.gdir)
assert isinstance(result, dict)
assert 'required_key_1' in result
assert 'required_key_2' in result
def test_task_output_written_to_disk(self):
"""The task should write output to the glacier directory."""
from my_module import my_new_task
my_new_task(self.gdir)
output_path = self.gdir.get_filepath('my_new_output')
assert os.path.exists(output_path)
def test_task_numerical_accuracy(self):
"""Output values should be within physical bounds."""
from my_module import my_new_task
result = my_new_task(self.gdir)
assert 0 <= result['volume_km3'] <= 1e6 # No glacier > 10^6 km³
def test_task_edge_case_empty_glacier(self):
"""The task should handle glaciers with zero thickness."""
# Create a pathological glacier directory
# ...
from my_module import my_new_task
result = my_new_task(empty_gdir)
assert result['volume_km3'] == 0.0
def test_task_deterministic(self):
"""Same inputs should produce identical outputs."""
from my_module import my_new_task
r1 = my_new_task(self.gdir, seed=42)
r2 = my_new_task(self.gdir, seed=42)
assert r1['value'] == r2['value']
def test_my_numerical_function():
"""Verify numerical correctness against analytical reference."""
from my_module import my_numerical_function
# Known analytical case
x = np.linspace(0, 1, 100)
y_numerical = my_numerical_function(x)
y_analytical = known_analytical_solution(x)
# Relative error should be small
rel_error = np.abs(y_numerical - y_analytical) / (np.abs(y_analytical) + 1e-10)
assert np.max(rel_error) < 1e-6
# Mass/energy should be conserved
integral_numerical = np.trapz(y_numerical, x)
integral_analytical = np.trapz(y_analytical, x)
assert np.abs(integral_numerical - integral_analytical) < 1e-8
OGGM 使用 pytest-cov 跟踪代码覆盖率:
# Run tests with coverage
pytest --cov=oggm --cov-report=html
# View coverage report
# open htmlcov/index.html
虽然 OGGM 不强制要求严格的覆盖率百分比,但遵循以下准则:
额外的质量门禁包括:
setup.cfg 中维护项目特有的配置。