test: expand suite with helpers, encoding, writer and e2e cases

Author: wietzesuijker

tests/conftest.py

@@ -0,0 +1,15 @@
+import warnings
+
+# Filter noisy Zarr v3 consolidated metadata warning in tests
+warnings.filterwarnings(
+    "ignore",
+    message="Consolidated metadata is currently not part in the Zarr format 3 specification",
+    category=UserWarning,
+)
+
+# Future-proof xarray dims warning until upstream change lands
+warnings.filterwarnings(
+    "ignore",
+    message="The return type of `Dataset.dims` will be changed",
+    category=FutureWarning,
+)

tests/test_cli_e2e.py

@@ -111,13 +111,9 @@ def test_cli_convert_real_sentinel2_data(self, temp_output_dir: str) -> None:
 
         cmd_info = ["python", "-m", "eopf_geozarr", "info", str(output_path)]
 
-        result_info = subprocess.run(
-            cmd_info, capture_output=True, text=True, timeout=60
-        )
+        result_info = subprocess.run(cmd_info, capture_output=True, text=True, timeout=60)
 
-        assert (
-            result_info.returncode == 0
-        ), f"CLI info command failed: {result_info.stderr}"
+        assert result_info.returncode == 0, f"CLI info command failed: {result_info.stderr}"
         print("✅ CLI info command succeeded")
         print(f"Info output: {result_info.stdout}")
 
@@ -138,9 +134,7 @@ def test_cli_convert_real_sentinel2_data(self, temp_output_dir: str) -> None:
             str(output_path),
         ]
 
