More

lib/iris/tests/unit/fileformats/test_load_functions.py

@@ -18,7 +18,7 @@
 import pytest
 
 import iris
-from iris.coords import DimCoord
+from iris.coords import AuxCoord, DimCoord
 from iris.cube import Cube
 
 _time_unit = "days since 2001-01-01"
@@ -79,8 +79,19 @@ def mock_generate_cubes(uris, callback, constraints):
     return result
 
 
+def debug_result(cubes):
+    print()
+    print(cubes)
+    if isinstance(cubes, iris.cube.CubeList):
+        print(len(cubes), " cubes..")
+        for i_cube, cube in enumerate(cubes):
+            vh = cube.coord("height").points
+            vt = cube.coord("time").points
+            print(i_cube, cube.name(), ": h=", vh, " :: t=", vt)
+
+
 def check_result(input_cubes, loadfunc_name, result, expected_results):
-    if loadfunc_name == "load_raw":
+    if "load_raw" not in expected_results and loadfunc_name == "load_raw":
         expected = input_cubes
     else:
         expected = expected_results[loadfunc_name]
@@ -109,7 +120,103 @@ def test_multiple(self, loadfunc_name):
         expected_results = {
             "load": [cu(), cu(n="b")],
             "load_cube": "ConstraintMismatchError.*failed to merge into a single cube",
+            "load_cubes": r"ConstraintMismatchError.*-> \d+ cubes",
+        }
+        result = run_testcase(input_cubes, loadfunc_name)
+        check_result(input_cubes, loadfunc_name, result, expected_results)
+
+    def test_multiple_constrained(self, loadfunc_name):
+        cube, cube_b = cu(), cu(n="b")
+        input_cubes = [cube, cube_b]
+        constraint = "a"
+        expected_results = {
+            "load": [cube],
+            "load_cube": cube,
+            "load_cubes": [cube],
+            "load_raw": [cube],
+        }
+        result = run_testcase(input_cubes, loadfunc_name, constraints=constraint)
+        check_result(input_cubes, loadfunc_name, result, expected_results)
+
+    def test_multiple_multi_constraints(self, loadfunc_name):
+        ca, cb, cc = cu(), cu(n="b"), cu(n="c")
+        input_cubes = [ca, cb, cc]
+        constraints = ["c", "a"]
+        expected_results = {
+            "load": [cc, ca],
+            "load_cube": "ValueError.*only a single constraint is allowed",
+            "load_cubes": [cc, ca],
+            "load_raw": [cc, ca],
+        }
+        result = run_testcase(input_cubes, loadfunc_name, constraints=constraints)
+        check_result(input_cubes, loadfunc_name, result, expected_results)
+
+    def test_nonmergeable_part_missing(self, loadfunc_name):
+        c1, c2, c3, c4 = [cu(t=i_t, z=i_z) for i_t in (0, 1) for i_z in (0, 1)]
+        input_cubes = [c1, c2, c4]
+
+        c124 = cu(t=(0, 1, 2))
+        c124.remove_coord("time")  # we now have an unnamed dimension
+        c124.remove_coord("height")  # we now have an unnamed dimension
+        c124.add_aux_coord(AuxCoord([0.0, 1, 1], standard_name="height", units="m"), 0)
+        c124.add_aux_coord(
+            AuxCoord([0.0, 0, 1], standard_name="time", units=_time_unit), 0
+        )
+        expected_results = {
+            "load": [c124],
+            "load_cube": c124,
+            "load_cubes": [c124],
+        }
+        result = run_testcase(input_cubes, loadfunc_name)
+        check_result(input_cubes, loadfunc_name, result, expected_results)
+
+    def test_nonmergeable_part_extra(self, loadfunc_name):
+        cube_all = cu(t=(0, 1), z=(0, 1))
+        c1, c2, c3, c4 = [cu(t=i_t, z=i_z) for i_t in (0, 1) for i_z in (0, 1)]
+        c5 = cu(t=5)
+        input_cubes = [c1, c2, c5, c4, c3]  # scramble order, just to test
+        expected_results = {
+            "load": [cube_all, c5],
+            "load_cube": "ConstraintMismatchError.*failed to merge into a single cube",
             "load_cubes": "ConstraintMismatchError.*-> 2 cubes",
         }
         result = run_testcase(input_cubes, loadfunc_name)
         check_result(input_cubes, loadfunc_name, result, expected_results)
+
+    def test_constraint_overlap(self, loadfunc_name):
+        c1, c2, c3, c4, c5, c6 = (cu(z=ind) for ind in (1, 2, 3, 4, 5, 6))
+        input_cubes = [c1, c2, c3, c4, c5, c6]
+        constraints = [
+            iris.Constraint(height=[1, 2]),
+            iris.Constraint(height=[1, 4, 5]),
+        ]
+        c12 = cu(z=[1, 2])
+        c145 = cu(z=[1, 4, 5])
+        expected_results = {
+            "load": [c12, c145],
+            "load_cube": "ValueError.*only a single constraint is allowed",
+            "load_cubes": [c12, c145],  # selected parts merge, as for load
+            "load_raw": [c1, c2, c1, c4, c5],  # THIS VERY STRANGE BEHAVIOUR!!
+        }
+        result = run_testcase(input_cubes, loadfunc_name, constraints=constraints)
+        check_result(input_cubes, loadfunc_name, result, expected_results)
+
+    def test_multiple_match(self, loadfunc_name):
+        c1 = cu(z=1)
+        c2 = cu(z=2)
+        c3 = cu(n="b", z=1)
+        c4 = cu(n="b", z=2)
+        input_cubes = [c1, c2, c3, c4]
+        constraints = [
+            iris.Constraint("a") & iris.Constraint(height=1),
+            iris.Constraint(height=2),
+        ]
+        expected_results = {
+            "load": [c1, c2, c4],
+            "load_cube": "ValueError.*only a single constraint is allowed",
+            "load_cubes": r"ConstraintMismatchError.*-> \d+ cubes",
+            "load_raw": [c1, c2, c4],
+        }
+        result = run_testcase(input_cubes, loadfunc_name, constraints=constraints)
+        debug_result(result)
+        check_result(input_cubes, loadfunc_name, result, expected_results)