climate_ref.cli.test_cases._common

Helpers shared across several ref test-cases commands.

VerbDriver owns the per-case loop machinery every verb repeats (registry construction, selector validation, case enumeration, skip policy, tally and summary).

VerbCase

Bases: NamedTuple

A test case ready for a verb's loop body, with its paths resolved.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

class VerbCase(NamedTuple):
    """A test case ready for a verb's loop body, with its paths resolved."""

    diag: Diagnostic
    tc: TestCase
    paths: TestCasePaths
    case_id: str

VerbDriver

Shared per-case driver for the ref test-cases verbs.

The loop body stays with the verb and reports via :meth:ok and :meth:fail.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

class VerbDriver:
    """
    Shared per-case driver for the ``ref test-cases`` verbs.

    The loop body stays with the verb and reports via :meth:`ok` and :meth:`fail`.
    """

    def __init__(
        self,
        ctx: typer.Context,
        *,
        provider: str | None,
        diagnostic: str | None,
        test_case: str | None,
    ) -> None:
        from climate_ref.provider_registry import ProviderRegistry

        self.console: Console = ctx.obj.console
        self.provider = provider
        # When a specific case is named, an unusable test case is a hard failure.
        self.named = bool(diagnostic or test_case)
        self.registry = ProviderRegistry.build_from_config(ctx.obj.config, ctx.obj.database)
        _validate_provider_in_registry(self.registry, provider)
        _validate_requested_filters(
            self.registry, provider=provider, diagnostic=diagnostic, test_case=test_case
        )
        self.cases = list(
            _iter_test_cases(self.registry, provider=provider, diagnostic=diagnostic, test_case=test_case)
        )
        self.successes = 0
        self.failures: list[str] = []

    def exit_if_empty(self) -> None:
        """Exit 0 with a warning when the selectors matched no test cases."""
        if self.cases:
            return
        logger.warning(f"No test cases found for provider {self.provider!r}")
        raise typer.Exit(code=0)

    def ready_cases(
        self, *, require_manifest: bool = False, require_catalog: bool = False
    ) -> Iterator[VerbCase]:
        """
        Yield the matched cases with paths resolved, applying the shared skip policy.

        An unlocatable test-data directory or a missing ``manifest.json`` is a hard failure
        when the case was named explicitly, and a warn-and-skip when sweeping.
        A missing catalog is always a hard failure.

        Skips and failures are recorded as the caller iterates,
        so the loop must run to exhaustion for the summary to see them all.
        """
        from climate_ref_core.testing import TestCasePaths

        for diag, tc in self.cases:
            case_id = f"{diag.provider.slug}/{diag.slug}/{tc.name}"
            paths = TestCasePaths.from_diagnostic(diag, tc.name)
            if paths is None:
                self._skip_or_fail(case_id, f"Could not determine test case directory for {case_id}")
                continue
            if require_manifest and not paths.manifest.exists():
                self._skip_or_fail(
                    case_id, f"No manifest.json for {case_id}. Run `ref test-cases mint` first"
                )
                continue
            if require_catalog and not paths.catalog.exists():
                self.fail(case_id, f"No catalog file for {case_id}. Run `ref test-cases fetch` first")
                continue
            yield VerbCase(diag, tc, paths, case_id)

    def ok(self) -> None:
        """Record a successful case."""
        self.successes += 1

    def fail(self, label: str, message: str | None = None) -> None:
        """
        Record a failure, logging ``message`` as an error when given.

        ``label`` is the entry listed under the summary's failed header, usually the case id.
        """
        if message:
            logger.error(message)
        self.failures.append(label)

    def _skip_or_fail(self, case_id: str, message: str) -> None:
        """Fail a case that was named explicitly, and warn-and-skip it when sweeping."""
        if self.named:
            self.fail(case_id, message)
        else:
            logger.warning(message)

    def finish(self, summary: VerbSummary) -> None:
        """Print the verb's summary and exit non-zero when any case failed."""
        self.console.print()

        if self.failures:
            mixed = summary.mixed.format(successes=self.successes, failures=len(self.failures))
            self.console.print(f"[yellow]{mixed}[/yellow]")
            self.console.print(f"[red]{summary.failed_header}[/red]")
            for case in self.failures:
                self.console.print(f"  - {case}")
            raise typer.Exit(code=1)
        self.console.print(f"[green]{summary.success.format(successes=self.successes)}[/green]")

