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()

cli_runner(monkeypatch)

Build a REF CLI runner with deterministic output and exit-code checks.

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

@pytest.fixture
def cli_runner(monkeypatch: pytest.MonkeyPatch) -> Callable[..., Result]:
    """Build a REF CLI runner with deterministic output and exit-code checks."""
    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

cmip6_data_catalog(sample_data, sample_data_dir, shared_session_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, shared_session_dir: Path) -> pd.DataFrame:
    """CMIP6 sample data catalog."""
    return _load_or_create_dataframe(
        shared_session_dir / "cmip6-data-catalog.pkl",
        lambda: CMIP6DatasetAdapter().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, obs4ref_data_catalog)

Provide combined data catalog with CMIP6, obs4MIPs and obs4REF 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,
    obs4ref_data_catalog: pd.DataFrame,
) -> dict[SourceDatasetType, pd.DataFrame]:
    """Provide combined data catalog with CMIP6, obs4MIPs and obs4REF sources."""
    return {
        SourceDatasetType.CMIP6: cmip6_data_catalog,
        SourceDatasetType.obs4MIPs: obs4mips_data_catalog,
        SourceDatasetType.obs4REF: obs4ref_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, cli_runner)

Invoke the REF CLI with an isolated configuration and database.

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

@pytest.fixture
def invoke_cli(config: Config, cli_runner: Callable[..., Result]) -> Callable[..., Result]:
    """Invoke the REF CLI with an isolated configuration and database."""
    return cli_runner

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, shared_session_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, shared_session_dir: Path) -> pd.DataFrame:
    """obs4MIPs sample data catalog."""
    return _load_or_create_dataframe(
        shared_session_dir / "obs4mips-data-catalog.pkl",
        lambda: Obs4MIPsDatasetAdapter().find_local_datasets(sample_data_dir / "obs4MIPs"),
    )

obs4ref_data_catalog(sample_data, sample_data_dir, shared_session_dir)

obs4REF sample data catalog.

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

@pytest.fixture(scope="session")
def obs4ref_data_catalog(sample_data: None, sample_data_dir: Path, shared_session_dir: Path) -> pd.DataFrame:
    """obs4REF sample data catalog."""
    return _load_or_create_dataframe(
        shared_session_dir / "obs4ref-data-catalog.pkl",
        lambda: Obs4REFDatasetAdapter().find_local_datasets(sample_data_dir / "obs4REF"),
    )

packaged_ignore_datasets_file() cached

Expose the grey list shipped inside the climate_ref package as a real file.

The path lives for the duration of the test session. For an ordinary install this is the packaged file itself rather than a copy, so treat it as read only.

Returns:

Type	Description
Path	Path to the packaged grey list.

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

@lru_cache(maxsize=1)
def packaged_ignore_datasets_file() -> Path:
    """
    Expose the grey list shipped inside the ``climate_ref`` package as a real file.

    The path lives for the duration of the test session.
    For an ordinary install this is the packaged file itself rather than a copy,
    so treat it as read only.

    Returns
    -------
    :
        Path to the packaged grey list.
    """
    stack = ExitStack()
    atexit.register(stack.close)
    return stack.enter_context(BUNDLED_IGNORE_DATASETS.as_path())

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(shared_session_dir)

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(shared_session_dir: Path) -> 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

    def _fetch() -> None:
        logger.disable("climate_ref_core.dataset_registry")
        try:
            fetch_sample_data(force_cleanup=False, symlink=False)
        finally:
            logger.enable("climate_ref_core.dataset_registry")

    _run_once(shared_session_dir / "sample-data.ready", _fetch)

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"

shared_session_dir(tmp_path_factory, request)

Temporary directory shared by every xdist worker in this pytest run.

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

@pytest.fixture(scope="session")
def shared_session_dir(tmp_path_factory: pytest.TempPathFactory, request: pytest.FixtureRequest) -> Path:
    """Temporary directory shared by every xdist worker in this pytest run."""
    base_temp = tmp_path_factory.getbasetemp()
    if hasattr(request.config, "workerinput"):
        return base_temp.parent
    return base_temp

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