dev

Author: davidhassell

cf/test/test_quantization.py

@@ -2,6 +2,7 @@
 import datetime
 import faulthandler
 import os
+import shutil
 import tempfile
 import unittest
 
@@ -20,6 +21,13 @@
 ]
 [tmpfile1, tmpfile2] = tmpfiles
 
+# Set up temporary directories
+tmpdirs = [
+    tempfile.mkdtemp("_test_quantization.zarr", dir=os.getcwd())
+    for i in range(1)
+]
+[tmpdir] = tmpdirs
+
 
 def _remove_tmpfiles():
     """Remove temporary files created during tests."""
@@ -29,6 +37,13 @@ def _remove_tmpfiles():
         except OSError:
             pass
 
+    for d in tmpdirs:
+        try:
+            shutil.rmtree(d)
+            os.rmdir(d)
+        except OSError:
+            pass
+
 
 atexit.register(_remove_tmpfiles)
 
@@ -222,6 +237,24 @@ def test_quantization_copy(self):
         g = f.copy()
         self.assertTrue(g.get_quantization().equals(q))
 
+    def test_quantization_backends(self):
+        """Test that quantization-on-write with different backends."""
+        f = self.f1.copy()
+        f.set_quantize_on_write(
+            algorithm="granular_bitround", quantization_nsd=8
+        )
+
+        # Backends that allow quantisation-on-write
+        for backend in ("netCDF4",):
+            cf.write(f, tmpfile1, netcdf_backend=backend)
+
+        # Backends that do not allow quantisation-on-write
+        with self.assertRaises(NotImplementedError):
+            cf.write(f, tmpdir, fmt="ZARR3", netcdf_backend="zarr")
+
+        with self.assertRaises(NotImplementedError):
+            cf.write(f, tmpfile1, netcdf_backend="h5netcdf-h5py")
+
 
 if __name__ == "__main__":
     print("Run date:", datetime.datetime.now())

cf/test/test_read_write.py

@@ -1025,20 +1025,29 @@ def test_write_h5py_options(self):
 
         cf.write(
             f,
-            tmpfile0,
+            tmpfile1,
             netcdf_backend="h5netcdf-h5py",
             h5py_options=h5py_options,
         )
-        self.assertTrue(os.path.getsize(tmpfile0) > size)
-
-        cf.write(
-            f, tmpfile1, netcdf_backend="netCDF4", h5py_options=h5py_options
-        )
+        self.assertTrue(os.path.getsize(tmpfile1) > size)
 
         f0 = cf.read(tmpfile0)[0]
         f1 = cf.read(tmpfile1)[0]
         self.assertTrue(f1.equals(f0))
 
+        with self.assertRaises(ValueError):
+            cf.write(
+                f,
+                tmpfile0,
+                netcdf_backend="netCDF4",
+                h5py_options=h5py_options,
+            )
+
+        with self.assertRaises(ValueError):
+            cf.write(
+                f, tmpfile0, fmt="NETCDF3_CLASSIC", h5py_options=h5py_options
+            )
+
     def test_read_netcdf_file(self):
         """Test cf.read for differing the netcdf_file backend."""
         f = self.f0