Skip to content

climate_ref.testing #

Testing utilities for running and validating diagnostic test cases.

This module provides: - Path resolution for package-local test data (catalogs, regression data) - Sample data fetching utilities - TestCaseRunner for executing diagnostics with test data - Result validation helpers

TEST_DATA_DIR = _determine_test_directory() module-attribute #

Path to the centralised test data directory (for sample data).

TestCaseRunner #

Helper class for running diagnostic test cases.

This runner handles: - Running the diagnostic with pre-resolved datasets - Setting up the execution definition

Source code in packages/climate-ref/src/climate_ref/testing.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@define
class TestCaseRunner:
    """
    Helper class for running diagnostic test cases.

    This runner handles:
    - Running the diagnostic with pre-resolved datasets
    - Setting up the execution definition
    """

    config: Config
    datasets: ExecutionDatasetCollection | None = None

    def run(
        self,
        diagnostic: Diagnostic,
        test_case_name: str = "default",
        output_dir: Path | None = None,
        clean: bool = False,
    ) -> ExecutionResult:
        """
        Run a specific test case for a diagnostic.

        Parameters
        ----------
        diagnostic
            The diagnostic to run
        test_case_name
            Name of the test case to run (default: "default")
        output_dir
            Optional output directory for results
        clean
            If True, delete the output directory before running

        Returns
        -------
        ExecutionResult
            The result of running the diagnostic

        Raises
        ------
        NoTestDataSpecError
            If the diagnostic has no test_data_spec
        TestCaseNotFoundError
            If the test case doesn't exist
        DatasetResolutionError
            If datasets cannot be resolved
        """
        if diagnostic.test_data_spec is None:
            raise NoTestDataSpecError(f"Diagnostic {diagnostic.slug} has no test_data_spec")

        if not diagnostic.test_data_spec.has_case(test_case_name):
            raise TestCaseNotFoundError(
                f"Test case {test_case_name!r} not found. Available: {diagnostic.test_data_spec.case_names}"
            )

        if self.datasets is None:
            raise DatasetResolutionError(
                "No datasets provided. Run 'ref test-cases fetch' first to build the catalog."
            )

        # Validate that all non-empty collections have the required 'path' column
        for src_type, collection in self.datasets.items():
            if len(collection.datasets) > 0 and "path" not in collection.datasets.columns:
                raise DatasetResolutionError(
                    f"Datasets for '{src_type}' are missing the required 'path' column. "
                    f"Run 'ref test-cases fetch' to generate the paths file."
                )

        # Determine output directory
        if output_dir is None:
            output_dir = (
                self.config.paths.results
                / "test-cases"
                / diagnostic.provider.slug
                / diagnostic.slug
                / test_case_name
            )

        if clean and output_dir.exists():
            shutil.rmtree(output_dir)

        output_dir.mkdir(parents=True, exist_ok=True)

        definition = ExecutionDefinition(
            diagnostic=diagnostic,
            key=f"test-{test_case_name}",
            datasets=self.datasets,
            output_directory=output_dir,
            root_directory=output_dir.parent,
        )

        # Run the diagnostic.
        # This mirrors ``Diagnostic.run`` but inserts the regression-capture hook before bundling.
        diagnostic.execute(definition)
        diagnostic.prepare_regression_output(definition)
        return diagnostic.build_execution_result(definition)

run(diagnostic, test_case_name='default', output_dir=None, clean=False) #

Run a specific test case for a diagnostic.

Parameters:

Name Type Description Default
diagnostic Diagnostic

The diagnostic to run

 required
test_case_name str

Name of the test case to run (default: "default")

 'default'
output_dir Path | None

Optional output directory for results

 None
clean bool

If True, delete the output directory before running

 False

Returns:

Type Description
ExecutionResult

The result of running the diagnostic

Raises:

Type Description
NoTestDataSpecError

If the diagnostic has no test_data_spec
TestCaseNotFoundError

If the test case doesn't exist
DatasetResolutionError

If datasets cannot be resolved
Source code in packages/climate-ref/src/climate_ref/testing.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def run(
    self,
    diagnostic: Diagnostic,
    test_case_name: str = "default",
    output_dir: Path | None = None,
    clean: bool = False,
) -> ExecutionResult:
    """
    Run a specific test case for a diagnostic.

    Parameters
    ----------
    diagnostic
        The diagnostic to run
    test_case_name
        Name of the test case to run (default: "default")
    output_dir
        Optional output directory for results
    clean
        If True, delete the output directory before running

    Returns
    -------
    ExecutionResult
        The result of running the diagnostic

    Raises
    ------
    NoTestDataSpecError
        If the diagnostic has no test_data_spec
    TestCaseNotFoundError
        If the test case doesn't exist
    DatasetResolutionError
        If datasets cannot be resolved
    """
    if diagnostic.test_data_spec is None:
        raise NoTestDataSpecError(f"Diagnostic {diagnostic.slug} has no test_data_spec")

    if not diagnostic.test_data_spec.has_case(test_case_name):
        raise TestCaseNotFoundError(
            f"Test case {test_case_name!r} not found. Available: {diagnostic.test_data_spec.case_names}"
        )

    if self.datasets is None:
        raise DatasetResolutionError(
            "No datasets provided. Run 'ref test-cases fetch' first to build the catalog."
        )

    # Validate that all non-empty collections have the required 'path' column
    for src_type, collection in self.datasets.items():
        if len(collection.datasets) > 0 and "path" not in collection.datasets.columns:
            raise DatasetResolutionError(
                f"Datasets for '{src_type}' are missing the required 'path' column. "
                f"Run 'ref test-cases fetch' to generate the paths file."
            )

    # Determine output directory
    if output_dir is None:
        output_dir = (
            self.config.paths.results
            / "test-cases"
            / diagnostic.provider.slug
            / diagnostic.slug
            / test_case_name
        )

    if clean and output_dir.exists():
        shutil.rmtree(output_dir)

    output_dir.mkdir(parents=True, exist_ok=True)

    definition = ExecutionDefinition(
        diagnostic=diagnostic,
        key=f"test-{test_case_name}",
        datasets=self.datasets,
        output_directory=output_dir,
        root_directory=output_dir.parent,
    )

    # Run the diagnostic.
    # This mirrors ``Diagnostic.run`` but inserts the regression-capture hook before bundling.
    diagnostic.execute(definition)
    diagnostic.prepare_regression_output(definition)
    return diagnostic.build_execution_result(definition)