exit_if_empty()

Exit 0 with a warning when the selectors matched no test cases.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

def exit_if_empty(self) -> None:
    """Exit 0 with a warning when the selectors matched no test cases."""
    if self.cases:
        return
    logger.warning(f"No test cases found for provider {self.provider!r}")
    raise typer.Exit(code=0)

fail(label, message=None)

Record a failure, logging message as an error when given.

label is the entry listed under the summary's failed header, usually the case id.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

def fail(self, label: str, message: str | None = None) -> None:
    """
    Record a failure, logging ``message`` as an error when given.

    ``label`` is the entry listed under the summary's failed header, usually the case id.
    """
    if message:
        logger.error(message)
    self.failures.append(label)

finish(summary)

Print the verb's summary and exit non-zero when any case failed.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

def finish(self, summary: VerbSummary) -> None:
    """Print the verb's summary and exit non-zero when any case failed."""
    self.console.print()

    if self.failures:
        mixed = summary.mixed.format(successes=self.successes, failures=len(self.failures))
        self.console.print(f"[yellow]{mixed}[/yellow]")
        self.console.print(f"[red]{summary.failed_header}[/red]")
        for case in self.failures:
            self.console.print(f"  - {case}")
        raise typer.Exit(code=1)
    self.console.print(f"[green]{summary.success.format(successes=self.successes)}[/green]")

ok()

Record a successful case.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

def ok(self) -> None:
    """Record a successful case."""
    self.successes += 1

ready_cases(*, require_manifest=False, require_catalog=False)

Yield the matched cases with paths resolved, applying the shared skip policy.

An unlocatable test-data directory or a missing manifest.json is a hard failure when the case was named explicitly, and a warn-and-skip when sweeping. A missing catalog is always a hard failure.

Skips and failures are recorded as the caller iterates, so the loop must run to exhaustion for the summary to see them all.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

def ready_cases(
    self, *, require_manifest: bool = False, require_catalog: bool = False
) -> Iterator[VerbCase]:
    """
    Yield the matched cases with paths resolved, applying the shared skip policy.

    An unlocatable test-data directory or a missing ``manifest.json`` is a hard failure
    when the case was named explicitly, and a warn-and-skip when sweeping.
    A missing catalog is always a hard failure.

    Skips and failures are recorded as the caller iterates,
    so the loop must run to exhaustion for the summary to see them all.
    """
    from climate_ref_core.testing import TestCasePaths

    for diag, tc in self.cases:
        case_id = f"{diag.provider.slug}/{diag.slug}/{tc.name}"
        paths = TestCasePaths.from_diagnostic(diag, tc.name)
        if paths is None:
            self._skip_or_fail(case_id, f"Could not determine test case directory for {case_id}")
            continue
        if require_manifest and not paths.manifest.exists():
            self._skip_or_fail(
                case_id, f"No manifest.json for {case_id}. Run `ref test-cases mint` first"
            )
            continue
        if require_catalog and not paths.catalog.exists():
            self.fail(case_id, f"No catalog file for {case_id}. Run `ref test-cases fetch` first")
            continue
        yield VerbCase(diag, tc, paths, case_id)

VerbSummary

Bases: NamedTuple

Summary for a per-case verb.

Source code in packages/climate-ref/src/climate_ref/cli/test_cases/_common.py

class VerbSummary(NamedTuple):
    """Summary for a per-case verb."""

    mixed: str
    """Yellow tally line when any case failed, e.g. ``"Replay: {successes} passed, {failures} failed"``."""

    failed_header: str
    """Red header above the failed case list, e.g. ``"Failed replays:"``."""

    success: str
    """Green line when every case succeeded, e.g. ``"All {successes} replay(s) matched ..."``."""

failed_header instance-attribute

Red header above the failed case list, e.g. "Failed replays:".

mixed instance-attribute

Yellow tally line when any case failed, e.g. "Replay: {successes} passed, {failures} failed".

success instance-attribute

Green line when every case succeeded, e.g. "All {successes} replay(s) matched ...".