-        result_validate = subprocess.run(
-            cmd_validate, capture_output=True, text=True, timeout=60
-        )
+        result_validate = subprocess.run(cmd_validate, capture_output=True, text=True, timeout=60)
 
         assert (
             result_validate.returncode == 0
@@ -151,7 +145,8 @@ def test_cli_convert_real_sentinel2_data(self, temp_output_dir: str) -> None:
         # Verify validation output
         validate_output = result_validate.stdout
         assert "Validation Results:" in validate_output, "Should show validation header"
-        assert "✅" in validate_output, "Should show successful validations"
+        success_ok = ("✅" in validate_output) or ("GeoZarr compliant" in validate_output)
+        assert success_ok, "Validation should indicate success"
 
         # Test 4: Verify data structure and compliance
         print("\n=== Verifying converted data structure ===")
@@ -160,9 +155,7 @@ def test_cli_convert_real_sentinel2_data(self, temp_output_dir: str) -> None:
 
         print("✅ All CLI end-to-end tests passed!")
 
-    def _verify_converted_data_structure(
-        self, output_path: Path, groups: list[str]
-    ) -> None:
+    def _verify_converted_data_structure(self, output_path: Path, groups: list[str]) -> None:
         """Verify the structure and compliance of converted data."""
         # Check each group was converted
         for group in groups:
@@ -215,9 +208,7 @@ def _verify_converted_data_structure(
             ds.close()
 
             # Check for overview levels
-            level_dirs = [
-                d for d in group_path.iterdir() if d.is_dir() and d.name.isdigit()
-            ]
+            level_dirs = [d for d in group_path.iterdir() if d.is_dir() and d.name.isdigit()]
             print(f"    Overview levels: {sorted([d.name for d in level_dirs])}")
 
             if len(level_dirs) > 1:
@@ -352,24 +343,15 @@ def test_cli_convert_with_crs_groups(self, temp_output_dir: str) -> None:
 
         # Check command succeeded
         if result.returncode != 0:
-            print(
-                f"CLI convert with --crs-groups failed with return code {result.returncode}"
-            )
+            print(f"CLI convert with --crs-groups failed with return code {result.returncode}")
             print(f"STDOUT: {result.stdout}")
             print(f"STDERR: {result.stderr}")
             # Don't fail the test if CRS groups don't exist in the dataset
             # This is expected behavior for best-effort processing
-            if (
-                "not found in DataTree" in result.stdout
-                or "not found in DataTree" in result.stderr
-            ):
-                print(
-                    "✅ CLI handled missing CRS groups gracefully (expected behavior)"
-                )
+            if "not found in DataTree" in result.stdout or "not found in DataTree" in result.stderr:
+                print("✅ CLI handled missing CRS groups gracefully (expected behavior)")
                 return
-            pytest.fail(
-                f"CLI convert with --crs-groups command failed: {result.stderr}"
-            )
+            pytest.fail(f"CLI convert with --crs-groups command failed: {result.stderr}")
 
         print("✅ CLI convert with --crs-groups command succeeded")
 
@@ -434,9 +416,7 @@ def test_cli_crs_groups_empty_list(self, temp_output_dir: str) -> None:
         result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
 
         # Should succeed (empty crs_groups list is valid)
-        assert (
-            result.returncode == 0
-        ), f"CLI with empty --crs-groups failed: {result.stderr}"
+        assert result.returncode == 0, f"CLI with empty --crs-groups failed: {result.stderr}"
         assert "CRS groups: []" in result.stdout, "Should show empty CRS groups list"
 
         print("✅ CLI with empty --crs-groups list works correctly")

tests/test_conversion.py

@@ -15,10 +15,8 @@
     setup_datatree_metadata_geozarr_spec_compliant,
     validate_existing_band_data,
 )
-from eopf_geozarr.conversion.geozarr import (
-    create_overview_dataset_all_vars,
-    prepare_dataset_with_crs_info,
-)
+from eopf_geozarr.conversion.geozarr import prepare_dataset_with_crs_info
+from eopf_geozarr.conversion.multiscales import create_overview_dataset_all_vars
 
 
 class TestUtilityFunctions:
@@ -123,9 +121,7 @@ def test_validate_existing_band_data_missing(self) -> None:
 
     def test_calculate_overview_levels(self) -> None:
         """Test overview levels calculation."""
-        levels = calculate_overview_levels(
-            1024, 1024, min_dimension=256, tile_width=256
-        )
+        levels = calculate_overview_levels(1024, 1024, min_dimension=256, tile_width=256)
 
         # Should have levels 0, 1, 2 (1024 -> 512 -> 256)
         assert len(levels) == 3
@@ -184,10 +180,7 @@ def test_setup_datatree_metadata_geozarr_spec_compliant(self) -> None:
             assert "standard_name" in processed_ds[band].attrs
             assert "_ARRAY_DIMENSIONS" in processed_ds[band].attrs
             assert "grid_mapping" in processed_ds[band].attrs
-            assert (
-                processed_ds[band].attrs["standard_name"]
-                == "toa_bidirectional_reflectance"
-            )
+            assert processed_ds[band].attrs["standard_name"] == "toa_bidirectional_reflectance"
 
         # Check coordinate attributes
         for coord in ["x", "y"]:
@@ -269,9 +262,7 @@ def test_prepare_dataset_with_crs_info_with_spatial_coordinates(self) -> None:
         dt["conditions/geometry"] = geometry_ds
 
         # Mock the output path and file operations
-        with patch(
-            "eopf_geozarr.conversion.geozarr.fs_utils.normalize_path"
-        ) as mock_normalize:
+        with patch("eopf_geozarr.conversion.geozarr.fs_utils.normalize_path") as mock_normalize:
             with patch(
                 "eopf_geozarr.conversion.geozarr.fs_utils.get_storage_options"
             ) as mock_storage:
@@ -437,15 +428,10 @@ def test_prepare_dataset_with_crs_info_data_variable_attributes(self) -> None:
             if var_name != "spatial_ref":  # Skip grid mapping variable
                 var_attrs = processed_ds[var_name].attrs
                 assert "_ARRAY_DIMENSIONS" in var_attrs
-                assert var_attrs["_ARRAY_DIMENSIONS"] == list(
-                    processed_ds[var_name].dims
-                )
+                assert var_attrs["_ARRAY_DIMENSIONS"] == list(processed_ds[var_name].dims)
 
                 # Variables with spatial coordinates should have grid_mapping
-                if (
-                    "x" in processed_ds[var_name].dims
-                    and "y" in processed_ds[var_name].dims
-                ):
+                if "x" in processed_ds[var_name].dims and "y" in processed_ds[var_name].dims:
                     assert "grid_mapping" in var_attrs
                     assert var_attrs["grid_mapping"] == "spatial_ref"
 
@@ -494,9 +480,7 @@ def test_prepare_dataset_with_crs_info_crs_inference(self) -> None:
                 pixel_size_y = float(y_coords[0] - y_coords[1])  # Usually negative
 
                 # Create GeoTransform (GDAL format)
-                transform_str = (
-                    f"{x_coords[0]} {pixel_size_x} 0.0 {y_coords[0]} 0.0 {pixel_size_y}"
-                )
+                transform_str = f"{x_coords[0]} {pixel_size_x} 0.0 {y_coords[0]} 0.0 {pixel_size_y}"
                 ds["spatial_ref"].attrs["GeoTransform"] = transform_str
 
         # Verify CRS was inferred and applied
@@ -613,9 +597,7 @@ def test_prepare_dataset_with_crs_info_no_spatial_coordinates(self) -> None:
         # Set up data variables with proper attributes
         for var_name in ds.data_vars:
             # Add _ARRAY_DIMENSIONS attribute if missing
-            if "_ARRAY_DIMENSIONS" not in ds[var_name].attrs and hasattr(
-                ds[var_name], "dims"
-            ):
+            if "_ARRAY_DIMENSIONS" not in ds[var_name].attrs and hasattr(ds[var_name], "dims"):
                 ds[var_name].attrs["_ARRAY_DIMENSIONS"] = list(ds[var_name].dims)
 
         # Verify the group was processed but no CRS was added

tests/test_encoding.py

@@ -0,0 +1,80 @@
+import numpy as np
+import xarray as xr
+
+from eopf_geozarr.conversion.encoding import create_geozarr_encoding
+
+
+def _make_ds(shape_2d=(1024, 2048), extra_dims=("band",), bands=3, dtype="uint16"):
+    h, w = shape_2d
+    coords = {"y": np.arange(h), "x": np.arange(w)}
+    data_vars = {}
+    if extra_dims:
+        coords[extra_dims[0]] = np.arange(bands)
+        data = np.random.randint(0, 1000, size=(bands, h, w), dtype=dtype)
+        data_vars["data"] = (extra_dims + ("y", "x"), data)
+    else:
+        data = np.random.randint(0, 1000, size=(h, w), dtype=dtype)
+        data_vars["data"] = (("y", "x"), data)
+    # Add a grid mapping var and reference it via a data variable's attrs
+    data_vars["spatial_ref"] = ((), np.array(0, dtype="int8"))
+    ds = xr.Dataset(data_vars=data_vars, coords=coords)
+    ds["data"].attrs["grid_mapping"] = "spatial_ref"
+    return ds
+
+
+def test_grid_mapping_excluded_from_compression():
+    ds = _make_ds()
+    enc = create_geozarr_encoding(ds, compressor="COMP", spatial_chunk=4096)
+    assert (
+        enc["spatial_ref"].get("compressors") is None
+    ), "grid mapping var should not be compressed"
+    assert enc["data"]["compressors"], "data variable should have compressors list"
+
+
+def test_chunk_shape_3d_leading_dims_collapsed():
+    ds = _make_ds()
+    enc = create_geozarr_encoding(ds, compressor=None, spatial_chunk=512)
+    chunks = enc["data"]["chunks"]
+    # Expect (1, y_chunk, x_chunk)
+    assert len(chunks) == 3 and chunks[0] == 1, f"Leading dim should be 1-sized in chunks: {chunks}"
+    assert chunks[1] == chunks[2], "Spatial chunks should be square"
+    assert chunks[1] <= 512, "Spatial chunk should not exceed requested size"
+
+
+def test_chunk_shape_2d():
+    ds = _make_ds(extra_dims=None)
+    enc = create_geozarr_encoding(ds, compressor=None, spatial_chunk=256)
+    chunks = enc["data"]["chunks"]
+    assert len(chunks) == 2, "2D var should have 2 chunk dims"
+    assert chunks[0] == chunks[1] <= 256
+
+
+def test_max_chunk_bytes_enforced(monkeypatch):
+    ds = _make_ds()
+    # Force very low limit so logic must reduce
+    monkeypatch.setenv("EOPF_MAX_CHUNK_BYTES", "16384")  # 16 KiB
+    enc = create_geozarr_encoding(ds, compressor=None, spatial_chunk=2048)
+    chunks = enc["data"]["chunks"]
+    # Compute approximate bytes
+    dtype_size = ds["data"].dtype.itemsize
+    try:
+        from math import prod
+    except Exception:
+
+        def prod(vals):
+            out = 1
+            for v in vals:
+                out *= v
+            return out
+
+    est_bytes = prod(chunks) * dtype_size
+    assert est_bytes <= 16384, f"Chunk bytes {est_bytes} exceed limit with chunks {chunks}"
+
+
+def test_coord_variables_no_compression():
+    ds = _make_ds()
+    enc = create_geozarr_encoding(ds, compressor=None, spatial_chunk=512)
+    for coord in ds.coords:
+        assert (
+            enc[coord].get("compressors") is None
+        ), f"Coordinate {coord} should not have compressors"