diff --git a/AGENTS.md b/AGENTS.md index 5ad21e3d45..aeebb3196f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -223,6 +223,12 @@ make lockfiles 10. **Environment Reproducibility**: Set CPU features for consistent results: `export NPY_DISABLE_CPU_FEATURES="AVX512F,AVX512CD,AVX512_SKX"` +11. **Pytest in Conda Env**: If `conda run -n pytest ...` shows NumPy + re-import errors (e.g. "cannot load module more than once per process"), + run tests from an activated shell instead: + `source /opt/conda/etc/profile.d/conda.sh && conda activate && pytest ...` + E.g. this error appears when using pytest-cov. + ## Getting Help diff --git a/lib/iris/fileformats/cf.py b/lib/iris/fileformats/cf.py index 2879e71596..fd639e060d 100644 --- a/lib/iris/fileformats/cf.py +++ b/lib/iris/fileformats/cf.py @@ -1661,6 +1661,7 @@ def _span_check( # those formula terms that are reference surface/phenomenon. for cf_var in self.cf_group.formula_terms.values(): if iris.FUTURE.derived_bounds: + # TODO: is this supposed to be an isinstance()? if self.cf_group[cf_var.cf_name] is CFBoundaryVariable: continue for cf_root, cf_term in cf_var.cf_terms_by_root.items(): diff --git a/lib/iris/tests/test_cf.py b/lib/iris/tests/integration/test_cf.py similarity index 86% rename from lib/iris/tests/test_cf.py rename to lib/iris/tests/integration/test_cf.py index ca066c6c7b..c6d633bef7 100644 --- a/lib/iris/tests/test_cf.py +++ b/lib/iris/tests/integration/test_cf.py @@ -2,10 +2,8 @@ # # This file is part of Iris and is released under the BSD license. # See LICENSE in the root of the repository for full licensing details. -"""Test the cf module.""" +"""Integration tests for :mod:`iris.fileformats.cf`.""" -import contextlib -import io from typing import Iterable import pytest @@ -30,37 +28,6 @@ def fetch_cfvar_data(var, indices=()): return data -class TestCaching: - def test_cached(self, mocker): - # Make sure attribute access to the underlying netCDF4.Variable - # is cached. - name = "foo" - nc_var = mocker.MagicMock() - cf_var = cf.CFAncillaryDataVariable(name, nc_var) - assert nc_var.ncattrs.call_count == 1 - - # Accessing a netCDF attribute should result in no further calls - # to nc_var.ncattrs() and the creation of an attribute on the - # cf_var. - # NB. Can't use hasattr() because that triggers the attribute - # to be created! - assert "coordinates" not in cf_var.__dict__ - _ = cf_var.coordinates - assert nc_var.ncattrs.call_count == 1 - assert "coordinates" in cf_var.__dict__ - - # Trying again results in no change. - _ = cf_var.coordinates - assert nc_var.ncattrs.call_count == 1 - assert "coordinates" in cf_var.__dict__ - - # Trying another attribute results in just a new attribute. - assert "standard_name" not in cf_var.__dict__ - _ = cf_var.standard_name - assert nc_var.ncattrs.call_count == 1 - assert "standard_name" in cf_var.__dict__ - - @_shared_utils.skip_data class TestCFReader: @pytest.fixture(autouse=True) @@ -245,30 +212,6 @@ def test_variable_attribute_touch_pass_0(self): ("units", "degrees_north"), ) - def test_destructor(self, tmp_path): - """Test the destructor when reading the dataset fails. - Related to issue #3312: previously, the `CFReader` would - always call `close()` on its `_dataset` attribute, even if it - didn't exist because opening the dataset had failed. - """ - fn = tmp_path / "tmp.nc" - with fn.open("wb+") as fh: - fh.write(b"\x89HDF\r\n\x1a\nBroken file with correct signature") - fh.flush() - - with io.StringIO() as buf: - with contextlib.redirect_stderr(buf): - try: - _ = cf.CFReader(str(fn)) - except OSError: - pass - try: - _ = iris.load_cubes(str(fn)) - except OSError: - pass - buf.seek(0) - assert buf.read() == "" - @_shared_utils.skip_data class TestLoad: diff --git a/lib/iris/tests/unit/fileformats/cf/conftest.py b/lib/iris/tests/unit/fileformats/cf/conftest.py new file mode 100644 index 0000000000..1eaee0a17e --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/conftest.py @@ -0,0 +1,53 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Shared fixtures for CF fileformat unit tests.""" + +import re + +import numpy as np +import pytest + + +class _NetCDFVariableStub: + """Minimal netCDF-like variable object for CF identify tests.""" + + def __init__(self, name, dtype): + self.name = name + self.dtype = np.dtype(dtype) + + def ncattrs(self): + return [ + attr + for attr in self.__dict__ + if not attr.startswith("_") and attr not in ["name", "dtype"] + ] + + +@pytest.fixture +def named_variable(): + def _factory(name, dtype=int): + return _NetCDFVariableStub(name=name, dtype=dtype) + + return _factory + + +@pytest.fixture +def assert_warning_gated(): + def _assert(operation, warning_category, warning_regex): + with pytest.warns(warning_category, match=warning_regex): + operation(warn=True) + + try: + with pytest.warns(warning_category, match=warning_regex): + operation(warn=False) + except pytest.fail.Exception: + pass + else: + pytest.fail( + f"Operation {operation.__name__} raised {warning_category.__name__} " + "when warn=False" + ) + + return _assert diff --git a/lib/iris/tests/unit/fileformats/cf/identify_catalogue.py b/lib/iris/tests/unit/fileformats/cf/identify_catalogue.py new file mode 100644 index 0000000000..9303eb7be7 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/identify_catalogue.py @@ -0,0 +1,395 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Shared test catalog for CF variable identify() behaviour. + +This module provides reusable pytest test classes which concrete test modules +can subclass and configure via class attributes. +""" + +from abc import ABC +import warnings + +import numpy as np +import pytest + +from iris.fileformats.cf import CFVariable +import iris.warnings + + +class IdentifyByAttributeCatalog(ABC): + """Catalog tests for CF variables identified via a single attribute.""" + + __test__ = False + + CF_CLASS: type[CFVariable] + CF_IDENTITY: str + MISSING_WARN_REGEX: str + + @classmethod + def _set_ref(cls, source_var, value): + setattr(source_var, cls.CF_IDENTITY, value) + + @classmethod + def _expected_var(cls, name, var): + return cls.CF_CLASS(name, var) + + @classmethod + def _make_subject(cls, named_variable, name): + return named_variable(name) + + def test_one_ref(self, named_variable): + subject_name = "ref_subject" + ref_subject = self._make_subject(named_variable, subject_name) + ref_source = named_variable("ref_source") + self._set_ref(ref_source, subject_name) + vars_all = { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + expected = {subject_name: self._expected_var(subject_name, ref_subject)} + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_two_refs(self, named_variable): + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = { + name: self._make_subject(named_variable, name) for name in subject_names + } + + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for ix, var in enumerate(ref_source_vars.values()): + self._set_ref(var, subject_names[ix]) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + expected = { + name: self._expected_var(name, var) + for name, var in ref_subject_vars.items() + } + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_duplicate_refs(self, named_variable): + subject_name = "ref_subject" + ref_subject = self._make_subject(named_variable, subject_name) + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for var in ref_source_vars.values(): + self._set_ref(var, subject_name) + vars_all = { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + **ref_source_vars, + } + + expected = {subject_name: self._expected_var(subject_name, ref_subject)} + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_ignore(self, named_variable): + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = { + name: self._make_subject(named_variable, name) for name in subject_names + } + + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for ix, var in enumerate(ref_source_vars.values()): + self._set_ref(var, subject_names[ix]) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + expected_name = subject_names[0] + expected = { + expected_name: self._expected_var( + expected_name, ref_subject_vars[expected_name] + ) + } + result = self.CF_CLASS.identify(vars_all, ignore=subject_names[1]) + assert expected == result + + def test_target(self, named_variable): + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = { + name: self._make_subject(named_variable, name) for name in subject_names + } + + source_names = ("ref_source_1", "ref_source_2") + ref_source_vars = {name: named_variable(name) for name in source_names} + for ix, var in enumerate(ref_source_vars.values()): + self._set_ref(var, subject_names[ix]) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + expected_name = subject_names[0] + expected = { + expected_name: self._expected_var( + expected_name, ref_subject_vars[expected_name] + ) + } + result = self.CF_CLASS.identify(vars_all, target=source_names[0]) + assert expected == result + + def test_target_unknown_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Cannot identify unknown target CF-netCDF variable 'unknown'" + with pytest.raises(ValueError, match=message): + self.CF_CLASS.identify(vars_all, target="unknown") + + def test_target_wrong_type_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Expect a target CF-netCDF variable name" + with pytest.raises(TypeError, match=message): + self.CF_CLASS.identify(vars_all, target=object()) + + def test_warn(self, named_variable, assert_warning_gated): + subject_name = "ref_subject" + ref_source = named_variable("ref_source") + self._set_ref(ref_source, subject_name) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + def operation(warn: bool): + warnings.warn( + "emit at least 1 warning", + category=iris.warnings.IrisUserWarning, + ) + self.CF_CLASS.identify(vars_all, warn=warn) + + assert_warning_gated( + operation, + iris.warnings.IrisCfMissingVarWarning, + self.MISSING_WARN_REGEX.format(subject=subject_name), + ) + + +class IdentifyByAttributeListCatalog(ABC): + """Catalog tests for UGRID variables identified by one of many attributes.""" + + __test__ = False + + CF_CLASS: type[CFVariable] + CF_IDENTITIES: list[str] + MISSING_WARN_REGEX: str + + @classmethod + def _set_ref(cls, source_var, identity, value): + setattr(source_var, identity, value) + + @classmethod + def _expected_var(cls, name, var): + return cls.CF_CLASS(name, var) + + def test_cf_identities(self, named_variable): + assert self.CF_IDENTITIES + + for identity in self.CF_IDENTITIES: + subject_name = "ref_subject" + ref_subject = named_variable(subject_name) + vars_common = { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + } + expected = {subject_name: self._expected_var(subject_name, ref_subject)} + + ref_source = named_variable("ref_source") + self._set_ref(ref_source, identity, subject_name) + vars_all = dict({"ref_source": ref_source}, **vars_common) + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_duplicate_refs(self, named_variable): + subject_name = "ref_subject" + ref_subject = named_variable(subject_name) + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for var in ref_source_vars.values(): + self._set_ref(var, self.CF_IDENTITIES[0], subject_name) + vars_all = dict( + { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + }, + **ref_source_vars, + ) + + expected = {subject_name: self._expected_var(subject_name, ref_subject)} + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_two_identities(self, named_variable): + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for ix, var in enumerate(ref_source_vars.values()): + self._set_ref(var, self.CF_IDENTITIES[ix], subject_names[ix]) + vars_all = dict( + {"ref_not_subject": named_variable("ref_not_subject")}, + **ref_subject_vars, + **ref_source_vars, + ) + + expected = { + name: self._expected_var(name, var) + for name, var in ref_subject_vars.items() + } + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_two_part_ref(self, named_variable): + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source = named_variable("ref_source") + self._set_ref(ref_source, self.CF_IDENTITIES[0], " ".join(subject_names)) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + **ref_subject_vars, + } + + result = self.CF_CLASS.identify(vars_all) + assert {} == result + + def test_string_type_ignored(self, named_variable): + subject_name = "ref_subject" + ref_source = named_variable("ref_source") + self._set_ref(ref_source, self.CF_IDENTITIES[0], subject_name) + vars_all = { + subject_name: named_variable(subject_name, dtype=np.bytes_), + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + result = self.CF_CLASS.identify(vars_all) + assert {} == result + + def test_ignore(self, named_variable): + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for ix, var in enumerate(ref_source_vars.values()): + self._set_ref(var, self.CF_IDENTITIES[0], subject_names[ix]) + vars_all = dict( + {"ref_not_subject": named_variable("ref_not_subject")}, + **ref_subject_vars, + **ref_source_vars, + ) + + expected_name = subject_names[0] + expected = { + expected_name: self._expected_var( + expected_name, ref_subject_vars[expected_name] + ) + } + result = self.CF_CLASS.identify(vars_all, ignore=subject_names[1]) + assert expected == result + + def test_target(self, named_variable): + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + source_names = ("ref_source_1", "ref_source_2") + ref_source_vars = {name: named_variable(name) for name in source_names} + for ix, var in enumerate(ref_source_vars.values()): + self._set_ref(var, self.CF_IDENTITIES[0], subject_names[ix]) + vars_all = dict( + {"ref_not_subject": named_variable("ref_not_subject")}, + **ref_subject_vars, + **ref_source_vars, + ) + + expected_name = subject_names[0] + expected = { + expected_name: self._expected_var( + expected_name, ref_subject_vars[expected_name] + ) + } + result = self.CF_CLASS.identify(vars_all, target=source_names[0]) + assert expected == result + + def test_target_unknown_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Cannot identify unknown target CF-netCDF variable 'unknown'" + with pytest.raises(ValueError, match=message): + self.CF_CLASS.identify(vars_all, target="unknown") + + def test_target_wrong_type_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Expect a target CF-netCDF variable name" + with pytest.raises(TypeError, match=message): + self.CF_CLASS.identify(vars_all, target=object()) + + def test_warn(self, named_variable, assert_warning_gated): + subject_name = "ref_subject" + ref_source = named_variable("ref_source") + self._set_ref(ref_source, self.CF_IDENTITIES[0], subject_name) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + def operation(warn: bool): + warnings.warn( + "emit at least 1 warning", + category=iris.warnings.IrisUserWarning, + ) + result = self.CF_CLASS.identify(vars_all, warn=warn) + assert {} == result + + assert_warning_gated( + operation, + iris.warnings.IrisCfMissingVarWarning, + self.MISSING_WARN_REGEX.format(subject=subject_name), + ) + + def test_warn_string_type(self, named_variable, assert_warning_gated): + subject_name = "ref_subject" + ref_source = named_variable("ref_source") + self._set_ref(ref_source, self.CF_IDENTITIES[0], subject_name) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + subject_name: named_variable(subject_name, dtype=np.bytes_), + } + + def operation(warn: bool): + warnings.warn( + "emit at least 1 warning", + category=iris.warnings.IrisUserWarning, + ) + result = self.CF_CLASS.identify(vars_all, warn=warn) + assert {} == result + + warn_regex = r".*is a CF-netCDF label variable.*" + assert_warning_gated(operation, iris.warnings.IrisCfLabelVarWarning, warn_regex) diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFAncillaryDataVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFAncillaryDataVariable.py new file mode 100644 index 0000000000..771226c870 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFAncillaryDataVariable.py @@ -0,0 +1,38 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFAncillaryDataVariable`.""" + +from iris.fileformats.cf import CFAncillaryDataVariable + +from .identify_catalogue import IdentifyByAttributeCatalog + + +class TestIdentify(IdentifyByAttributeCatalog): + __test__ = True + + CF_CLASS = CFAncillaryDataVariable + CF_IDENTITY = "ancillary_variables" + MISSING_WARN_REGEX = r"Missing CF-netCDF ancillary data variable {subject!r}.*" + + def test_two_refs(self, named_variable): + # Ancillary vars are commonly referenced as a space-delimited list + # on a single source variable. + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source = named_variable("ref_source") + setattr(ref_source, self.CF_IDENTITY, " ".join(subject_names)) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + **ref_subject_vars, + } + + expected = { + name: self._expected_var(name, var) + for name, var in ref_subject_vars.items() + } + result = self.CF_CLASS.identify(vars_all) + assert expected == result diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFAuxiliaryCoordinateVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFAuxiliaryCoordinateVariable.py new file mode 100644 index 0000000000..cae3aa9545 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFAuxiliaryCoordinateVariable.py @@ -0,0 +1,55 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFAuxiliaryCoordinateVariable`.""" + +import numpy as np + +from iris.fileformats.cf import CFAuxiliaryCoordinateVariable + +from .identify_catalogue import IdentifyByAttributeCatalog + + +class TestIdentify(IdentifyByAttributeCatalog): + __test__ = True + + CF_CLASS = CFAuxiliaryCoordinateVariable + CF_IDENTITY = "coordinates" + MISSING_WARN_REGEX = ( + r"Missing CF-netCDF auxiliary coordinate variable {subject!r}.*" + ) + + def test_two_refs(self, named_variable): + # Auxiliary coordinates can be listed space-delimited on one source. + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source = named_variable("ref_source") + setattr(ref_source, self.CF_IDENTITY, " ".join(subject_names)) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + **ref_subject_vars, + } + + expected = { + name: self._expected_var(name, var) + for name, var in ref_subject_vars.items() + } + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_string_type_ignored(self, named_variable): + # Coordinate-variable identify should reject label/string subjects. + subject_name = "ref_subject" + ref_source = named_variable("ref_source") + self._set_ref(ref_source, subject_name) + vars_all = { + subject_name: named_variable(subject_name, dtype=np.bytes_), + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + result = self.CF_CLASS.identify(vars_all) + assert {} == result diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFBoundaryVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFBoundaryVariable.py new file mode 100644 index 0000000000..82a6c605a3 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFBoundaryVariable.py @@ -0,0 +1,85 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFBoundaryVariable`.""" + +from iris.fileformats.cf import CFBoundaryVariable + +from .identify_catalogue import IdentifyByAttributeCatalog + + +class _NetCDFVarWithDimensions: + """Stub with a dimensions attribute for spans() tests.""" + + def __init__(self, name, dimensions, dtype=int): + import numpy as np + + self.name = name + self.dtype = np.dtype(dtype) + self.dimensions = dimensions + + def ncattrs(self): + return [ + attr + for attr in self.__dict__ + if not attr.startswith("_") and attr not in ["name", "dtype", "dimensions"] + ] + + +class TestIdentify(IdentifyByAttributeCatalog): + __test__ = True + + CF_CLASS = CFBoundaryVariable + CF_IDENTITY = "bounds" + MISSING_WARN_REGEX = r"Missing CF-netCDF boundary variable {subject!r}.*" + + def test_whitespace_padded_ref(self, named_variable): + # CF boundary references accept surrounding whitespace. + subject_name = "ref_subject" + ref_subject = self._make_subject(named_variable, subject_name) + ref_source = named_variable("ref_source") + self._set_ref(ref_source, f" {subject_name} ") + vars_all = { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + expected = {subject_name: self._expected_var(subject_name, ref_subject)} + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + +class TestSpans: + """Tests for CFBoundaryVariable.spans().""" + + def _make_cf_var(self, name, dimensions): + stub = _NetCDFVarWithDimensions(name, dimensions) + return CFBoundaryVariable(name, stub) + + def test_empty_dimensions_spans(self): + """Scalar boundary variable always spans the target.""" + cf_boundary = self._make_cf_var("bounds_var", ()) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_boundary.spans(cf_target) + + def test_source_trailing_subset_spans(self): + """source[:-1] is a subset of target dimensions => spans.""" + # bounds_var has dims (x, y, bounds_extent); data_var has (x, y) + cf_boundary = self._make_cf_var("bounds_var", ("x", "y", "bounds_extent")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_boundary.spans(cf_target) + + def test_source_leading_subset_spans(self): + """source[1:] is a subset of target dimensions => spans.""" + # bounds_var has dims (bounds_extent, x, y); data_var has (x, y) + cf_boundary = self._make_cf_var("bounds_var", ("bounds_extent", "x", "y")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_boundary.spans(cf_target) + + def test_non_spanning(self): + """Dimensions that don't fit either slice => does not span.""" + cf_boundary = self._make_cf_var("bounds_var", ("x", "b", "c")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert not cf_boundary.spans(cf_target) diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFClimatologyVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFClimatologyVariable.py new file mode 100644 index 0000000000..25e980ba94 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFClimatologyVariable.py @@ -0,0 +1,83 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFClimatologyVariable`.""" + +from iris.fileformats.cf import CFClimatologyVariable + +from .identify_catalogue import IdentifyByAttributeCatalog + + +class _NetCDFVarWithDimensions: + """Stub with a dimensions attribute for spans() tests.""" + + def __init__(self, name, dimensions, dtype=int): + import numpy as np + + self.name = name + self.dtype = np.dtype(dtype) + self.dimensions = dimensions + + def ncattrs(self): + return [ + attr + for attr in self.__dict__ + if not attr.startswith("_") and attr not in ["name", "dtype", "dimensions"] + ] + + +class TestIdentify(IdentifyByAttributeCatalog): + __test__ = True + + CF_CLASS = CFClimatologyVariable + CF_IDENTITY = "climatology" + MISSING_WARN_REGEX = r"Missing CF-netCDF climatology variable {subject!r}.*" + + def test_whitespace_padded_ref(self, named_variable): + # CF climatology references accept surrounding whitespace. + subject_name = "ref_subject" + ref_subject = self._make_subject(named_variable, subject_name) + ref_source = named_variable("ref_source") + self._set_ref(ref_source, f" {subject_name} ") + vars_all = { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + expected = {subject_name: self._expected_var(subject_name, ref_subject)} + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + +class TestSpans: + """Tests for CFClimatologyVariable.spans().""" + + def _make_cf_var(self, name, dimensions): + stub = _NetCDFVarWithDimensions(name, dimensions) + return CFClimatologyVariable(name, stub) + + def test_empty_dimensions_spans(self): + """Scalar climatology variable always spans the target.""" + cf_clim = self._make_cf_var("clim_var", ()) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_clim.spans(cf_target) + + def test_source_trailing_subset_spans(self): + """source[:-1] is a subset of target dimensions => spans.""" + cf_clim = self._make_cf_var("clim_var", ("x", "y", "clim_extent")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_clim.spans(cf_target) + + def test_source_leading_subset_spans(self): + """source[1:] is a subset of target dimensions => spans.""" + cf_clim = self._make_cf_var("clim_var", ("clim_extent", "x", "y")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_clim.spans(cf_target) + + def test_non_spanning(self): + """Dimensions that don't fit either slice => does not span.""" + cf_clim = self._make_cf_var("clim_var", ("x", "b", "c")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert not cf_clim.spans(cf_target) diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFCoordinateVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFCoordinateVariable.py new file mode 100644 index 0000000000..e9b1047f56 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFCoordinateVariable.py @@ -0,0 +1,166 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFCoordinateVariable`.""" + +import numpy as np +import numpy.ma as ma +import pytest + +from iris.fileformats.cf import CFCoordinateVariable + + +class _CoordVariableStub: + """Stub for a 1D netCDF variable acting as a coordinate variable.""" + + def __init__(self, name, dimensions, data, dtype=float): + self.name = name + self.dtype = np.dtype(dtype) + self.dimensions = dimensions + self.ndim = len(dimensions) + data_arr = np.asarray(data) + self.shape = data_arr.shape or () + self._data = data + + def ncattrs(self): + return [] + + def __getitem__(self, key): + if self._data.ndim == 0: + return self._data + return self._data[key] + + def __len__(self): + return len(self._data) + + +def _make_coord_var(name, data, dtype=float): + """Helper: create a valid 1D coord variable with name == dimension.""" + return _CoordVariableStub(name=name, dimensions=(name,), data=data, dtype=dtype) + + +class TestIdentify: + def test_valid_coordinate_identified(self): + nc_var = _make_coord_var("lat", [1.0, 2.0, 3.0]) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all) + assert "lat" in result + assert isinstance(result["lat"], CFCoordinateVariable) + + def test_string_dtype_rejected(self): + nc_var = _CoordVariableStub( + name="lat", dimensions=("lat",), data=["a", "b"], dtype=np.bytes_ + ) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all) + assert {} == result + + def test_ndim_not_one_rejected(self): + stub = _CoordVariableStub( + name="lat", dimensions=("lat", "lon"), data=[[1.0, 2.0]], dtype=float + ) + assert stub.ndim == 2 + assert stub.shape == (1, 2) + vars_all = {"lat": stub} + + result = CFCoordinateVariable.identify(vars_all) + assert {} == result + + def test_name_not_in_dimensions_rejected(self): + stub = _CoordVariableStub( + name="lat", dimensions=("x",), data=[1.0, 2.0], dtype=float + ) + vars_all = {"lat": stub} + + result = CFCoordinateVariable.identify(vars_all) + assert {} == result + + def test_ignored_name_excluded(self): + nc_var = _make_coord_var("lat", [1.0, 2.0, 3.0]) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all, ignore=["lat"]) + assert {} == result + + def test_target_filters_to_named_var(self): + lat = _make_coord_var("lat", [1.0, 2.0]) + lon = _make_coord_var("lon", [10.0, 20.0]) + vars_all = {"lat": lat, "lon": lon} + + result = CFCoordinateVariable.identify(vars_all, target="lat") + assert "lat" in result + assert "lon" not in result + + def test_target_unknown_raises(self): + vars_all = {"lat": _make_coord_var("lat", [1.0])} + + message = "Cannot identify unknown target CF-netCDF variable 'unknown'" + with pytest.raises(ValueError, match=message): + CFCoordinateVariable.identify(vars_all, target="unknown") + + def test_target_wrong_type_raises(self): + vars_all = {"lat": _make_coord_var("lat", [1.0])} + + message = "Expect a target CF-netCDF variable name" + with pytest.raises(TypeError, match=message): + CFCoordinateVariable.identify(vars_all, target=object()) + + +class TestIdentifyMonotonic: + def test_monotonic_increasing_accepted(self): + nc_var = _make_coord_var("lat", np.array([1.0, 2.0, 3.0])) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all, monotonic=True) + assert "lat" in result + + def test_monotonic_decreasing_accepted(self): + nc_var = _make_coord_var("lat", np.array([3.0, 2.0, 1.0])) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all, monotonic=True) + assert "lat" in result + + def test_non_monotonic_rejected(self): + nc_var = _make_coord_var("lat", np.array([1.0, 3.0, 2.0])) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all, monotonic=True) + assert {} == result + + def test_scalar_shape_accepted(self): + """Shape () is always accepted under monotonic mode.""" + stub = _make_coord_var("lat", np.float64(1.0)) + vars_all = {"lat": stub} + + result = CFCoordinateVariable.identify(vars_all, monotonic=True) + assert "lat" in result + + def test_single_element_shape_accepted(self): + """Shape (1,) is always accepted under monotonic mode.""" + nc_var = _make_coord_var("lat", np.array([42.0])) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all, monotonic=True) + assert "lat" in result + + def test_masked_array_accepted_when_monotonic(self): + """Masked arrays are filled before monotonic check.""" + data = ma.masked_array([1.0, 2.0, 3.0], mask=[False, False, False]) + nc_var = _make_coord_var("lat", data) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all, monotonic=True) + assert "lat" in result + + def test_masked_array_rejected_when_masked(self): + """Masked arrays are rejected under monotonic mode if any elements are masked.""" + data = ma.masked_array([1.0, 2.0, 3.0], mask=[False, True, False]) + nc_var = _make_coord_var("lat", data) + vars_all = {"lat": nc_var} + + result = CFCoordinateVariable.identify(vars_all, monotonic=True) + assert {} == result diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFDataVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFDataVariable.py new file mode 100644 index 0000000000..a56b80b47a --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFDataVariable.py @@ -0,0 +1,24 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFDataVariable`.""" + +import pytest + +from iris.fileformats.cf import CFDataVariable + + +class TestIdentify: + def test_identify_raises_not_implemented(self, named_variable): + vars_all = {"data_var": named_variable("data_var")} + with pytest.raises(NotImplementedError): + CFDataVariable.identify(vars_all) + + +class TestConstructor: + def test_cf_name_and_data_stored(self, named_variable): + nc_var = named_variable("data_var") + cf_var = CFDataVariable("data_var", nc_var) + assert cf_var.cf_name == "data_var" + assert cf_var.cf_data is nc_var diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFGridMappingVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFGridMappingVariable.py new file mode 100644 index 0000000000..c8a9ab4c3a --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFGridMappingVariable.py @@ -0,0 +1,238 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFGridMappingVariable`.""" + +import warnings + +import pytest + +from iris.fileformats.cf import CFGridMappingVariable +import iris.warnings + +CF_IDENTITY = "grid_mapping" + + +class TestIdentify: + def test_no_coord_system_mappings_returns_empty(self, named_variable): + """When coord_system_mappings is absent, no results.""" + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, "crs_var") + crs_var = named_variable("crs_var") + vars_all = {"crs_var": crs_var, "ref_source": ref_source} + + result = CFGridMappingVariable.identify(vars_all, coord_system_mappings=None) + assert {} == result + + def test_no_mapping_entry_for_source_returns_empty(self, named_variable): + """Data var has grid_mapping attr but no entry in coord_system_mappings.""" + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, "crs_var") + crs_var = named_variable("crs_var") + vars_all = {"crs_var": crs_var, "ref_source": ref_source} + + # Mapping dict exists but has no entry for "ref_source". + result = CFGridMappingVariable.identify( + vars_all, coord_system_mappings={"other_var": {"crs_var": [None]}} + ) + assert {} == result + + def test_simple_mapping_none_coord_identified(self, named_variable): + """A mapping with coord=None (simple grid_mapping style) is accepted.""" + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, "crs_var") + crs_var = named_variable("crs_var") + vars_all = {"crs_var": crs_var, "ref_source": ref_source} + + # {coord_name -> cs_name}; None coord means simple style. + cs_mappings = {"ref_source": {None: "crs_var"}} + + result = CFGridMappingVariable.identify( + vars_all, coord_system_mappings=cs_mappings + ) + assert "crs_var" in result + assert isinstance(result["crs_var"], CFGridMappingVariable) + + def test_valid_coord_ref_identified(self, named_variable): + """Mapping with a real coordinate reference that exists is accepted.""" + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, "crs_var") + crs_var = named_variable("crs_var") + coord_var = named_variable("lat") + vars_all = { + "crs_var": crs_var, + "lat": coord_var, + "ref_source": ref_source, + } + + cs_mappings = {"ref_source": {"lat": "crs_var"}} + + result = CFGridMappingVariable.identify( + vars_all, coord_system_mappings=cs_mappings + ) + assert "crs_var" in result + + def test_missing_mapping_variable_warns(self, named_variable, assert_warning_gated): + """Missing grid mapping variable itself emits a warning.""" + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, "crs_var") + # crs_var is NOT in vars_all + vars_all = {"ref_source": ref_source} + + cs_mappings = {"ref_source": {None: "crs_var"}} + + def operation(warn: bool): + warnings.warn( + "emit at least 1 warning", + category=iris.warnings.IrisUserWarning, + ) + CFGridMappingVariable.identify( + vars_all, coord_system_mappings=cs_mappings, warn=warn + ) + + warn_regex = r"Missing CF-netCDF grid mapping variable 'crs_var'.*" + assert_warning_gated( + operation, iris.warnings.IrisCfMissingVarWarning, warn_regex + ) + + def test_missing_coord_ref_warns(self, named_variable): + """Missing coordinate associated with a grid mapping emits a warning. + + Note: this warning is not gated by the `warn` argument - it is always + emitted when a referenced coordinate variable is absent. + """ + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, "crs_var") + crs_var = named_variable("crs_var") + # lat is NOT in vars_all + vars_all = {"crs_var": crs_var, "ref_source": ref_source} + + cs_mappings = {"ref_source": {"lat": "crs_var"}} + + warn_regex = r"Missing CF-netCDF coordinate variable 'lat'.*" + with pytest.warns(iris.warnings.IrisCfMissingVarWarning, match=warn_regex): + CFGridMappingVariable.identify(vars_all, coord_system_mappings=cs_mappings) + + def test_ignore(self, named_variable): + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, "crs_var") + crs_var = named_variable("crs_var") + vars_all = {"crs_var": crs_var, "ref_source": ref_source} + + cs_mappings = {"ref_source": {None: "crs_var"}} + + result = CFGridMappingVariable.identify( + vars_all, ignore=["crs_var"], coord_system_mappings=cs_mappings + ) + assert {} == result + + def test_target_unknown_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Cannot identify unknown target CF-netCDF variable 'unknown'" + with pytest.raises(ValueError, match=message): + CFGridMappingVariable.identify(vars_all, target="unknown") + + def test_target_wrong_type_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Expect a target CF-netCDF variable name" + with pytest.raises(TypeError, match=message): + CFGridMappingVariable.identify(vars_all, target=object()) + + +class TestIdentifyGroupingByCRS: + """Tests exercising the cs_coord_mappings grouping (lines 698-708 of cf.py). + + The grouping step inverts {coord -> cs_name} into {cs_name -> [coords]}, + so that each unique coordinate system is iterated once. + """ + + def test_multiple_coords_one_cs(self, named_variable): + """Two coordinates both mapping to the same CRS: one CRS identified.""" + ref_source = named_variable("ref_source") + setattr(ref_source, "grid_mapping", "crs_1") + crs_1 = named_variable("crs_1") + lat = named_variable("lat") + lon = named_variable("lon") + vars_all = { + "crs_1": crs_1, + "lat": lat, + "lon": lon, + "ref_source": ref_source, + } + + # Both lat and lon reference the same coordinate system. + cs_mappings = {"ref_source": {"lat": "crs_1", "lon": "crs_1"}} + + result = CFGridMappingVariable.identify( + vars_all, coord_system_mappings=cs_mappings + ) + assert list(result.keys()) == ["crs_1"] + + def test_multiple_cs_both_identified(self, named_variable): + """Two coordinates each referencing a different CRS: both CRSs identified.""" + ref_source = named_variable("ref_source") + setattr(ref_source, "grid_mapping", "crs_1 crs_2") + crs_1 = named_variable("crs_1") + crs_2 = named_variable("crs_2") + lat = named_variable("lat") + height = named_variable("height") + vars_all = { + "crs_1": crs_1, + "crs_2": crs_2, + "lat": lat, + "height": height, + "ref_source": ref_source, + } + + cs_mappings = {"ref_source": {"lat": "crs_1", "height": "crs_2"}} + + result = CFGridMappingVariable.identify( + vars_all, coord_system_mappings=cs_mappings + ) + assert set(result.keys()) == {"crs_1", "crs_2"} + + def test_partial_coords_missing_cs_still_identified(self, named_variable): + """One coord present, one missing for the same CRS: CRS still identified + (has_a_valid_coord is True from the present coord), but a warning is + issued for the missing one. + """ + ref_source = named_variable("ref_source") + setattr(ref_source, "grid_mapping", "crs_1") + crs_1 = named_variable("crs_1") + lat = named_variable("lat") + # lon is intentionally absent from vars_all + vars_all = { + "crs_1": crs_1, + "lat": lat, + "ref_source": ref_source, + } + + cs_mappings = {"ref_source": {"lat": "crs_1", "lon": "crs_1"}} + + warn_regex = r"Missing CF-netCDF coordinate variable 'lon'.*" + with pytest.warns(iris.warnings.IrisCfMissingVarWarning, match=warn_regex): + result = CFGridMappingVariable.identify( + vars_all, coord_system_mappings=cs_mappings + ) + # CRS still in result because lat was valid. + assert "crs_1" in result + + def test_all_coords_missing_cs_excluded(self, named_variable): + """All coords missing for a CRS: has_a_valid_coord stays False, CRS excluded.""" + ref_source = named_variable("ref_source") + setattr(ref_source, "grid_mapping", "crs_1") + crs_1 = named_variable("crs_1") + # Both lat and lon absent from vars_all + vars_all = {"crs_1": crs_1, "ref_source": ref_source} + + cs_mappings = {"ref_source": {"lat": "crs_1", "lon": "crs_1"}} + + warn_regex = "Missing CF-netCDF coordinate variable" + with pytest.warns(iris.warnings.IrisCfMissingVarWarning, match=warn_regex): + result = CFGridMappingVariable.identify( + vars_all, coord_system_mappings=cs_mappings + ) + assert {} == result diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFGroup.py b/lib/iris/tests/unit/fileformats/cf/test_CFGroup.py index 3724a2f628..8d5b81243d 100644 --- a/lib/iris/tests/unit/fileformats/cf/test_CFGroup.py +++ b/lib/iris/tests/unit/fileformats/cf/test_CFGroup.py @@ -4,108 +4,183 @@ # See LICENSE in the root of the repository for full licensing details. """Unit tests for the :class:`iris.fileformats.cf.CFGroup` class.""" +from typing import Callable from unittest.mock import MagicMock import pytest -from iris.fileformats.cf import ( - CFAuxiliaryCoordinateVariable, - CFCoordinateVariable, - CFDataVariable, - CFGroup, - CFUGridAuxiliaryCoordinateVariable, - CFUGridConnectivityVariable, - CFUGridMeshVariable, +from iris.fileformats import cf + +VariableMap: dict[type[cf.CFVariable], Callable] = { + cf._CFFormulaTermsVariable: cf.CFGroup.formula_terms, + cf.CFAncillaryDataVariable: cf.CFGroup.ancillary_variables, + cf.CFAuxiliaryCoordinateVariable: cf.CFGroup.auxiliary_coordinates, + cf.CFBoundaryVariable: cf.CFGroup.bounds, + cf.CFClimatologyVariable: cf.CFGroup.climatology, + cf.CFCoordinateVariable: cf.CFGroup.coordinates, + cf.CFDataVariable: cf.CFGroup.data_variables, + cf.CFGridMappingVariable: cf.CFGroup.grid_mappings, + cf.CFLabelVariable: cf.CFGroup.labels, + cf.CFMeasureVariable: cf.CFGroup.cell_measures, + cf.CFUGridAuxiliaryCoordinateVariable: cf.CFGroup.ugrid_coords, + cf.CFUGridConnectivityVariable: cf.CFGroup.connectivities, + cf.CFUGridMeshVariable: cf.CFGroup.meshes, +} + + +def get_mocked_var(class_: type[cf.CFVariable], mocker) -> MagicMock: + cf_name = f"{class_.__name__}_var" + mocked = mocker.MagicMock(spec=class_, cf_name=cf_name) + return mocked + + +@pytest.fixture( + params=[ + (class_, scenario) + for class_ in VariableMap + for scenario in ("single", "duplicates", "multiple") + ], + ids=lambda value: f"{value[0].__name__}-{value[1]}", ) +def mock_variables( + request, mocker +) -> tuple[type[cf.CFVariable], tuple[MagicMock, ...]]: + class_, scenario = request.param + mocked = get_mocked_var(class_, mocker) + variables = [mocked] + if scenario == "duplicates": + variables.append(get_mocked_var(class_, mocker)) + elif scenario == "multiple": + mocked_2 = get_mocked_var(class_, mocker) + mocked_2.cf_name = f"{mocked.cf_name}_2" + variables.append(mocked_2) + + return class_, tuple(variables) + + +@pytest.fixture(params=["single", "duplicates", "multiple"]) +def mock_variables_all(request, mocker) -> tuple[MagicMock, ...]: + scenario = request.param + all_variables = [] + + for class_ in VariableMap: + mocked = get_mocked_var(class_, mocker) + variables = [mocked] + + if scenario == "duplicates": + variables.append(get_mocked_var(class_, mocker)) + elif scenario == "multiple": + mocked_2 = get_mocked_var(class_, mocker) + mocked_2.cf_name = f"{mocked.cf_name}_2" + variables.append(mocked_2) + + all_variables.extend(variables) + + return tuple(all_variables) + + +@pytest.fixture +def cf_group() -> cf.CFGroup: + return cf.CFGroup() -class Tests: - # TODO: unit tests for existing functionality pre 2021-03-11. - @pytest.fixture(autouse=True) - def _setup(self): - self.cf_group = CFGroup() - def test_non_data_names(self): - data_var = MagicMock(spec=CFDataVariable, cf_name="data_var") - aux_var = MagicMock(spec=CFAuxiliaryCoordinateVariable, cf_name="aux_var") - coord_var = MagicMock(spec=CFCoordinateVariable, cf_name="coord_var") - coord_var2 = MagicMock(spec=CFCoordinateVariable, cf_name="coord_var2") - duplicate_name_var = MagicMock(spec=CFCoordinateVariable, cf_name="aux_var") +@pytest.fixture +def cf_group_populated(cf_group, mock_variables_all) -> cf.CFGroup: + for mocked in mock_variables_all: + cf_group[mocked.cf_name] = mocked + return cf_group - for var in ( - data_var, - aux_var, - coord_var, - coord_var2, - duplicate_name_var, - ): - self.cf_group[var.cf_name] = var - expected_names = [var.cf_name for var in (aux_var, coord_var, coord_var2)] - expected = set(expected_names) - assert self.cf_group.non_data_variable_names == expected +class TestProperties: + def test_common(self, cf_group, mock_variables): + class_, variables = mock_variables + for mocked in variables: + cf_group[mocked.cf_name] = mocked + property_ = VariableMap[class_] + result = property_.fget(cf_group) + expected_names = {var.cf_name for var in variables} + assert len(result) == len(expected_names) + for expected_name in expected_names: + assert expected_name in result -class TestUgrid: - """Separate class to test UGRID functionality.""" + def test_non_data_names(self, cf_group_populated, mock_variables_all): + expected = { + mocked.cf_name + for mocked in mock_variables_all + if not isinstance(mocked, (cf.CFDataVariable, cf._CFFormulaTermsVariable)) + } + assert cf_group_populated.non_data_variable_names == expected + +class TestReturns: @pytest.fixture(autouse=True) - def _setup(self): - self.cf_group = CFGroup() - - def test_inherited(self): - coord_var = MagicMock(spec=CFCoordinateVariable, cf_name="coord_var") - self.cf_group[coord_var.cf_name] = coord_var - assert self.cf_group.coordinates[coord_var.cf_name] == coord_var - - def test_connectivities(self): - conn_var = MagicMock(spec=CFUGridConnectivityVariable, cf_name="conn_var") - self.cf_group[conn_var.cf_name] = conn_var - assert self.cf_group.connectivities[conn_var.cf_name] == conn_var - - def test_ugrid_coords(self): - coord_var = MagicMock( - spec=CFUGridAuxiliaryCoordinateVariable, cf_name="coord_var" - ) - self.cf_group[coord_var.cf_name] = coord_var - assert self.cf_group.ugrid_coords[coord_var.cf_name] == coord_var - - def test_meshes(self): - mesh_var = MagicMock(spec=CFUGridMeshVariable, cf_name="mesh_var") - self.cf_group[mesh_var.cf_name] = mesh_var - assert self.cf_group.meshes[mesh_var.cf_name] == mesh_var - - def test_non_data_names(self): - data_var = MagicMock(spec=CFDataVariable, cf_name="data_var") - coord_var = MagicMock(spec=CFCoordinateVariable, cf_name="coord_var") - conn_var = MagicMock(spec=CFUGridConnectivityVariable, cf_name="conn_var") - ugrid_coord_var = MagicMock( - spec=CFUGridAuxiliaryCoordinateVariable, cf_name="ugrid_coord_var" + def _setup(self, cf_group_populated, mock_variables_all): + self.variables = mock_variables_all + self.cf_group = cf_group_populated + + def test_keys(self): + expected_names = {mocked.cf_name for mocked in self.variables} + assert set(self.cf_group.keys()) == expected_names + + def test_len(self): + expected_names = {mocked.cf_name for mocked in self.variables} + assert len(self.cf_group) == len(expected_names) + + def test_iter(self): + expected_names = {mocked.cf_name for mocked in self.variables} + actual_names = set() + for name in self.cf_group: + actual_names.add(name) + + assert actual_names == expected_names + + def test_getitem(self): + expected_by_name = {} + for mocked in self.variables: + expected_by_name[mocked.cf_name] = mocked + + for expected_name, expected_variable in expected_by_name.items(): + assert self.cf_group[expected_name] is expected_variable + + with pytest.raises(KeyError, match="Cannot get unknown CF-netCDF variable"): + _ = self.cf_group["unknown_name"] + + def test_repr(self): + self.cf_group.global_attributes["global_attr"] = "value" + self.cf_group.promoted["promoted_var"] = self.variables[0] + + expected_names = {mocked.cf_name for mocked in self.variables} + expected = ( + "" ) - mesh_var = MagicMock(spec=CFUGridMeshVariable, cf_name="mesh_var") - mesh_var2 = MagicMock(spec=CFUGridMeshVariable, cf_name="mesh_var2") - duplicate_name_var = MagicMock(spec=CFUGridMeshVariable, cf_name="coord_var") - - for var in ( - data_var, - coord_var, - conn_var, - ugrid_coord_var, - mesh_var, - mesh_var2, - duplicate_name_var, - ): - self.cf_group[var.cf_name] = var - - expected_names = [ - var.cf_name - for var in ( - coord_var, - conn_var, - ugrid_coord_var, - mesh_var, - mesh_var2, - ) - ] - expected = set(expected_names) - assert self.cf_group.non_data_variable_names == expected + assert repr(self.cf_group) == expected + + +class TestMutations: + def test_setitem(self, cf_group, mock_variables_all): + mocked = mock_variables_all[0] + + cf_group[mocked.cf_name] = mocked + assert cf_group[mocked.cf_name] is mocked + + with pytest.raises(TypeError, match="Attempted to add an invalid"): + cf_group[mocked.cf_name] = object() + + with pytest.raises(ValueError, match="Mismatch between key name"): + cf_group[f"{mocked.cf_name}_mismatch"] = mocked + + def test_delitem(self, cf_group_populated, mock_variables_all): + expected_names = {mocked.cf_name for mocked in mock_variables_all} + name_to_delete = next(iter(expected_names)) + + del cf_group_populated[name_to_delete] + assert name_to_delete not in cf_group_populated + assert len(cf_group_populated) == len(expected_names) - 1 + + with pytest.raises(KeyError, match="Cannot delete unknown CF-netcdf"): + del cf_group_populated["unknown_name"] diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFLabelVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFLabelVariable.py new file mode 100644 index 0000000000..4e355c17ce --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFLabelVariable.py @@ -0,0 +1,137 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFLabelVariable`.""" + +import numpy as np +import pytest + +from iris.fileformats.cf import CFDataVariable, CFLabelVariable + +from .identify_catalogue import IdentifyByAttributeCatalog + + +class _NetCDFVarWithDimensions: + """Stub with a dimensions attribute for spans() and cf_label_dimensions() tests.""" + + def __init__(self, name, dimensions, dtype=int): + self.name = name + self.dtype = np.dtype(dtype) + self.dimensions = dimensions + + def ncattrs(self): + return [ + attr + for attr in self.__dict__ + if not attr.startswith("_") and attr not in ["name", "dtype", "dimensions"] + ] + + +class TestIdentify(IdentifyByAttributeCatalog): + __test__ = True + + CF_CLASS = CFLabelVariable + CF_IDENTITY = "coordinates" + MISSING_WARN_REGEX = r"Missing CF-netCDF label variable {subject!r}.*" + + @classmethod + def _make_subject(cls, named_variable, name): + # Label identify expects referenced variables to be string-typed. + return named_variable(name, dtype=np.bytes_) + + def test_two_refs(self, named_variable): + # Label coordinates may be listed space-delimited on a single source. + subject_names = ("ref_subject_1", "ref_subject_2") + ref_subject_vars = { + name: self._make_subject(named_variable, name) for name in subject_names + } + + ref_source = named_variable("ref_source") + setattr(ref_source, self.CF_IDENTITY, " ".join(subject_names)) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + **ref_subject_vars, + } + + expected = { + name: self._expected_var(name, var) + for name, var in ref_subject_vars.items() + } + result = self.CF_CLASS.identify(vars_all) + assert expected == result + + def test_non_string_ref_ignored(self, named_variable): + # Label identify should reject non-string referenced variables. + subject_name = "ref_subject" + ref_source = named_variable("ref_source") + self._set_ref(ref_source, subject_name) + vars_all = { + subject_name: named_variable(subject_name, dtype=int), + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + result = self.CF_CLASS.identify(vars_all) + assert {} == result + + +class TestCfLabelDimensions: + def _make_label_var(self, label_dims): + stub = _NetCDFVarWithDimensions("label_var", label_dims, dtype=np.bytes_) + return CFLabelVariable("label_var", stub) + + def _make_data_var(self, data_dims): + stub = _NetCDFVarWithDimensions("data_var", data_dims) + return CFDataVariable("data_var", stub) + + def test_raises_for_non_cfdata_var(self): + label_var = self._make_label_var(("x",)) + message = "cf_data_var argument should be of type CFDataVariable" + with pytest.raises(TypeError, match=message): + label_var.cf_label_dimensions(object()) + + def test_returns_overlap_dimensions(self): + label_var = self._make_label_var(("x", "strlen")) + data_var = self._make_data_var(("x", "y")) + result = label_var.cf_label_dimensions(data_var) + assert result == ("x",) + + def test_no_overlap_returns_empty(self): + label_var = self._make_label_var(("a", "b")) + data_var = self._make_data_var(("x", "y")) + result = label_var.cf_label_dimensions(data_var) + assert result == () + + +class TestSpans: + """Tests for CFLabelVariable.spans().""" + + def _make_cf_var(self, name, dimensions): + stub = _NetCDFVarWithDimensions(name, dimensions, dtype=np.bytes_) + return CFLabelVariable(name, stub) + + def test_empty_dimensions_spans(self): + """Scalar label variable always spans the target.""" + cf_label = self._make_cf_var("label_var", ()) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_label.spans(cf_target) + + def test_source_trailing_subset_spans(self): + """source[:-1] (drop string length dim) is subset of target => spans.""" + cf_label = self._make_cf_var("label_var", ("x", "strlen")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_label.spans(cf_target) + + def test_source_leading_subset_spans(self): + """source[1:] is a subset of target dimensions => spans.""" + cf_label = self._make_cf_var("label_var", ("strlen", "x")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert cf_label.spans(cf_target) + + def test_non_spanning(self): + """Dimensions that don't fit either slice => does not span.""" + cf_label = self._make_cf_var("label_var", ("x", "b", "c")) + cf_target = self._make_cf_var("data_var", ("x", "y")) + assert not cf_label.spans(cf_target) diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFMeasureVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFMeasureVariable.py new file mode 100644 index 0000000000..a344dd2b7d --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFMeasureVariable.py @@ -0,0 +1,180 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFMeasureVariable`.""" + +import warnings + +import pytest + +from iris.fileformats.cf import CFMeasureVariable +import iris.warnings + +CF_IDENTITY = "cell_measures" + + +class TestIdentify: + def test_one_measure_ref(self, named_variable): + subject_name = "ref_subject" + ref_subject = named_variable(subject_name) + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, f"area: {subject_name}") + vars_all = { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + expected = {subject_name: CFMeasureVariable(subject_name, ref_subject, "area")} + result = CFMeasureVariable.identify(vars_all) + assert expected == result + + def test_measure_stored_on_instance(self, named_variable): + subject_name = "ref_subject" + ref_subject = named_variable(subject_name) + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, f"volume: {subject_name}") + vars_all = { + subject_name: ref_subject, + "ref_source": ref_source, + } + + result = CFMeasureVariable.identify(vars_all) + assert result[subject_name].cf_measure == "volume" + + def test_multi_term(self, named_variable): + subject_names = ("ref_area", "ref_volume") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + ref_source = named_variable("ref_source") + setattr( + ref_source, + CF_IDENTITY, + f"area: {subject_names[0]} volume: {subject_names[1]}", + ) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + **ref_subject_vars, + } + + result = CFMeasureVariable.identify(vars_all) + assert set(result.keys()) == set(subject_names) + assert result[subject_names[0]].cf_measure == "area" + assert result[subject_names[1]].cf_measure == "volume" + + def test_two_refs(self, named_variable): + """Two source variables each referencing a different measure variable.""" + subject_names = ("ref_area", "ref_volume") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for ix, var in enumerate(ref_source_vars.values()): + setattr(var, CF_IDENTITY, f"area: {subject_names[ix]}") + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + expected = { + name: CFMeasureVariable(name, var, "area") + for name, var in ref_subject_vars.items() + } + result = CFMeasureVariable.identify(vars_all) + assert expected == result + + def test_self_reference_ignored(self, named_variable): + """A variable cannot reference itself as a cell measure.""" + nc_var = named_variable("self_ref") + setattr(nc_var, CF_IDENTITY, "area: self_ref") + vars_all = { + "self_ref": nc_var, + } + + result = CFMeasureVariable.identify(vars_all) + assert {} == result + + def test_ignore(self, named_variable): + subject_names = ("ref_area", "ref_volume") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for ix, var in enumerate(ref_source_vars.values()): + setattr(var, CF_IDENTITY, f"area: {subject_names[ix]}") + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + expected_name = subject_names[0] + expected = { + expected_name: CFMeasureVariable( + expected_name, ref_subject_vars[expected_name], "area" + ) + } + result = CFMeasureVariable.identify(vars_all, ignore=subject_names[1]) + assert expected == result + + def test_target(self, named_variable): + subject_names = ("ref_area", "ref_volume") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + source_names = ("ref_source_1", "ref_source_2") + ref_source_vars = {name: named_variable(name) for name in source_names} + for ix, var in enumerate(ref_source_vars.values()): + setattr(var, CF_IDENTITY, f"area: {subject_names[ix]}") + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + expected_name = subject_names[0] + expected = { + expected_name: CFMeasureVariable( + expected_name, ref_subject_vars[expected_name], "area" + ) + } + result = CFMeasureVariable.identify(vars_all, target=source_names[0]) + assert expected == result + + def test_target_unknown_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Cannot identify unknown target CF-netCDF variable 'unknown'" + with pytest.raises(ValueError, match=message): + CFMeasureVariable.identify(vars_all, target="unknown") + + def test_target_wrong_type_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Expect a target CF-netCDF variable name" + with pytest.raises(TypeError, match=message): + CFMeasureVariable.identify(vars_all, target=object()) + + def test_warn(self, named_variable, assert_warning_gated): + subject_name = "ref_subject" + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, f"area: {subject_name}") + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + def operation(warn: bool): + warnings.warn( + "emit at least 1 warning", + category=iris.warnings.IrisUserWarning, + ) + CFMeasureVariable.identify(vars_all, warn=warn) + + warn_regex = rf"Missing CF-netCDF measure variable {subject_name!r}.*" + assert_warning_gated( + operation, iris.warnings.IrisCfMissingVarWarning, warn_regex + ) diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFReader.py b/lib/iris/tests/unit/fileformats/cf/test_CFReader.py index c5d0e068bf..c160b3c21a 100644 --- a/lib/iris/tests/unit/fileformats/cf/test_CFReader.py +++ b/lib/iris/tests/unit/fileformats/cf/test_CFReader.py @@ -4,12 +4,18 @@ # See LICENSE in the root of the repository for full licensing details. """Unit tests for the `iris.fileformats.cf.CFReader` class.""" -from unittest import mock +import contextlib +import io import numpy as np import pytest +import iris +import iris.exceptions +from iris.fileformats import cf from iris.fileformats.cf import ( + CFAuxiliaryCoordinateVariable, + CFBoundaryVariable, CFCoordinateVariable, CFDataVariable, CFGridMappingVariable, @@ -19,9 +25,11 @@ CFUGridConnectivityVariable, CFUGridMeshVariable, ) +import iris.warnings def netcdf_variable( + mocker, name, dimensions, dtype, @@ -47,10 +55,10 @@ def netcdf_variable( + CFUGridConnectivityVariable.cf_identities + [CFUGridMeshVariable.cf_identity] ) - ncvar = mock.Mock( + ncvar = mocker.Mock( name=name, dimensions=dimensions, - ncattrs=mock.Mock(return_value=[]), + ncattrs=mocker.Mock(return_value=[]), ndim=ndim, dtype=dtype, ancillary_variables=ancillary_variables, @@ -69,10 +77,10 @@ def netcdf_variable( class Test_translate__global_attributes: @pytest.fixture(autouse=True) def _setup(self, mocker): - ncvar = netcdf_variable("ncvar", "height", np.float64) - ncattrs = mock.Mock(return_value=["dimensions"]) - getncattr = mock.Mock(return_value="something something_else") - dataset = mock.Mock( + ncvar = netcdf_variable(mocker, "ncvar", "height", np.float64) + ncattrs = mocker.Mock(return_value=["dimensions"]) + getncattr = mocker.Mock(return_value="something something_else") + dataset = mocker.Mock( file_format="NetCDF4", variables={"ncvar": ncvar}, ncattrs=ncattrs, @@ -91,14 +99,23 @@ def test_create_global_attributes(self, mocker): class Test_translate__formula_terms: @pytest.fixture(autouse=True) def _setup(self, mocker): - self.delta = netcdf_variable("delta", "height", np.float64, bounds="delta_bnds") - self.delta_bnds = netcdf_variable("delta_bnds", "height bnds", np.float64) - self.sigma = netcdf_variable("sigma", "height", np.float64, bounds="sigma_bnds") - self.sigma_bnds = netcdf_variable("sigma_bnds", "height bnds", np.float64) - self.orography = netcdf_variable("orography", "lat lon", np.float64) + self.delta = netcdf_variable( + mocker, "delta", "height", np.float64, bounds="delta_bnds" + ) + self.delta_bnds = netcdf_variable( + mocker, "delta_bnds", "height bnds", np.float64 + ) + self.sigma = netcdf_variable( + mocker, "sigma", "height", np.float64, bounds="sigma_bnds" + ) + self.sigma_bnds = netcdf_variable( + mocker, "sigma_bnds", "height bnds", np.float64 + ) + self.orography = netcdf_variable(mocker, "orography", "lat lon", np.float64) formula_terms = "a: delta b: sigma orog: orography" standard_name = "atmosphere_hybrid_height_coordinate" self.height = netcdf_variable( + mocker, "height", "height", np.float64, @@ -110,16 +127,17 @@ def _setup(self, mocker): # which will be ignored by the cf loader. formula_terms = "a: delta_bnds b: sigma_bnds orog: orography" self.height_bnds = netcdf_variable( + mocker, "height_bnds", "height bnds", np.float64, formula_terms=formula_terms, ) - self.lat = netcdf_variable("lat", "lat", np.float64) - self.lon = netcdf_variable("lon", "lon", np.float64) + self.lat = netcdf_variable(mocker, "lat", "lat", np.float64) + self.lon = netcdf_variable(mocker, "lon", "lon", np.float64) # Note that, only lat and lon are explicitly associated as coordinates. self.temp = netcdf_variable( - "temp", "height lat lon", np.float64, coordinates="lat lon" + mocker, "temp", "height lat lon", np.float64, coordinates="lat lon" ) self.variables = dict( @@ -134,8 +152,8 @@ def _setup(self, mocker): sigma_bnds=self.sigma_bnds, height_bnds=self.height_bnds, ) - ncattrs = mock.Mock(return_value=[]) - self.dataset = mock.Mock( + ncattrs = mocker.Mock(return_value=[]) + self.dataset = mocker.Mock( file_format="NetCDF4", variables=self.variables, ncattrs=ncattrs ) # Restrict the CFReader functionality to only performing translations. @@ -183,14 +201,23 @@ def test_create_formula_terms(self, mocker): class Test_build_cf_groups__formula_terms: @pytest.fixture(autouse=True) def _setup(self, mocker): - self.delta = netcdf_variable("delta", "height", np.float64, bounds="delta_bnds") - self.delta_bnds = netcdf_variable("delta_bnds", "height bnds", np.float64) - self.sigma = netcdf_variable("sigma", "height", np.float64, bounds="sigma_bnds") - self.sigma_bnds = netcdf_variable("sigma_bnds", "height bnds", np.float64) - self.orography = netcdf_variable("orography", "lat lon", np.float64) + self.delta = netcdf_variable( + mocker, "delta", "height", np.float64, bounds="delta_bnds" + ) + self.delta_bnds = netcdf_variable( + mocker, "delta_bnds", "height bnds", np.float64 + ) + self.sigma = netcdf_variable( + mocker, "sigma", "height", np.float64, bounds="sigma_bnds" + ) + self.sigma_bnds = netcdf_variable( + mocker, "sigma_bnds", "height bnds", np.float64 + ) + self.orography = netcdf_variable(mocker, "orography", "lat lon", np.float64) formula_terms = "a: delta b: sigma orog: orography" standard_name = "atmosphere_hybrid_height_coordinate" self.height = netcdf_variable( + mocker, "height", "height", np.float64, @@ -202,18 +229,19 @@ def _setup(self, mocker): # which will be ignored by the cf loader. formula_terms = "a: delta_bnds b: sigma_bnds orog: orography" self.height_bnds = netcdf_variable( + mocker, "height_bnds", "height bnds", np.float64, formula_terms=formula_terms, ) - self.lat = netcdf_variable("lat", "lat", np.float64) - self.lon = netcdf_variable("lon", "lon", np.float64) - self.x = netcdf_variable("x", "lat lon", np.float64) - self.y = netcdf_variable("y", "lat lon", np.float64) + self.lat = netcdf_variable(mocker, "lat", "lat", np.float64) + self.lon = netcdf_variable(mocker, "lon", "lon", np.float64) + self.x = netcdf_variable(mocker, "x", "lat lon", np.float64) + self.y = netcdf_variable(mocker, "y", "lat lon", np.float64) # Note that, only lat and lon are explicitly associated as coordinates. self.temp = netcdf_variable( - "temp", "height lat lon", np.float64, coordinates="x y" + mocker, "temp", "height lat lon", np.float64, coordinates="x y" ) self.variables = dict( @@ -230,8 +258,8 @@ def _setup(self, mocker): x=self.x, y=self.y, ) - ncattrs = mock.Mock(return_value=[]) - self.dataset = mock.Mock( + ncattrs = mocker.Mock(return_value=[]) + self.dataset = mocker.Mock( file_format="NetCDF4", variables=self.variables, ncattrs=ncattrs ) # Restrict the CFReader functionality to only performing translations @@ -242,6 +270,8 @@ def _setup(self, mocker): return_value=self.dataset, ) + self.wibble = netcdf_variable(mocker, "wibble", "lat wibble", np.float64) + def test_associate_formula_terms_with_data_variable(self, mocker): cf_group = CFReader("dummy").cf_group assert len(cf_group) == len(self.variables) @@ -317,7 +347,6 @@ def test_auxiliary_ignore(self): assert group[name].cf_data == getattr(self, name) def test_promoted_auxiliary_ignore(self): - self.wibble = netcdf_variable("wibble", "lat wibble", np.float64) self.variables["wibble"] = self.wibble self.orography.coordinates = "wibble" @@ -336,15 +365,15 @@ class Test_build_cf_groups__ugrid: @pytest.fixture(autouse=True) def _setup_class(self, mocker): # Replicating syntax from test_CFReader.Test_build_cf_groups__formula_terms. - self.mesh = netcdf_variable("mesh", "", int) - self.node_x = netcdf_variable("node_x", "node", float) - self.node_y = netcdf_variable("node_y", "node", float) - self.face_x = netcdf_variable("face_x", "face", float) - self.face_y = netcdf_variable("face_y", "face", float) - self.face_nodes = netcdf_variable("face_nodes", "face vertex", int) - self.levels = netcdf_variable("levels", "levels", int) + self.mesh = netcdf_variable(mocker, "mesh", "", int) + self.node_x = netcdf_variable(mocker, "node_x", "node", float) + self.node_y = netcdf_variable(mocker, "node_y", "node", float) + self.face_x = netcdf_variable(mocker, "face_x", "face", float) + self.face_y = netcdf_variable(mocker, "face_y", "face", float) + self.face_nodes = netcdf_variable(mocker, "face_nodes", "face vertex", int) + self.levels = netcdf_variable(mocker, "levels", "levels", int) self.data = netcdf_variable( - "data", "levels face", float, coordinates="face_x face_y" + mocker, "data", "levels face", float, coordinates="face_x face_y" ) # Add necessary attributes for mesh recognition. @@ -365,8 +394,8 @@ def _setup_class(self, mocker): levels=self.levels, data=self.data, ) - ncattrs = mock.Mock(return_value=[]) - self.dataset = mock.Mock( + ncattrs = mocker.Mock(return_value=[]) + self.dataset = mocker.Mock( file_format="NetCDF4", variables=self.variables, ncattrs=ncattrs ) @@ -413,11 +442,12 @@ def test_is_cf_ugrid_group(self): class Test_build_cf_groups__nczarr_scalar_grid_mapping: @pytest.fixture(autouse=True) def _setup_class(self, mocker): - self.lat = netcdf_variable("lat", "lat", np.float64) - self.lon = netcdf_variable("lon", "lon", np.float64) - self.crs = netcdf_variable("crs", "_scalar_", np.int32) + self.lat = netcdf_variable(mocker, "lat", "lat", np.float64) + self.lon = netcdf_variable(mocker, "lon", "lon", np.float64) + self.crs = netcdf_variable(mocker, "crs", "_scalar_", np.int32) self.crs.grid_mapping_name = "latitude_longitude" self.temp = netcdf_variable( + mocker, "temp", "lat lon", np.float64, @@ -435,8 +465,8 @@ def _setup_class(self, mocker): "crs": self.crs, "temp": self.temp, } - ncattrs = mock.Mock(return_value=[]) - self.dataset = mock.Mock( + ncattrs = mocker.Mock(return_value=[]) + self.dataset = mocker.Mock( file_format="NetCDF4", variables=self.variables, ncattrs=ncattrs ) mocker.patch("iris.fileformats.cf.CFReader._reset") @@ -454,3 +484,427 @@ def test_nczarr_scalar_grid_mapping_retains_type(self): def test_nczarr_scalar_grid_mapping_spans_data_var(self): temp_cf_group = self.cf_group["temp"].cf_group assert "crs" in temp_cf_group.grid_mappings + + +def test_destructor(tmp_path): + """Test the destructor when reading the dataset fails. + Related to issue #3312: previously, the `CFReader` would + always call `close()` on its `_dataset` attribute, even if it + didn't exist because opening the dataset had failed. + """ + fn = tmp_path / "tmp.nc" + with fn.open("wb+") as fh: + fh.write(b"\x89HDF\r\n\x1a\nBroken file with correct signature") + fh.flush() + + with io.StringIO() as buf: + with contextlib.redirect_stderr(buf): + try: + _ = cf.CFReader(str(fn)) + except OSError: + pass + try: + _ = iris.load_cubes(str(fn)) + except OSError: + pass + buf.seek(0) + assert buf.read() == "" + + +class Test_init_and_lifecycle: + @pytest.fixture(autouse=True) + def _setup(self, mocker): + self.variables = {"x": netcdf_variable(mocker, "x", "x", np.float64)} + self.dataset = mocker.Mock( + file_format="NetCDF4", + variables=self.variables, + ncattrs=mocker.Mock(return_value=[]), + filepath=mocker.Mock(return_value="in-memory.nc"), + ) + self.encoded_ds = mocker.patch( + "iris.fileformats.netcdf._bytecoding_datasets.EncodedDataset", + return_value=self.dataset, + ) + mocker.patch("iris.fileformats.cf.CFReader._translate") + mocker.patch("iris.fileformats.cf.CFReader._build_cf_groups") + mocker.patch("iris.fileformats.cf.CFReader._reset") + + def test_init_with_url_source_preserves_filename_string(self): + url = "https://example.com/some/file.nc" + reader = CFReader(url) + + self.encoded_ds.assert_called_once_with(url, mode="r") + assert reader.filename == url + assert repr(reader) == f"CFReader('{url}')" + + def test_init_uses_dataset_wrapper_when_string_decode_disabled(self, mocker): + mocker.patch( + "iris.fileformats.netcdf._bytecoding_datasets.DECODE_TO_STRINGS_ON_READ", + False, + ) + wrapper_ds = mocker.patch( + "iris.fileformats.cf._thread_safe_nc.DatasetWrapper", + return_value=self.dataset, + ) + + reader = CFReader("dummy.nc") + + assert reader.filename.name == "dummy.nc" + wrapper_ds.assert_called_once() + self.encoded_ds.assert_not_called() + + def test_init_with_open_dataset_does_not_close_on_context_exit(self): + with CFReader(self.dataset) as reader: + assert reader is not None + assert reader.filename == "in-memory.nc" + + self.dataset.close.assert_not_called() + + def test_init_warns_for_netcdf3_when_requested(self): + self.dataset.file_format = "NETCDF3_CLASSIC" + + with pytest.warns(iris.warnings.IrisLoadWarning, match="Optimise CF-netCDF"): + CFReader("dummy.nc", warn=True) + + def test_init_with_no_meshes_trims_ugrid_variable_types(self, mocker): + self.dataset.variables = {"a": object(), "b": mocker.Mock(mesh=None)} + + reader = CFReader("dummy.nc") + + assert reader._with_ugrid is True + + mesh_free = mocker.Mock( + file_format="NetCDF4", + variables={}, + ncattrs=mocker.Mock(return_value=[]), + filepath=mocker.Mock(return_value="in-memory.nc"), + ) + self.encoded_ds.return_value = mesh_free + reader = CFReader("dummy.nc") + + assert reader._with_ugrid is False + assert CFUGridMeshVariable not in reader._variable_types + + +class Test_translate__grid_mapping_parse_errors: + @pytest.fixture(autouse=True) + def _setup(self, mocker): + self.lat = netcdf_variable(mocker, "lat", "lat", np.float64) + self.lon = netcdf_variable(mocker, "lon", "lon", np.float64) + self.temp = netcdf_variable( + mocker, + "temp", + "lat lon", + np.float64, + coordinates="lat lon", + grid_mapping="bad syntax", + ) + self.temp.name = "temp" + variables = {"lat": self.lat, "lon": self.lon, "temp": self.temp} + self.dataset = mocker.Mock( + file_format="NetCDF4", + variables=variables, + ncattrs=mocker.Mock(return_value=[]), + ) + mocker.patch( + "iris.fileformats.netcdf._bytecoding_datasets.EncodedDataset", + return_value=self.dataset, + ) + mocker.patch( + "iris.fileformats.cf.hh._parse_extended_grid_mapping", + side_effect=iris.exceptions.CFParseError("failed to parse"), + ) + + def test_parse_failure_warns_and_reader_continues(self): + with pytest.warns( + iris.warnings.IrisCfWarning, + match=r"Error parsing `grid_mapping` attribute for temp: failed to parse", + ): + cf_group = CFReader("dummy.nc").cf_group + + assert "temp" in cf_group.data_variables + + +class Test_translate__formula_terms_derived_bounds: + @pytest.fixture(autouse=True) + def _setup(self, mocker): + self.term = netcdf_variable(mocker, "term", "z", np.float64) + self.root = netcdf_variable( + mocker, + "z", + "z", + np.float64, + formula_terms="a: term", + bounds="z_bnds", + standard_name="atmosphere_hybrid_height_coordinate", + ) + self.root_bnds = netcdf_variable( + mocker, + "z_bnds", + "z bnds", + np.float64, + formula_terms="a: term", + ) + self.data = netcdf_variable(mocker, "temp", "z", np.float64) + self.variables = { + "z": self.root, + "z_bnds": self.root_bnds, + "term": self.term, + "temp": self.data, + } + + @staticmethod + def _make_dataset(mocker, variables): + return mocker.Mock( + file_format="NetCDF4", + variables=variables, + ncattrs=mocker.Mock(return_value=[]), + ) + + def _patch_encoded_dataset(self, mocker): + dataset = self._make_dataset(mocker, self.variables) + mocker.patch( + "iris.fileformats.netcdf._bytecoding_datasets.EncodedDataset", + return_value=dataset, + ) + + @pytest.fixture(params=[True, False], ids=["FUTURE", "not_FUTURE"]) + def future_context(self, request): + if request.param: + result = iris.FUTURE.context(derived_bounds=True) + else: + result = contextlib.nullcontext() + return result + + def test_derived_bounds_term(self, mocker, future_context): + self._patch_encoded_dataset(mocker) + + with future_context: + cf_group = CFReader("dummy.nc").cf_group + + assert "term" in cf_group.formula_terms + assert isinstance(cf_group["term"], CFAuxiliaryCoordinateVariable) + + def test_derived_bounds_skips_when_term_missing(self, mocker, future_context): + self.root_bnds.formula_terms = "a: missing_term" + self._patch_encoded_dataset(mocker) + + with future_context: + cf_group = CFReader("dummy.nc").cf_group + + assert "term" in cf_group.formula_terms + assert isinstance(cf_group["term"], CFAuxiliaryCoordinateVariable) + + def test_promotes_non_formula_root_bounds_to_data(self, mocker): + self.root_bnds = netcdf_variable(mocker, "z_bnds", "_scalar_", np.float64) + # With valid formula terms, the variable would instead be recorded + # correctly as a bounds variable. + del self.root_bnds.formula_terms + self.variables["z_bnds"] = self.root_bnds + self._patch_encoded_dataset(mocker) + + with iris.FUTURE.context(derived_bounds=True): + cf_group = CFReader("dummy.nc").cf_group + + assert "z_bnds" in cf_group.promoted + assert cf_group["z"].bounds is None + + def test_reclassifies_formula_term_bounds_variable(self, mocker, future_context): + # If not referenced by any formula terms, the bounds variable is + # promoted to a data variable. + # Referenced + delta = netcdf_variable( + mocker, "delta", "height", np.float64, bounds="delta_bnds" + ) + # Referenced + sigma = netcdf_variable( + mocker, "sigma", "height", np.float64, bounds="sigma_bnds" + ) + formula_terms = "a: delta b: sigma" + height = netcdf_variable( + mocker, + "height", + "height", + np.float64, + formula_terms=formula_terms, + bounds="height_bnds", + standard_name="atmosphere_hybrid_height_coordinate", + ) + # Referenced + formula_terms_bnds = "a: delta_bnds b: sigma_bnds" + height_bnds = netcdf_variable( + mocker, + "height_bnds", + "height bnds", + np.float64, + formula_terms=formula_terms_bnds, + ) + delta_bnds = netcdf_variable(mocker, "delta_bnds", "height bnds", np.float64) + sigma_bnds = netcdf_variable(mocker, "sigma_bnds", "height bnds", np.float64) + data = netcdf_variable(mocker, "temp", "height", np.float64) + variables = { + "delta": delta, + "sigma": sigma, + "height": height, + "height_bnds": height_bnds, + "delta_bnds": delta_bnds, + "sigma_bnds": sigma_bnds, + "temp": data, + } + self.variables = variables + self._patch_encoded_dataset(mocker) + + with future_context: + cf_group = CFReader("dummy.nc").cf_group + + assert isinstance(cf_group["delta_bnds"], CFBoundaryVariable) + assert isinstance(cf_group["sigma_bnds"], CFBoundaryVariable) + assert cf_group["delta"].bounds == "delta_bnds" + assert cf_group["sigma"].bounds == "sigma_bnds" + + def test_formula_term_already_in_group_uses_existing_variable( + self, mocker, future_context + ): + self.root = netcdf_variable( + mocker, + "z", + "z", + np.float64, + formula_terms="a: pressure", + standard_name="atmosphere_hybrid_height_coordinate", + ) + self.pressure = netcdf_variable(mocker, "pressure", "pressure", np.float64) + self.variables = {"z": self.root, "pressure": self.pressure, "temp": self.data} + self._patch_encoded_dataset(mocker) + + with future_context: + cf_group = CFReader("dummy.nc").cf_group + + # When existing elsewhere, a variable referenced by a formula term is + # NOT created as an aux coord as it usually would. Instead the existing + # variable is re-used. + assert cf_group.formula_terms["pressure"] is cf_group.coordinates["pressure"] + assert "pressure" not in cf_group.auxiliary_coordinates + + @pytest.mark.parametrize( + "standard_name", [False, True], ids=["no_standard_name", "with_standard_name"] + ) + def test_derived_bounds_promotes_reference_terms( + self, mocker, future_context, standard_name + ): + self.root = netcdf_variable( + mocker, + "z", + "z", + np.float64, + formula_terms="a: pressure", + standard_name="custom_reference", + ) + if not standard_name: + if isinstance(future_context, contextlib.nullcontext): + pytest.skip("Test only applicable when FUTURE context is enabled.") + else: + del self.root.standard_name + self.pressure = netcdf_variable(mocker, "pressure", "z", np.float64) + self.variables = {"z": self.root, "pressure": self.pressure, "temp": self.data} + self._patch_encoded_dataset(mocker) + # reference_terms supports hybrid heights. Variables that are both + # named in formula_terms and referenced in reference_terms - "a" in + # this case - are always promoted. + mocker.patch.dict( + "iris.fileformats.cf.reference_terms", + # Real world example: {"atmosphere_sigma_coordinate": ["ps"]} + {"custom_reference": "a"}, + clear=False, + ) + + with future_context: + cf_group = CFReader("dummy.nc").cf_group + + if standard_name: + assert "pressure" in cf_group.promoted + else: + # Promotion step is skipped if standard_name is absent. + assert "pressure" not in cf_group.promoted + + +class Test_translate__global_attributes_missing: + @pytest.fixture(autouse=True) + def _setup(self, mocker): + self.var = netcdf_variable(mocker, "x", "x", np.float64) + self.dataset = mocker.Mock( + file_format="NetCDF4", + variables={"x": self.var}, + ncattrs=mocker.Mock(return_value=["history"]), + getncattr=mocker.Mock(side_effect=AttributeError), + ) + mocker.patch( + "iris.fileformats.netcdf._bytecoding_datasets.EncodedDataset", + return_value=self.dataset, + ) + mocker.patch("iris.fileformats.cf.CFReader._build_cf_groups") + mocker.patch("iris.fileformats.cf.CFReader._reset") + + def test_translate_global_attr_missing_falls_back_to_default(self): + cf_group = CFReader("dummy.nc").cf_group + + assert cf_group.global_attributes["history"] == "" + + +class Test_build_cf_groups__private_edge_cases: + # Achieving full coverage requires a limited amount of internal state + # manipulation (to be avoided where possible as it hurts future refactoring). + + @pytest.fixture(autouse=True) + def _setup(self): + self.reader = object.__new__(CFReader) + self.reader._own_file = False + self.reader._dataset = None + self.reader._variable_types = () + self.reader._coord_system_mappings = {} + self.reader.cf_group = CFGroup() + + def test_derived_bounds_relinks_spanning_bounds(self, mocker): + data = netcdf_variable( + mocker, + "temp", + "z", + np.float64, + bounds="temp_bnds", + standard_name="atmosphere_hybrid_height_coordinate", + ) + data.__len__ = mocker.Mock(return_value=1) + data_var = CFDataVariable( + "temp", + data, + ) + bnds_data = netcdf_variable(mocker, "temp_bnds", "z bnds", np.float64) + bnds_data.__len__ = mocker.Mock(return_value=1) + bnds_var = CFBoundaryVariable( + "temp_bnds", + bnds_data, + ) + self.reader.cf_group["temp"] = data_var + self.reader.cf_group["temp_bnds"] = bnds_var + + with iris.FUTURE.context(derived_bounds=True): + self.reader._build_cf_groups({}) + + assert "temp_bnds" in self.reader.cf_group["temp"].cf_group.bounds + + def test_derived_bounds_boundary_guard_continue_branch(self, mocker): + # TODO: maybe the code itself is wrong and should be using isinstance? + root = CFCoordinateVariable("z", netcdf_variable(mocker, "z", "z", np.float64)) + term = CFAuxiliaryCoordinateVariable( + "term", netcdf_variable(mocker, "term", "z", np.float64) + ) + term.add_formula_term("z", "a") + self.reader.cf_group["z"] = root + self.reader.cf_group["term"] = term + + # Force the exact identity check branch in CFReader._build_cf_groups. + mocker.patch("iris.fileformats.cf.CFBoundaryVariable", term) + with iris.FUTURE.context(derived_bounds=True): + self.reader._build_cf_groups({}) + + assert "term" in self.reader.cf_group.formula_terms diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFUGridAuxiliaryCoordinateVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFUGridAuxiliaryCoordinateVariable.py index 630359f681..5efbe67bd2 100644 --- a/lib/iris/tests/unit/fileformats/cf/test_CFUGridAuxiliaryCoordinateVariable.py +++ b/lib/iris/tests/unit/fileformats/cf/test_CFUGridAuxiliaryCoordinateVariable.py @@ -4,103 +4,30 @@ # See LICENSE in the root of the repository for full licensing details. """Unit tests for :class:`iris.fileformats.cf.CFUGridAuxiliaryCoordinateVariable`.""" -import re -import warnings - -import numpy as np -import pytest - from iris.fileformats.cf import CFUGridAuxiliaryCoordinateVariable -from iris.tests.unit.fileformats.cf.test_CFReader import netcdf_variable -import iris.warnings - - -def named_variable(name): - # Don't need to worry about dimensions or dtype for these tests. - return netcdf_variable(name, "", int) +from .identify_catalogue import IdentifyByAttributeListCatalog -class TestIdentify: - @pytest.fixture(autouse=True) - def _setup(self): - self.cf_identities = [ - "node_coordinates", - "edge_coordinates", - "face_coordinates", - "volume_coordinates", - ] - - def test_cf_identities(self): - subject_name = "ref_subject" - ref_subject = named_variable(subject_name) - vars_common = { - subject_name: ref_subject, - "ref_not_subject": named_variable("ref_not_subject"), - } - # ONLY expecting ref_subject, excluding ref_not_subject. - expected = { - subject_name: CFUGridAuxiliaryCoordinateVariable(subject_name, ref_subject) - } - - for identity in self.cf_identities: - ref_source = named_variable("ref_source") - setattr(ref_source, identity, subject_name) - vars_all = dict({"ref_source": ref_source}, **vars_common) - result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all) - assert expected == result - - def test_duplicate_refs(self): - subject_name = "ref_subject" - ref_subject = named_variable(subject_name) - ref_source_vars = { - name: named_variable(name) for name in ("ref_source_1", "ref_source_2") - } - for var in ref_source_vars.values(): - setattr(var, self.cf_identities[0], subject_name) - vars_all = dict( - { - subject_name: ref_subject, - "ref_not_subject": named_variable("ref_not_subject"), - }, - **ref_source_vars, - ) - - # ONLY expecting ref_subject, excluding ref_not_subject. - expected = { - subject_name: CFUGridAuxiliaryCoordinateVariable(subject_name, ref_subject) - } - result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all) - assert expected == result - - def test_two_coords(self): - subject_names = ("ref_subject_1", "ref_subject_2") - ref_subject_vars = {name: named_variable(name) for name in subject_names} - ref_source_vars = { - name: named_variable(name) for name in ("ref_source_1", "ref_source_2") - } - for ix, var in enumerate(ref_source_vars.values()): - setattr(var, self.cf_identities[ix], subject_names[ix]) - vars_all = dict( - {"ref_not_subject": named_variable("ref_not_subject")}, - **ref_subject_vars, - **ref_source_vars, - ) +class TestIdentify(IdentifyByAttributeListCatalog): + __test__ = True - # Not expecting ref_not_subject. - expected = { - name: CFUGridAuxiliaryCoordinateVariable(name, var) - for name, var in ref_subject_vars.items() - } - result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all) - assert expected == result + CF_CLASS = CFUGridAuxiliaryCoordinateVariable + CF_IDENTITIES = [ + "node_coordinates", + "edge_coordinates", + "face_coordinates", + "volume_coordinates", + ] + MISSING_WARN_REGEX = r"Missing CF-netCDF auxiliary coordinate variable {subject}.*" - def test_two_part_ref(self): + def test_two_part_ref(self, named_variable): + # UGRID auxiliary coordinate attributes can contain multiple refs. subject_names = ("ref_subject_1", "ref_subject_2") ref_subject_vars = {name: named_variable(name) for name in subject_names} ref_source = named_variable("ref_source") - setattr(ref_source, self.cf_identities[0], " ".join(subject_names)) + self._set_ref(ref_source, self.CF_IDENTITIES[0], " ".join(subject_names)) vars_all = { "ref_not_subject": named_variable("ref_not_subject"), "ref_source": ref_source, @@ -108,112 +35,8 @@ def test_two_part_ref(self): } expected = { - name: CFUGridAuxiliaryCoordinateVariable(name, var) + name: self._expected_var(name, var) for name, var in ref_subject_vars.items() } - result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all) - assert expected == result - - def test_string_type_ignored(self): - subject_name = "ref_subject" - ref_source = named_variable("ref_source") - setattr(ref_source, self.cf_identities[0], subject_name) - vars_all = { - subject_name: netcdf_variable(subject_name, "", np.bytes_), - "ref_not_subject": named_variable("ref_not_subject"), - "ref_source": ref_source, - } - - result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all) - assert {} == result - - def test_ignore(self): - subject_names = ("ref_subject_1", "ref_subject_2") - ref_subject_vars = {name: named_variable(name) for name in subject_names} - - ref_source_vars = { - name: named_variable(name) for name in ("ref_source_1", "ref_source_2") - } - for ix, var in enumerate(ref_source_vars.values()): - setattr(var, self.cf_identities[0], subject_names[ix]) - vars_all = dict( - {"ref_not_subject": named_variable("ref_not_subject")}, - **ref_subject_vars, - **ref_source_vars, - ) - - # ONLY expect the subject variable that hasn't been ignored. - expected_name = subject_names[0] - expected = { - expected_name: CFUGridAuxiliaryCoordinateVariable( - expected_name, ref_subject_vars[expected_name] - ) - } - result = CFUGridAuxiliaryCoordinateVariable.identify( - vars_all, ignore=subject_names[1] - ) - assert expected == result - - def test_target(self): - subject_names = ("ref_subject_1", "ref_subject_2") - ref_subject_vars = {name: named_variable(name) for name in subject_names} - - source_names = ("ref_source_1", "ref_source_2") - ref_source_vars = {name: named_variable(name) for name in source_names} - for ix, var in enumerate(ref_source_vars.values()): - setattr(var, self.cf_identities[0], subject_names[ix]) - vars_all = dict( - {"ref_not_subject": named_variable("ref_not_subject")}, - **ref_subject_vars, - **ref_source_vars, - ) - - # ONLY expect the variable referenced by the named ref_source_var. - expected_name = subject_names[0] - expected = { - expected_name: CFUGridAuxiliaryCoordinateVariable( - expected_name, ref_subject_vars[expected_name] - ) - } - result = CFUGridAuxiliaryCoordinateVariable.identify( - vars_all, target=source_names[0] - ) + result = self.CF_CLASS.identify(vars_all) assert expected == result - - def test_warn(self): - subject_name = "ref_subject" - ref_source = named_variable("ref_source") - setattr(ref_source, self.cf_identities[0], subject_name) - vars_all = { - "ref_not_subject": named_variable("ref_not_subject"), - "ref_source": ref_source, - } - - def operation(warn: bool): - warnings.warn( - "emit at least 1 warning", - category=iris.warnings.IrisUserWarning, - ) - result = CFUGridAuxiliaryCoordinateVariable.identify(vars_all, warn=warn) - assert {} == result - - # Missing warning. - warn_regex = ( - rf"Missing CF-netCDF auxiliary coordinate variable {subject_name}.*" - ) - with pytest.warns(iris.warnings.IrisCfMissingVarWarning, match=warn_regex): - operation(warn=True) - with pytest.warns() as record: - operation(warn=False) - warn_list = [str(w.message) for w in record] - assert list(filter(re.compile(warn_regex).match, warn_list)) == [] - - # String variable warning. - warn_regex = r".*is a CF-netCDF label variable.*" - vars_all[subject_name] = netcdf_variable(subject_name, "", np.bytes_) - with pytest.warns(iris.warnings.IrisCfLabelVarWarning, match=warn_regex): - operation(warn=True) - with pytest.warns() as record: - operation(warn=False) - warn_list = [str(w.message) for w in record] - assert list(filter(re.compile(warn_regex).match, warn_list)) == [] diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFUGridConnectivityVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFUGridConnectivityVariable.py index d319f3660f..d4976ecfd6 100644 --- a/lib/iris/tests/unit/fileformats/cf/test_CFUGridConnectivityVariable.py +++ b/lib/iris/tests/unit/fileformats/cf/test_CFUGridConnectivityVariable.py @@ -4,198 +4,15 @@ # See LICENSE in the root of the repository for full licensing details. """Unit tests for :class:`iris.fileformats.cf.CFUGridConnectivityVariable`.""" -import re -import warnings - -import numpy as np -import pytest - from iris.fileformats.cf import CFUGridConnectivityVariable from iris.mesh import Connectivity -from iris.tests.unit.fileformats.cf.test_CFReader import netcdf_variable -import iris.warnings - - -def named_variable(name): - # Don't need to worry about dimensions or dtype for these tests. - return netcdf_variable(name, "", int) - - -class TestIdentify: - def test_cf_identities(self): - subject_name = "ref_subject" - ref_subject = named_variable(subject_name) - vars_common = { - subject_name: ref_subject, - "ref_not_subject": named_variable("ref_not_subject"), - } - # ONLY expecting ref_subject, excluding ref_not_subject. - expected = { - subject_name: CFUGridConnectivityVariable(subject_name, ref_subject) - } - - for identity in Connectivity.UGRID_CF_ROLES: - ref_source = named_variable("ref_source") - setattr(ref_source, identity, subject_name) - vars_all = dict({"ref_source": ref_source}, **vars_common) - result = CFUGridConnectivityVariable.identify(vars_all) - assert expected == result - - def test_duplicate_refs(self): - subject_name = "ref_subject" - ref_subject = named_variable(subject_name) - ref_source_vars = { - name: named_variable(name) for name in ("ref_source_1", "ref_source_2") - } - for var in ref_source_vars.values(): - setattr(var, Connectivity.UGRID_CF_ROLES[0], subject_name) - vars_all = dict( - { - subject_name: ref_subject, - "ref_not_subject": named_variable("ref_not_subject"), - }, - **ref_source_vars, - ) - - # ONLY expecting ref_subject, excluding ref_not_subject. - expected = { - subject_name: CFUGridConnectivityVariable(subject_name, ref_subject) - } - result = CFUGridConnectivityVariable.identify(vars_all) - assert expected == result - - def test_two_cf_roles(self): - subject_names = ("ref_subject_1", "ref_subject_2") - ref_subject_vars = {name: named_variable(name) for name in subject_names} - - ref_source_vars = { - name: named_variable(name) for name in ("ref_source_1", "ref_source_2") - } - for ix, var in enumerate(ref_source_vars.values()): - setattr(var, Connectivity.UGRID_CF_ROLES[ix], subject_names[ix]) - vars_all = dict( - {"ref_not_subject": named_variable("ref_not_subject")}, - **ref_subject_vars, - **ref_source_vars, - ) - - # Not expecting ref_not_subject. - expected = { - name: CFUGridConnectivityVariable(name, var) - for name, var in ref_subject_vars.items() - } - result = CFUGridConnectivityVariable.identify(vars_all) - assert expected == result - - def test_two_part_ref_ignored(self): - # Not expected to handle more than one variable for a connectivity - # cf role - invalid UGRID. - subject_name = "ref_subject" - ref_source = named_variable("ref_source") - setattr(ref_source, Connectivity.UGRID_CF_ROLES[0], subject_name + " foo") - vars_all = { - subject_name: named_variable(subject_name), - "ref_not_subject": named_variable("ref_not_subject"), - "ref_source": ref_source, - } - - result = CFUGridConnectivityVariable.identify(vars_all) - assert {} == result - - def test_string_type_ignored(self): - subject_name = "ref_subject" - ref_source = named_variable("ref_source") - setattr(ref_source, Connectivity.UGRID_CF_ROLES[0], subject_name) - vars_all = { - subject_name: netcdf_variable(subject_name, "", np.bytes_), - "ref_not_subject": named_variable("ref_not_subject"), - "ref_source": ref_source, - } - - result = CFUGridConnectivityVariable.identify(vars_all) - assert {} == result - - def test_ignore(self): - subject_names = ("ref_subject_1", "ref_subject_2") - ref_subject_vars = {name: named_variable(name) for name in subject_names} - - ref_source_vars = { - name: named_variable(name) for name in ("ref_source_1", "ref_source_2") - } - for ix, var in enumerate(ref_source_vars.values()): - setattr(var, Connectivity.UGRID_CF_ROLES[0], subject_names[ix]) - vars_all = dict( - {"ref_not_subject": named_variable("ref_not_subject")}, - **ref_subject_vars, - **ref_source_vars, - ) - - # ONLY expect the subject variable that hasn't been ignored. - expected_name = subject_names[0] - expected = { - expected_name: CFUGridConnectivityVariable( - expected_name, ref_subject_vars[expected_name] - ) - } - result = CFUGridConnectivityVariable.identify(vars_all, ignore=subject_names[1]) - assert expected == result - - def test_target(self): - subject_names = ("ref_subject_1", "ref_subject_2") - ref_subject_vars = {name: named_variable(name) for name in subject_names} - - source_names = ("ref_source_1", "ref_source_2") - ref_source_vars = {name: named_variable(name) for name in source_names} - for ix, var in enumerate(ref_source_vars.values()): - setattr(var, Connectivity.UGRID_CF_ROLES[0], subject_names[ix]) - vars_all = dict( - {"ref_not_subject": named_variable("ref_not_subject")}, - **ref_subject_vars, - **ref_source_vars, - ) - - # ONLY expect the variable referenced by the named ref_source_var. - expected_name = subject_names[0] - expected = { - expected_name: CFUGridConnectivityVariable( - expected_name, ref_subject_vars[expected_name] - ) - } - result = CFUGridConnectivityVariable.identify(vars_all, target=source_names[0]) - assert expected == result - def test_warn(self): - subject_name = "ref_subject" - ref_source = named_variable("ref_source") - setattr(ref_source, Connectivity.UGRID_CF_ROLES[0], subject_name) - vars_all = { - "ref_not_subject": named_variable("ref_not_subject"), - "ref_source": ref_source, - } +from .identify_catalogue import IdentifyByAttributeListCatalog - def operation(warn: bool): - warnings.warn( - "emit at least 1 warning", - category=iris.warnings.IrisUserWarning, - ) - result = CFUGridConnectivityVariable.identify(vars_all, warn=warn) - assert {} == result - # Missing warning. - warn_regex = rf"Missing CF-UGRID connectivity variable {subject_name}.*" - with pytest.warns(iris.warnings.IrisCfMissingVarWarning, match=warn_regex): - operation(warn=True) - with pytest.warns() as record: - operation(warn=False) - warn_list = [str(w.message) for w in record] - assert list(filter(re.compile(warn_regex).match, warn_list)) == [] +class TestIdentify(IdentifyByAttributeListCatalog): + __test__ = True - # String variable warning. - warn_regex = r".*is a CF-netCDF label variable.*" - vars_all[subject_name] = netcdf_variable(subject_name, "", np.bytes_) - with pytest.warns(iris.warnings.IrisCfLabelVarWarning, match=warn_regex): - operation(warn=True) - with pytest.warns() as record: - operation(warn=False) - warn_list = [str(w.message) for w in record] - assert list(filter(re.compile(warn_regex).match, warn_list)) == [] + CF_CLASS = CFUGridConnectivityVariable + CF_IDENTITIES = Connectivity.UGRID_CF_ROLES + MISSING_WARN_REGEX = r"Missing CF-UGRID connectivity variable {subject}.*" diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFUGridMeshVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFUGridMeshVariable.py index e6e2db58af..4093c9b913 100644 --- a/lib/iris/tests/unit/fileformats/cf/test_CFUGridMeshVariable.py +++ b/lib/iris/tests/unit/fileformats/cf/test_CFUGridMeshVariable.py @@ -4,28 +4,19 @@ # See LICENSE in the root of the repository for full licensing details. """Unit tests for :class:`iris.fileformats.cf.CFUGridMeshVariable`.""" -import re import warnings import numpy as np import pytest from iris.fileformats.cf import CFUGridMeshVariable -from iris.tests.unit.fileformats.cf.test_CFReader import netcdf_variable import iris.warnings - -def named_variable(name): - # Don't need to worry about dimensions or dtype for these tests. - return netcdf_variable(name, "", int) +CF_IDENTITY = "mesh" class TestIdentify: - @pytest.fixture(autouse=True) - def _setup(self): - self.cf_identity = "mesh" - - def test_cf_role(self): + def test_cf_role(self, named_variable): # Test that mesh variables can be identified by having `cf_role="mesh_topology"`. match_name = "match" match = named_variable(match_name) @@ -42,13 +33,13 @@ def test_cf_role(self): result = CFUGridMeshVariable.identify(vars_all) assert expected == result - def test_cf_identity(self): + def test_cf_identity(self, named_variable): # Test that mesh variables can be identified by being another variable's # `mesh` attribute. subject_name = "ref_subject" ref_subject = named_variable(subject_name) ref_source = named_variable("ref_source") - setattr(ref_source, self.cf_identity, subject_name) + setattr(ref_source, CF_IDENTITY, subject_name) vars_all = { subject_name: ref_subject, "ref_not_subject": named_variable("ref_not_subject"), @@ -60,7 +51,7 @@ def test_cf_identity(self): result = CFUGridMeshVariable.identify(vars_all) assert expected == result - def test_cf_role_and_identity(self): + def test_cf_role_and_identity(self, named_variable): # Test that identification can successfully handle a combination of # mesh variables having `cf_role="mesh_topology"` AND being referenced as # another variable's `mesh` attribute. @@ -68,12 +59,12 @@ def test_cf_role_and_identity(self): role_match = named_variable(role_match_name) setattr(role_match, "cf_role", "mesh_topology") ref_source_1 = named_variable("ref_source_1") - setattr(ref_source_1, self.cf_identity, role_match_name) + setattr(ref_source_1, CF_IDENTITY, role_match_name) subject_name = "ref_subject" ref_subject = named_variable(subject_name) ref_source_2 = named_variable("ref_source_2") - setattr(ref_source_2, self.cf_identity, subject_name) + setattr(ref_source_2, CF_IDENTITY, subject_name) vars_all = { role_match_name: role_match, @@ -91,14 +82,14 @@ def test_cf_role_and_identity(self): result = CFUGridMeshVariable.identify(vars_all) assert expected == result - def test_duplicate_refs(self): + def test_duplicate_refs(self, named_variable): subject_name = "ref_subject" ref_subject = named_variable(subject_name) ref_source_vars = { name: named_variable(name) for name in ("ref_source_1", "ref_source_2") } for var in ref_source_vars.values(): - setattr(var, self.cf_identity, subject_name) + setattr(var, CF_IDENTITY, subject_name) vars_all = dict( { subject_name: ref_subject, @@ -112,7 +103,7 @@ def test_duplicate_refs(self): result = CFUGridMeshVariable.identify(vars_all) assert expected == result - def test_two_refs(self): + def test_two_refs(self, named_variable): subject_names = ("ref_subject_1", "ref_subject_2") ref_subject_vars = {name: named_variable(name) for name in subject_names} @@ -120,7 +111,7 @@ def test_two_refs(self): name: named_variable(name) for name in ("ref_source_1", "ref_source_2") } for ix, var in enumerate(ref_source_vars.values()): - setattr(var, self.cf_identity, subject_names[ix]) + setattr(var, CF_IDENTITY, subject_names[ix]) vars_all = dict( {"ref_not_subject": named_variable("ref_not_subject")}, **ref_subject_vars, @@ -135,12 +126,12 @@ def test_two_refs(self): result = CFUGridMeshVariable.identify(vars_all) assert expected == result - def test_two_part_ref_ignored(self): + def test_two_part_ref_ignored(self, named_variable): # Not expected to handle more than one variable for a mesh # cf role - invalid UGRID. subject_name = "ref_subject" ref_source = named_variable("ref_source") - setattr(ref_source, self.cf_identity, subject_name + " foo") + setattr(ref_source, CF_IDENTITY, subject_name + " foo") vars_all = { subject_name: named_variable(subject_name), "ref_not_subject": named_variable("ref_not_subject"), @@ -150,12 +141,12 @@ def test_two_part_ref_ignored(self): result = CFUGridMeshVariable.identify(vars_all) assert {} == result - def test_string_type_ignored(self): + def test_string_type_ignored(self, named_variable): subject_name = "ref_subject" ref_source = named_variable("ref_source") - setattr(ref_source, self.cf_identity, subject_name) + setattr(ref_source, CF_IDENTITY, subject_name) vars_all = { - subject_name: netcdf_variable(subject_name, "", np.bytes_), + subject_name: named_variable(subject_name, dtype=np.bytes_), "ref_not_subject": named_variable("ref_not_subject"), "ref_source": ref_source, } @@ -163,7 +154,7 @@ def test_string_type_ignored(self): result = CFUGridMeshVariable.identify(vars_all) assert {} == result - def test_ignore(self): + def test_ignore(self, named_variable): subject_names = ("ref_subject_1", "ref_subject_2") ref_subject_vars = {name: named_variable(name) for name in subject_names} @@ -171,7 +162,7 @@ def test_ignore(self): name: named_variable(name) for name in ("ref_source_1", "ref_source_2") } for ix, var in enumerate(ref_source_vars.values()): - setattr(var, self.cf_identity, subject_names[ix]) + setattr(var, CF_IDENTITY, subject_names[ix]) vars_all = dict( {"ref_not_subject": named_variable("ref_not_subject")}, **ref_subject_vars, @@ -188,14 +179,14 @@ def test_ignore(self): result = CFUGridMeshVariable.identify(vars_all, ignore=subject_names[1]) assert expected == result - def test_target(self): + def test_target(self, named_variable): subject_names = ("ref_subject_1", "ref_subject_2") ref_subject_vars = {name: named_variable(name) for name in subject_names} source_names = ("ref_source_1", "ref_source_2") ref_source_vars = {name: named_variable(name) for name in source_names} for ix, var in enumerate(ref_source_vars.values()): - setattr(var, self.cf_identity, subject_names[ix]) + setattr(var, CF_IDENTITY, subject_names[ix]) vars_all = dict( {"ref_not_subject": named_variable("ref_not_subject")}, **ref_subject_vars, @@ -212,10 +203,24 @@ def test_target(self): result = CFUGridMeshVariable.identify(vars_all, target=source_names[0]) assert expected == result - def test_warn(self): + def test_target_unknown_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Cannot identify unknown target CF-netCDF variable 'unknown'" + with pytest.raises(ValueError, match=message): + CFUGridMeshVariable.identify(vars_all, target="unknown") + + def test_target_wrong_type_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Expect a target CF-netCDF variable name" + with pytest.raises(TypeError, match=message): + CFUGridMeshVariable.identify(vars_all, target=object()) + + def test_warn(self, named_variable, assert_warning_gated): subject_name = "ref_subject" ref_source = named_variable("ref_source") - setattr(ref_source, self.cf_identity, subject_name) + setattr(ref_source, CF_IDENTITY, subject_name) vars_all = { "ref_not_subject": named_variable("ref_not_subject"), "ref_source": ref_source, @@ -231,19 +236,11 @@ def operation(warn: bool): # Missing warning. warn_regex = rf"Missing CF-UGRID mesh variable {subject_name}.*" - with pytest.warns(iris.warnings.IrisCfMissingVarWarning, match=warn_regex): - operation(warn=True) - with pytest.warns() as record: - operation(warn=False) - warn_list = [str(w.message) for w in record] - assert list(filter(re.compile(warn_regex).match, warn_list)) == [] + assert_warning_gated( + operation, iris.warnings.IrisCfMissingVarWarning, warn_regex + ) # String variable warning. warn_regex = r".*is a CF-netCDF label variable.*" - vars_all[subject_name] = netcdf_variable(subject_name, "", np.bytes_) - with pytest.warns(iris.warnings.IrisCfLabelVarWarning, match=warn_regex): - operation(warn=True) - with pytest.warns() as record: - operation(warn=False) - warn_list = [str(w.message) for w in record] - assert list(filter(re.compile(warn_regex).match, warn_list)) == [] + vars_all[subject_name] = named_variable(subject_name, dtype=np.bytes_) + assert_warning_gated(operation, iris.warnings.IrisCfLabelVarWarning, warn_regex) diff --git a/lib/iris/tests/unit/fileformats/cf/test_CFVariable.py b/lib/iris/tests/unit/fileformats/cf/test_CFVariable.py new file mode 100644 index 0000000000..890d2f4080 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test_CFVariable.py @@ -0,0 +1,218 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf.CFVariable`.""" + +import pytest + +from iris.fileformats import cf as cf + + +class CFVariableSub(cf.CFVariable): + """A subclass of CFVariable for testing purposes.""" + + def identify(self, variables, ignore=None, target=None, warn=True): + return super().identify(variables, ignore=ignore, target=target, warn=warn) + + +def make_nc_var(mocker): + nc_var = mocker.MagicMock() + nc_var.ncattrs.return_value = ["coordinates", "standard_name", "_FillValue"] + nc_var.getncattr.side_effect = { + "coordinates": "x y", + "standard_name": "air_temperature", + "_FillValue": -999, + }.__getitem__ + nc_var.coordinates = "x y" + nc_var.standard_name = "air_temperature" + nc_var.dimensions = ("time", "lat") + nc_var.__len__.return_value = 4 + nc_var.__getitem__.return_value = "payload" + nc_var.group.return_value.filepath.return_value = "/tmp/file.nc" + + return nc_var + + +@pytest.fixture +def nc_var(mocker): + return make_nc_var(mocker) + + +@pytest.fixture +def nc_var_without_group(nc_var): + del nc_var.group + + return nc_var + + +@pytest.fixture +def nc_vars(mocker): + # Three is the maximum number of independent mock variables needed in one test. + return tuple(make_nc_var(mocker) for _ in range(3)) + + +def test_init_records_filename_from_group(nc_var): + cf_var = CFVariableSub("foo", nc_var) + + assert cf_var.filename == "/tmp/file.nc" + assert cf_var.cf_name == "foo" + assert cf_var.cf_data is nc_var + assert cf_var.cf_group is None + assert cf_var.cf_terms_by_root == {} + assert cf_var._to_be_promoted is False + + +def test_init_falls_back_to_unknown_filename_without_group(nc_var_without_group): + cf_var = CFVariableSub("foo", nc_var_without_group) + + assert cf_var.filename == "" + + +def test_identify_common_handles_defaults_and_target_selection(): + variables = {"a": object(), "b": object()} + + ignore, target = CFVariableSub._identify_common(variables, None, None) + assert ignore == [] + assert target is variables + + ignore, target = CFVariableSub._identify_common(variables, ["a"], "b") + assert ignore == ["a"] + assert target == {"b": variables["b"]} + + +def test_identify_common_raises_for_unknown_target(): + with pytest.raises(ValueError, match="Cannot identify unknown target"): + CFVariableSub._identify_common({"a": object()}, None, "missing") + + +def test_identify_common_raises_for_invalid_target_type(): + with pytest.raises(TypeError, match="Expect a target CF-netCDF variable name"): + CFVariableSub._identify_common({"a": object()}, None, object()) + + +def test_spans_scalar_dimension_always_true(mocker, nc_var): + nc_var.dimensions = (cf._NCZARR_SCALAR_DIMENSION,) + cf_var = CFVariableSub("scalar", nc_var) + + other = mocker.MagicMock() + other.dimensions = ("time",) + + assert cf_var.spans(other) + + +def test_spans_is_subset_check(mocker, nc_vars): + lhs_nc_var, other_nc_var, _ = nc_vars + lhs_nc_var.dimensions = ("time",) + lhs = CFVariableSub("lhs", lhs_nc_var) + + rhs = mocker.MagicMock() + rhs.dimensions = ("time", "lat") + + assert lhs.spans(rhs) + + other_nc_var.dimensions = ("height",) + other = CFVariableSub("other", other_nc_var) + assert not other.spans(rhs) + + +def test_equality_inequality_and_hash_by_name(nc_vars): + first, second, third = nc_vars + one = CFVariableSub("same", first) + two = CFVariableSub("same", second) + other = CFVariableSub("different", third) + + assert one == two + assert one != other + assert hash(one) == hash(two) + assert hash(one) != hash(other) + + +def test_cached(nc_var): + # Make sure attribute access to the underlying netCDF4.Variable + # is cached. + name = "foo" + cf_var = CFVariableSub(name, nc_var) + assert nc_var.ncattrs.call_count == 1 + + # Accessing a netCDF attribute should result in no further calls + # to nc_var.ncattrs() and the creation of an attribute on the + # cf_var. + # NB. Can't use hasattr() because that triggers the attribute + # to be created! + assert "coordinates" not in cf_var.__dict__ + _ = cf_var.coordinates + assert nc_var.ncattrs.call_count == 1 + assert "coordinates" in cf_var.__dict__ + + # Trying again results in no change. + _ = cf_var.coordinates + assert nc_var.ncattrs.call_count == 1 + assert "coordinates" in cf_var.__dict__ + + # Trying another attribute results in just a new attribute. + assert "standard_name" not in cf_var.__dict__ + _ = cf_var.standard_name + assert nc_var.ncattrs.call_count == 1 + assert "standard_name" in cf_var.__dict__ + + +def test_getattr_non_ncattr_value_is_cached_but_not_marked_used(nc_var): + nc_var.not_an_ncattr = 42 + cf_var = CFVariableSub("foo", nc_var) + + assert cf_var.not_an_ncattr == 42 + assert "not_an_ncattr" in cf_var.__dict__ + assert "not_an_ncattr" not in cf_var.cf_attrs() + + +def test_getitem_and_len_delegate_to_underlying_variable(nc_var): + cf_var = CFVariableSub("foo", nc_var) + + assert len(cf_var) == 4 + assert cf_var[0] == "payload" + nc_var.__len__.assert_called_once_with() + nc_var.__getitem__.assert_called_once_with(0) + + +def test_repr_contains_class_name_name_and_data_repr(nc_var): + cf_var = CFVariableSub("foo", nc_var) + + assert repr(cf_var) == f"CFVariableSub('foo', {nc_var!r})" + + +def test_cf_attrs_access_helpers_and_reset(nc_var): + cf_var = CFVariableSub("foo", nc_var) + + assert cf_var.cf_attrs() == ( + ("_FillValue", -999), + ("coordinates", "x y"), + ("standard_name", "air_temperature"), + ) + assert cf_var.cf_attrs_ignored() == (("_FillValue", -999),) + assert cf_var.cf_attrs_used() == (("_FillValue", -999),) + assert cf_var.cf_attrs_unused() == ( + ("coordinates", "x y"), + ("standard_name", "air_temperature"), + ) + + _ = cf_var.coordinates + assert cf_var.cf_attrs_used() == (("_FillValue", -999), ("coordinates", "x y")) + + cf_var.cf_attrs_reset() + assert cf_var.cf_attrs_used() == (("_FillValue", -999),) + + +def test_formula_term_registration_and_presence(nc_var): + cf_var = CFVariableSub("foo", nc_var) + + assert not cf_var.has_formula_terms() + cf_var.add_formula_term("root", "a") + assert cf_var.has_formula_terms() + assert cf_var.cf_terms_by_root == {"root": "a"} + + +def test_identify_subclass_stub_returns_none(nc_var): + cf_var = CFVariableSub("foo", nc_var) + + assert cf_var.identify({}) is None diff --git a/lib/iris/tests/unit/fileformats/cf/test__CFFormulaTermsVariable.py b/lib/iris/tests/unit/fileformats/cf/test__CFFormulaTermsVariable.py new file mode 100644 index 0000000000..9bf320ad98 --- /dev/null +++ b/lib/iris/tests/unit/fileformats/cf/test__CFFormulaTermsVariable.py @@ -0,0 +1,177 @@ +# Copyright Iris contributors +# +# This file is part of Iris and is released under the BSD license. +# See LICENSE in the root of the repository for full licensing details. +"""Unit tests for :class:`iris.fileformats.cf._CFFormulaTermsVariable`.""" + +import warnings + +import pytest + +from iris.fileformats.cf import _CFFormulaTermsVariable +import iris.warnings + +CF_IDENTITY = "formula_terms" + + +class TestIdentify: + def test_single_formula_term(self, named_variable): + subject_name = "ref_sigma" + ref_subject = named_variable(subject_name) + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, f"sigma: {subject_name}") + vars_all = { + subject_name: ref_subject, + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + result = _CFFormulaTermsVariable.identify(vars_all) + assert subject_name in result + assert result[subject_name].cf_terms_by_root == {"ref_source": "sigma"} + + def test_multiple_terms_one_source(self, named_variable): + subject_names = ("ref_sigma", "ref_ps") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + ref_source = named_variable("ref_source") + setattr( + ref_source, + CF_IDENTITY, + f"sigma: {subject_names[0]} ps: {subject_names[1]}", + ) + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + **ref_subject_vars, + } + + result = _CFFormulaTermsVariable.identify(vars_all) + assert set(result.keys()) == set(subject_names) + assert result[subject_names[0]].cf_terms_by_root == {"ref_source": "sigma"} + assert result[subject_names[1]].cf_terms_by_root == {"ref_source": "ps"} + + def test_term_name_lowercased(self, named_variable): + """Formula term names must be normalised to lowercase.""" + subject_name = "ref_sigma" + ref_subject = named_variable(subject_name) + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, f"SIGMA: {subject_name}") + vars_all = { + subject_name: ref_subject, + "ref_source": ref_source, + } + + result = _CFFormulaTermsVariable.identify(vars_all) + assert result[subject_name].cf_terms_by_root == {"ref_source": "sigma"} + + def test_same_variable_multiple_roots_aggregates(self, named_variable): + """Same variable referenced by two roots accumulates both terms.""" + subject_name = "ref_sigma" + ref_subject = named_variable(subject_name) + + source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + source_vars["ref_source_1"].formula_terms = f"sigma: {subject_name}" + setattr(source_vars["ref_source_1"], CF_IDENTITY, f"sigma: {subject_name}") + setattr(source_vars["ref_source_2"], CF_IDENTITY, f"eta: {subject_name}") + + vars_all = { + subject_name: ref_subject, + **source_vars, + } + + result = _CFFormulaTermsVariable.identify(vars_all) + assert subject_name in result + terms = result[subject_name].cf_terms_by_root + assert terms.get("ref_source_1") == "sigma" + assert terms.get("ref_source_2") == "eta" + + def test_ignore(self, named_variable): + subject_names = ("ref_sigma", "ref_ps") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + ref_source_vars = { + name: named_variable(name) for name in ("ref_source_1", "ref_source_2") + } + for ix, var in enumerate(ref_source_vars.values()): + setattr(var, CF_IDENTITY, f"sigma: {subject_names[ix]}") + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + result = _CFFormulaTermsVariable.identify(vars_all, ignore=subject_names[1]) + assert subject_names[0] in result + assert subject_names[1] not in result + + def test_target(self, named_variable): + subject_names = ("ref_sigma", "ref_ps") + ref_subject_vars = {name: named_variable(name) for name in subject_names} + + source_names = ("ref_source_1", "ref_source_2") + ref_source_vars = {name: named_variable(name) for name in source_names} + for ix, var in enumerate(ref_source_vars.values()): + setattr(var, CF_IDENTITY, f"sigma: {subject_names[ix]}") + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + **ref_subject_vars, + **ref_source_vars, + } + + result = _CFFormulaTermsVariable.identify(vars_all, target=source_names[0]) + assert subject_names[0] in result + assert subject_names[1] not in result + + def test_target_unknown_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Cannot identify unknown target CF-netCDF variable 'unknown'" + with pytest.raises(ValueError, match=message): + _CFFormulaTermsVariable.identify(vars_all, target="unknown") + + def test_target_wrong_type_raises(self, named_variable): + vars_all = {"ref_source": named_variable("ref_source")} + + message = "Expect a target CF-netCDF variable name" + with pytest.raises(TypeError, match=message): + _CFFormulaTermsVariable.identify(vars_all, target=object()) + + def test_warn(self, named_variable, assert_warning_gated): + subject_name = "ref_sigma" + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, f"sigma: {subject_name}") + vars_all = { + "ref_not_subject": named_variable("ref_not_subject"), + "ref_source": ref_source, + } + + def operation(warn: bool): + warnings.warn( + "emit at least 1 warning", + category=iris.warnings.IrisUserWarning, + ) + _CFFormulaTermsVariable.identify(vars_all, warn=warn) + + warn_regex = rf"Missing CF-netCDF formula term variable {subject_name!r}.*" + assert_warning_gated( + operation, iris.warnings.IrisCfMissingVarWarning, warn_regex + ) + + +class TestRepr: + def test_repr_contains_terms_by_root(self, named_variable): + subject_name = "ref_sigma" + ref_subject = named_variable(subject_name) + ref_source = named_variable("ref_source") + setattr(ref_source, CF_IDENTITY, f"sigma: {subject_name}") + vars_all = { + subject_name: ref_subject, + "ref_source": ref_source, + } + + result = _CFFormulaTermsVariable.identify(vars_all) + repr_str = repr(result[subject_name]) + assert "sigma" in repr_str + assert "ref_source" in repr_str