climate_ref.conftest_plugin

Pytest plugin providing shared fixtures for diagnostic provider testing.

This plugin is registered as climate_ref in the pytest11 entry point group. Provider packages can use these fixtures by adding climate-ref[test] to their test dependencies.

Usage in a provider's conftest.py::

pytest_plugins = ("climate_ref.conftest_plugin",)

Or install climate-ref[test] and the plugin is auto-discovered.

Provided fixtures

• config -- per-test Config with isolated directories
• caplog -- loguru-compatible log capture
• test_data_dir / sample_data_dir -- data paths
• sample_data -- session-scoped sample data fetch
• cmip6_data_catalog / obs4mips_data_catalog / data_catalog -- data catalogs
• definition_factory -- create ExecutionDefinition instances
• provider / mock_diagnostic -- mock diagnostic provider
• solved_definition_factory -- build an ExecutionDefinition from the sample-data catalog
• invoke_cli -- CLI test runner

FailedDiagnostic

Bases: Diagnostic

A diagnostic that always fails, for testing.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

class FailedDiagnostic(Diagnostic):
    """A diagnostic that always fails, for testing."""

    name = "failed"
    slug = "failed"
    data_requirements = (DataRequirement(source_type=SourceDatasetType.CMIP6, filters=(), group_by=None),)

    def run(self, definition: ExecutionDefinition, *, capture_regression: bool = False) -> ExecutionResult:
        """Run a diagnostic that always returns a failure result."""
        return ExecutionResult.build_from_failure(definition)

run(definition, *, capture_regression=False)

Run a diagnostic that always returns a failure result.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def run(self, definition: ExecutionDefinition, *, capture_regression: bool = False) -> ExecutionResult:
    """Run a diagnostic that always returns a failure result."""
    return ExecutionResult.build_from_failure(definition)

MockDiagnostic

Bases: Diagnostic

A no-op diagnostic for testing.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

class MockDiagnostic(Diagnostic):
    """A no-op diagnostic for testing."""

    name = "mock"
    slug = "mock"
    data_requirements = (DataRequirement(source_type=SourceDatasetType.CMIP6, filters=(), group_by=None),)

    def run(self, definition: ExecutionDefinition, *, capture_regression: bool = False) -> ExecutionResult:
        """Run a no-op diagnostic that always succeeds."""
        return ExecutionResult(
            output_bundle_filename=definition.output_directory / "output.json",
            metric_bundle_filename=definition.output_directory / "diagnostic.json",
            successful=True,
            definition=definition,
        )

run(definition, *, capture_regression=False)

Run a no-op diagnostic that always succeeds.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def run(self, definition: ExecutionDefinition, *, capture_regression: bool = False) -> ExecutionResult:
    """Run a no-op diagnostic that always succeeds."""
    return ExecutionResult(
        output_bundle_filename=definition.output_directory / "output.json",
        metric_bundle_filename=definition.output_directory / "diagnostic.json",
        successful=True,
        definition=definition,
    )

caplog(caplog)

Capture logs from the loguru default logger.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def caplog(caplog: LogCaptureFixture) -> Iterator[LogCaptureFixture]:
    """Capture logs from the loguru default logger."""

    def filter_(record: dict[str, Any]) -> bool:
        return bool(record["level"].no >= caplog.handler.level)

    add_log_handler(sink=caplog.handler, level=0, format="{message}", filter=filter_)
    yield caplog
    remove_log_handler()

catalog_pid_parser(file, **kwargs)

Record the PID of the process that parsed file (proves out-of-process runs).

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def catalog_pid_parser(file: str, **kwargs: Any) -> dict[str, Any]:
    """Record the PID of the process that parsed ``file`` (proves out-of-process runs)."""
    return {"path": file, "pid": os.getpid()}

catalog_reject_txt_parser(file, **kwargs)