fetch_sample_data(force_cleanup=False, symlink=False) #

Fetch the sample data for the given version.

The sample data is produced in the Climate-REF/ref-sample-data repository. This repository contains decimated versions of key datasets used by the diagnostics packages. Decimating these data greatly reduces the data volumes needed to run the test-suite.

Parameters:

Name Type Description Default
force_cleanup bool

If True, remove any existing files

 False
symlink bool

If True, symlink in the data otherwise copy the files

The symlink approach is faster, but will fail when running with a non-local executor because the symlinks can't be followed.

 False
Source code in packages/climate-ref/src/climate_ref/testing.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def fetch_sample_data(force_cleanup: bool = False, symlink: bool = False) -> None:
    """
    Fetch the sample data for the given version.

    The sample data is produced in the [Climate-REF/ref-sample-data](https://github.com/Climate-REF/ref-sample-data)
    repository.
    This repository contains decimated versions of key datasets used by the diagnostics packages.
    Decimating these data greatly reduces the data volumes needed to run the test-suite.

    Parameters
    ----------
    force_cleanup
        If True, remove any existing files
    symlink
        If True, symlink in the data otherwise copy the files

        The symlink approach is faster, but will fail when running with a non-local executor
        because the symlinks can't be followed.
    """

    if TEST_DATA_DIR is None:  # pragma: no cover
        logger.warning("Test data directory not found, skipping sample data fetch")
        return

    sample_data_registry = dataset_registry_manager["sample-data"]

    output_dir = TEST_DATA_DIR / "sample-data"
    version_file = output_dir / "version.txt"
    existing_version = None

    if output_dir.exists():  # pragma: no branch
        if version_file.exists():  # pragma: no branch
            with open(version_file) as fh:
                existing_version = fh.read().strip()

        if force_cleanup or existing_version != SAMPLE_DATA_VERSION:  # pragma: no branch
            logger.warning("Removing existing sample data")
            shutil.rmtree(output_dir)

    fetch_all_files(sample_data_registry, "sample", output_dir, symlink)

    # Write out the current sample data version to the copying as complete
    with open(output_dir / "version.txt", "w") as fh:
        fh.write(SAMPLE_DATA_VERSION)

validate_result(diagnostic, config, result) #

Asserts the correctness of the result of a diagnostic execution

This should only be used by the test suite as it will create a fake database entry for the diagnostic execution result.

Source code in packages/climate-ref/src/climate_ref/testing.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def validate_result(
    diagnostic: Diagnostic, config: Config, result: ExecutionResult
) -> None:  # pragma: no cover
    """
    Asserts the correctness of the result of a diagnostic execution

    This should only be used by the test suite as it will create a fake
    database entry for the diagnostic execution result.
    """
    # TODO: Remove this function once we have moved to using RegressionValidator
    # Add a fake execution/execution group in the Database
    database = Database.from_config(config)
    execution_group = ExecutionGroup(
        diagnostic_id=1, key=result.definition.key, dirty=True, selectors=result.definition.datasets.selectors
    )
    database.session.add(execution_group)
    database.session.flush()

    execution = Execution(
        execution_group_id=execution_group.id,
        dataset_hash=result.definition.datasets.hash,
        output_fragment=PLACEHOLDER_FRAGMENT,
    )
    database.session.add(execution)

    assign_execution_fragment(
        database.session,
        execution,
        provider_slug=diagnostic.provider.slug,
        diagnostic_slug=diagnostic.slug,
        selectors=result.definition.datasets.selectors,
        group_id=execution_group.id,
    )

    assert result.successful

    # Validate CMEC bundles
    validate_cmec_bundles(diagnostic, result)

    # Create a fake log file if one doesn't exist
    if not result.to_output_path("out.log").exists():
        result.to_output_path("out.log").touch()

    # Import late to avoid importing executors,
    # some of which have on-import side effects, at package load time
    from climate_ref.executor import handle_execution_result  # noqa: PLC0415

    # Process and store the result
    handle_execution_result(config, database=database, execution=execution, result=result)