Like :func:catalog_stub_parser but marks .txt files invalid.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def catalog_reject_txt_parser(file: str, **kwargs: Any) -> dict[str, Any]:
    """Like :func:`catalog_stub_parser` but marks ``.txt`` files invalid."""
    if file.endswith(".txt"):
        return {"INVALID_ASSET": file, "TRACEBACK": "not a netCDF file"}
    return {"path": file, "name": Path(file).name}

catalog_stub_parser(file, **kwargs)

Return trivial valid metadata for file without opening it.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def catalog_stub_parser(file: str, **kwargs: Any) -> dict[str, Any]:
    """Return trivial valid metadata for ``file`` without opening it."""
    return {"path": file, "name": Path(file).name}

cleanup_log_handlers(request)

Remove any dangling loguru handlers after each test.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(autouse=True)
def cleanup_log_handlers(request: pytest.FixtureRequest) -> Iterator[None]:
    """Remove any dangling loguru handlers after each test."""
    yield
    if hasattr(logger, "default_handler_id"):
        logger.warning("Logger handler not removed, removing it now")
        remove_log_handler()

cmip6_data_catalog(sample_data, sample_data_dir)

CMIP6 sample data catalog.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def cmip6_data_catalog(sample_data: None, sample_data_dir: Path) -> pd.DataFrame:
    """CMIP6 sample data catalog."""
    adapter = CMIP6DatasetAdapter()
    return adapter.find_local_datasets(sample_data_dir / "CMIP6")

config(tmp_path, monkeypatch, request)

Per-test Config with isolated directories.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Config:
    """Per-test Config with isolated directories."""
    root_output_dir = Path(os.environ.get("REF_TEST_OUTPUT", tmp_path / "climate_ref"))
    dir_name = re.sub(r"[^a-zA-Z0-9_.-]", "_", request.node.name)
    # Include the class name so two identically-named tests in different classes don't
    # collide on a shared REF_TEST_OUTPUT dir (request.node.name omits the class).
    class_segment = request.cls.__name__ if request.cls is not None else ""
    ref_config_dir = root_output_dir / request.module.__name__ / class_segment / dir_name

    software_path = Config.default().paths.software

    monkeypatch.setenv("REF_CONFIGURATION", str(ref_config_dir))
    cfg = Config.default()
    cfg.paths.software = software_path
    cfg.diagnostic_providers = [DiagnosticProviderConfig(provider="climate_ref_example")]
    cfg.executor.executor = "climate_ref.executor.SynchronousExecutor"
    _use_local_ignore_datasets_file(cfg)
    cfg.save()

    return cfg

data_catalog(cmip6_data_catalog, obs4mips_data_catalog)

Provide combined data catalog with CMIP6 and obs4MIPs sources.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def data_catalog(
    cmip6_data_catalog: pd.DataFrame, obs4mips_data_catalog: pd.DataFrame
) -> dict[SourceDatasetType, pd.DataFrame]:
    """Provide combined data catalog with CMIP6 and obs4MIPs sources."""
    return {
        SourceDatasetType.CMIP6: cmip6_data_catalog,
        SourceDatasetType.obs4MIPs: obs4mips_data_catalog,
    }

definition_factory(tmp_path, config)

Create ExecutionDefinition instances for testing.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def definition_factory(tmp_path: Path, config: Config) -> Callable[..., ExecutionDefinition]:
    """Create ExecutionDefinition instances for testing."""

    def _create_definition(
        *,
        diagnostic: Diagnostic,
        execution_dataset_collection: ExecutionDatasetCollection | None = None,
        cmip6: DatasetCollection | None = None,
        obs4mips: DatasetCollection | None = None,
        pmp_climatology: DatasetCollection | None = None,
    ) -> ExecutionDefinition:
        if execution_dataset_collection is None:
            datasets: dict[SourceDatasetType | str, DatasetCollection] = {}
            if cmip6:
                datasets[SourceDatasetType.CMIP6] = cmip6
            if obs4mips:
                datasets[SourceDatasetType.obs4MIPs] = obs4mips
            if pmp_climatology:
                datasets[SourceDatasetType.PMPClimatology] = pmp_climatology
            execution_dataset_collection = ExecutionDatasetCollection(datasets)

        return ExecutionDefinition(
            diagnostic=diagnostic,
            key="key",
            datasets=execution_dataset_collection,
            root_directory=config.paths.scratch,
            output_directory=config.paths.scratch / "output_fragment",
        )

    return _create_definition

esgf_data_catalog(esgf_solve_catalog, test_data_dir)

ESGF metadata catalog for tests that only need DataFrames, not actual files.

Uses pre-generated parquet catalogs from tests/test-data/esgf-catalog/. Fails if the catalog is not available (run scripts/generate_esgf_catalog.py).

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def esgf_data_catalog(
    esgf_solve_catalog: dict[SourceDatasetType, pd.DataFrame] | None,
    test_data_dir: Path,
) -> dict[SourceDatasetType, pd.DataFrame]:
    """
    ESGF metadata catalog for tests that only need DataFrames, not actual files.

    Uses pre-generated parquet catalogs from ``tests/test-data/esgf-catalog/``.
    Fails if the catalog is not available (run ``scripts/generate_esgf_catalog.py``).
    """
    if esgf_solve_catalog is None:
        expected_path = test_data_dir / "esgf-catalog"
        pytest.fail(
            f"ESGF parquet catalog not found in {expected_path}. "
            "Run scripts/generate_esgf_catalog.py to generate it."
        )
    return esgf_solve_catalog

esgf_data_catalog_trimmed(esgf_solve_catalog, test_data_dir)

Trimmed ESGF catalog for integration tests.

Uses a fixed set of CMIP6 source_ids to keep the catalog small and tests fast, while still covering multiple models.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def esgf_data_catalog_trimmed(
    esgf_solve_catalog: dict[SourceDatasetType, pd.DataFrame], test_data_dir: Path
) -> dict[SourceDatasetType, pd.DataFrame]:
    """
    Trimmed ESGF catalog for integration tests.

    Uses a fixed set of CMIP6 source_ids to keep the catalog small and tests fast,
    while still covering multiple models.
    """
    result: dict[SourceDatasetType, pd.DataFrame] = {}
    for source_type, df in esgf_solve_catalog.items():
        if source_type in (SourceDatasetType.CMIP6, SourceDatasetType.CMIP7):
            source_ids = ["ACCESS-ESM1-5", "CESM2", "MPI-ESM1-2-LR", "GFDL-ESM4"]
            result[source_type] = df[df["source_id"].isin(source_ids)]
        else:
            result[source_type] = df

    return result

esgf_solve_catalog(test_data_dir)

Load ESGF metadata catalog for solve tests, if available.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def esgf_solve_catalog(test_data_dir: Path) -> dict[SourceDatasetType, pd.DataFrame] | None:
    """Load ESGF metadata catalog for solve tests, if available."""
    return load_solve_catalog(test_data_dir / "esgf-catalog")

invoke_cli(config, monkeypatch)

Invoke the REF CLI and verify exit code.

Depends on the config fixture so every CLI invocation uses an isolated configuration directory and database rather than the user's real one.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def invoke_cli(config: Config, monkeypatch: pytest.MonkeyPatch) -> Callable[..., Result]:
    """Invoke the REF CLI and verify exit code.

    Depends on the ``config`` fixture so every CLI invocation uses an
    isolated configuration directory and database rather than the user's
    real one.
    """
    runner = CliRunner()
    # Older versions of typer mix stderr and stdout. This option has been removed in newer versions
    runner.mix_stderr = False  # type: ignore[attr-defined]

    def _invoke_cli(args: list[str], expected_exit_code: int = 0, always_log: bool = False) -> Result:
        # Render help/output deterministically for substring assertions.
        # TERM=dumb makes Rich emit no escape codes at all. COLUMNS keeps help from wrapping mid-token.
        monkeypatch.setenv("NO_COLOR", "1")
        monkeypatch.setenv("TERM", "dumb")
        monkeypatch.setenv("COLUMNS", "200")

        result = runner.invoke(app=cli.app, args=args)

        if hasattr(logger, "default_handler_id"):
            remove_log_handler()

        if always_log or result.exit_code != expected_exit_code:
            print("## Command: ", " ".join(args))
            print("Exit code: ", result.exit_code)
            print("Command stdout")
            print(result.stdout)
            print("Command stderr")
            print(result.stderr)
            print("## Command end")

        if result.exit_code != expected_exit_code:
            if result.exception:
                raise result.exception
            raise ValueError(f"Expected exit code {expected_exit_code}, got {result.exit_code}")
        return result

    return _invoke_cli

metric_definition(definition_factory, cmip6_data_catalog, mock_diagnostic)

Create an ExecutionDefinition with a selected CMIP6 dataset for metric tests.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def metric_definition(
    definition_factory: Callable[..., ExecutionDefinition],
    cmip6_data_catalog: pd.DataFrame,
    mock_diagnostic: MockDiagnostic,
) -> ExecutionDefinition:
    """Create an ExecutionDefinition with a selected CMIP6 dataset for metric tests."""
    selected_dataset = cmip6_data_catalog[
        cmip6_data_catalog["instance_id"].isin(
            {
                "CMIP6.ScenarioMIP.CSIRO.ACCESS-ESM1-5.ssp126.r1i1p1f1.Amon.tas.gn.v20210318",
                "CMIP6.ScenarioMIP.CSIRO.ACCESS-ESM1-5.ssp126.r1i1p1f1.fx.areacella.gn.v20210318",
            }
        )
    ]
    collection = ExecutionDatasetCollection(
        {
            SourceDatasetType.CMIP6: DatasetCollection(
                selected_dataset,
                "instance_id",
            )
        }
    )
    return definition_factory(diagnostic=mock_diagnostic, execution_dataset_collection=collection)

mock_diagnostic(provider)

Return the mock diagnostic from the mock provider.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def mock_diagnostic(provider: DiagnosticProvider) -> MockDiagnostic:
    """Return the mock diagnostic from the mock provider."""
    return cast(MockDiagnostic, provider.get("mock"))

obs4mips_data_catalog(sample_data, sample_data_dir)

obs4MIPs sample data catalog.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def obs4mips_data_catalog(sample_data: None, sample_data_dir: Path) -> pd.DataFrame:
    """obs4MIPs sample data catalog."""
    adapter = Obs4MIPsDatasetAdapter()
    obs4ref = adapter.find_local_datasets(sample_data_dir / "obs4REF")
    obs4mips = adapter.find_local_datasets(sample_data_dir / "obs4MIPs")
    return pd.concat([obs4ref, obs4mips], ignore_index=True)

packaged_ignore_datasets_file()

Locate the default_ignore_datasets.yaml shipped inside the climate_ref package.

Returns:

Type	Description
Path	Path to the packaged ignore datasets file.

Raises:

Type	Description
ValueError	If the file is missing from the installed package.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def packaged_ignore_datasets_file() -> Path:
    """
    Locate the ``default_ignore_datasets.yaml`` shipped inside the ``climate_ref`` package.

    Returns
    -------
    :
        Path to the packaged ignore datasets file.

    Raises
    ------
    ValueError
        If the file is missing from the installed package.
    """
    resource = importlib.resources.files("climate_ref") / DEFAULT_IGNORE_DATASETS_FILENAME
    local_ignore_file = Path(str(resource))
    if not local_ignore_file.is_file():  # pragma: no cover - indicates a packaging error
        raise ValueError(f"Could not find ignore file at {local_ignore_file}")
    return local_ignore_file

provider(tmp_path, config)

Create a mock provider with mock and failed diagnostics registered.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def provider(tmp_path: Path, config: Config) -> DiagnosticProvider:
    """Create a mock provider with mock and failed diagnostics registered."""
    provider = DiagnosticProvider("mock_provider", "v0.1.0")
    provider.register(MockDiagnostic())  # type: ignore
    provider.register(FailedDiagnostic())  # type: ignore
    provider.configure(config)
    return provider

pytest_addoption(parser)

Add custom CLI options.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def pytest_addoption(parser: pytest.Parser) -> None:
    """Add custom CLI options."""
    parser.addoption("--slow", action="store_true", help="include tests marked slow")
    parser.addoption("--no-docker", action="store_true", help="skip docker tests")

pytest_collection_modifyitems(config, items)

Skip slow/docker tests unless opted in.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
    """Skip slow/docker tests unless opted in."""
    if not config.getoption("--slow"):
        skip_slow = pytest.mark.skip(reason="need --slow option to run")
        for item in items:
            if item.get_closest_marker("slow"):
                item.add_marker(skip_slow)
    if config.getoption("--no-docker"):
        skip_docker = pytest.mark.skip(reason="--no-docker option provided")
        for item in items:
            if item.get_closest_marker("docker"):
                item.add_marker(skip_docker)

pytest_configure(config)

Register custom markers.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

def pytest_configure(config: pytest.Config) -> None:
    """Register custom markers."""
    config.addinivalue_line("markers", "slow: mark test as slow to run")
    config.addinivalue_line("markers", "docker: mark test requires docker to run")
    config.addinivalue_line("markers", "requires_esgf_data: mark test requires ESGF test data")

sample_data()

Download sample data if not already present.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def sample_data() -> None:
    """Download sample data if not already present."""
    if os.environ.get("REF_TEST_DATA_DIR"):
        logger.warning("Not fetching sample data. Using custom test data directory")
        return
    logger.disable("climate_ref_core.dataset_registry")
    fetch_sample_data(force_cleanup=False, symlink=False)
    logger.enable("climate_ref_core.dataset_registry")

sample_data_dir(sample_data, test_data_dir)

Path to the sample data directory.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def sample_data_dir(sample_data: None, test_data_dir: Path) -> Path:
    """Path to the sample data directory."""
    return test_data_dir / "sample-data"

solve_config()

Session-scoped Config that uses the local default_ignore_datasets.yaml

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def solve_config() -> Config:
    """Session-scoped Config that uses the local default_ignore_datasets.yaml"""
    cfg = Config.default()
    _use_local_ignore_datasets_file(cfg)
    return cfg

solved_definition_factory(config, mocker, data_catalog)

Build an ExecutionDefinition by solving a diagnostic against the sample-data catalog.

Returns a factory that solves the first execution for the given diagnostic and builds its definition under the test scratch directory. Useful for unit tests that need a realistic definition without running the diagnostic (e.g. to inspect the commands it would build).

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture
def solved_definition_factory(
    config: Config,
    mocker: Any,
    data_catalog: dict[SourceDatasetType, pd.DataFrame],
) -> Callable[[Diagnostic], ExecutionDefinition]:
    """
    Build an ``ExecutionDefinition`` by solving a diagnostic against the sample-data catalog.

    Returns a factory that solves the first execution for the given diagnostic and builds
    its definition under the test ``scratch`` directory. Useful for unit tests that need a
    realistic definition without running the diagnostic (e.g. to inspect the commands it
    would build).
    """
    mocker.patch.object(Execution, "execution_group")

    def _build(diagnostic: Diagnostic) -> ExecutionDefinition:
        diagnostic.provider.configure(config)
        execution = next(
            solve_executions(
                data_catalog=data_catalog,
                diagnostic=diagnostic,
                provider=diagnostic.provider,
            )
        )
        return execution.build_execution_definition(output_root=config.paths.scratch)

    return _build

test_data_dir()

Path to the centralised test data directory.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def test_data_dir() -> Path:
    """Path to the centralised test data directory."""
    if TEST_DATA_DIR is None:
        raise ValueError("Test data should exist when running tests")
    return TEST_DATA_DIR

tmp_path_session()

Session-scoped temporary directory.

Source code in packages/climate-ref/src/climate_ref/conftest_plugin.py

@pytest.fixture(scope="session")
def tmp_path_session() -> Iterator[Path]:
    """Session-scoped temporary directory."""
    with tempfile.TemporaryDirectory() as tmpdir:
        yield Path(tmpdir)