Merge pull request #90 from silx-kit/test-bin

Add more tests for /data/ endpoint

test/base_test.py

@@ -52,10 +52,9 @@ def test_attr_on_root(self, server):
         retrieved_attributes = decode_response(response)
         assert retrieved_attributes == nx_attributes
 
-    @pytest.mark.parametrize("format_arg", ("json", "npy"))
-    @pytest.mark.parametrize("flatten", (False, True))
-    def test_data_on_array(self, server, format_arg, flatten):
-        """Test /data/ endpoint on array dataset in a group"""
+    @pytest.mark.parametrize("format_arg", ("json", "bin", "npy", "csv", "tiff"))
+    def test_data_on_array_with_format(self, server, format_arg):
+        """Test /data/ endpoint on array dataset"""
         # Test condition
         tested_h5entity_path = "/entry/image"
         data = np.random.random((128, 128))
@@ -65,40 +64,40 @@ def test_data_on_array(self, server, format_arg, flatten):
             h5file[tested_h5entity_path] = data
 
         response = server.get(
-            f"/data/?{urlencode({'file': filename, 'path': tested_h5entity_path, 'format': format_arg, 'flatten': flatten})}"
+            f"/data/?{urlencode({'file': filename, 'path': tested_h5entity_path, 'format': format_arg})}"
+        )
+        retrieved_data = decode_array_response(
+            response, format_arg, data.dtype.str, data.shape
         )
-        retrieved_data = np.array(decode_response(response, format_arg))
 
-        assert np.array_equal(retrieved_data, data.flatten() if flatten else data)
+        assert np.array_equal(retrieved_data, data)
 
     @pytest.mark.parametrize("format_arg", ("npy", "bin"))
-    @pytest.mark.parametrize("dtype_arg", ("origin", "safe"))
-    def test_data_on_array_with_dtype(self, server, format_arg, dtype_arg):
-        """Test /data/ endpoint on array dataset with dtype"""
+    def test_data_on_array_with_dtype_safe(
+        self,
+        server,
+        format_arg,
+    ):
+        """Test /data/ endpoint on array dataset with dtype=safe"""
         # Test condition
         tested_h5entity_path = "/entry/image"
         data = np.random.random((128, 128)).astype(">f2")
         # No Float16Array in JS => converted to float32
-        ref_dtype = "<f4" if dtype_arg == "safe" else ">f2"
 
         filename = "test.h5"
         with h5py.File(server.served_directory / filename, mode="w") as h5file:
             h5file[tested_h5entity_path] = data
 
         response = server.get(
-            f"/data/?{urlencode({'file': filename, 'path': tested_h5entity_path, 'format': format_arg, 'dtype': dtype_arg})}"
-        )
-
-        retrieved_data = decode_array_response(
-            response, format_arg, ref_dtype, data.shape
+            f"/data/?{urlencode({'file': filename, 'path': tested_h5entity_path, 'format': format_arg, 'dtype': 'safe'})}"
         )
 
+        retrieved_data = decode_array_response(response, format_arg, "<f4", data.shape)
         assert np.array_equal(retrieved_data, data)
 
     @pytest.mark.parametrize("format_arg", ("json", "npy"))
-    @pytest.mark.parametrize("flatten", (False, True))
-    def test_data_on_slice(self, server, format_arg, flatten):
-        """Test /data/ endpoint on array dataset in a group"""
+    def test_data_on_slice_with_format_and_flatten(self, server, format_arg):
+        """Test /data/ endpoint on array dataset with flatten"""
         # Test condition
         tested_h5entity_path = "/entry/image"
         data = np.random.random((128, 128))
@@ -108,12 +107,31 @@ def test_data_on_slice(self, server, format_arg, flatten):
             h5file[tested_h5entity_path] = data
 
         response = server.get(
-            f"/data/?{urlencode({'file': filename, 'path': tested_h5entity_path, 'selection': '100,0', 'format': format_arg, 'flatten': flatten})}"
+            f"/data/?{urlencode({'file': filename, 'path': tested_h5entity_path, 'selection': '100,0', 'format': format_arg, 'flatten': True})}"
         )
         retrieved_data = np.array(decode_response(response, format_arg))
 
         assert retrieved_data - data[100, 0] < 1e-8
 
+    def test_data_on_opaque(self, server):
+        """Test /data/ endpoint on opaque dataset with format=bin"""
+        tested_h5entity_path = "/opaque"
+        data = np.void(b"\x00")
+
+        filename = "test.h5"
+        with h5py.File(server.served_directory / filename, mode="w") as h5file:
+            h5file[tested_h5entity_path] = data
+
+        response = server.get(
+            f"/data/?{urlencode({'file': filename, 'path': tested_h5entity_path, 'format': 'bin'})}"
+        )
+
+        content_type = response.find_header_value("content-type")
+        assert content_type == "application/octet-stream"
+
+        retrieved_data = np.void(response.content)
+        assert np.array_equal(retrieved_data, data)
+
     def test_meta_on_chunked_compressed_dataset(self, server):
         """Test /meta/ endpoint on a chunked and compressed dataset"""
         filename = "test.h5"
@@ -517,6 +535,21 @@ def test_422_on_dtype_safe_with_non_numeric_data(self, server):
 
         server.assert_error_code(f"/data/?file={filename}&path={path}&dtype=safe", 422)
 
+    @pytest.mark.parametrize(
+        "format_arg",
+        ("csv", "npy", "tiff"),
+    )
+    def test_422_on_format_incompatible_with_non_numeric_data(self, server, format_arg):
+        filename = "test.h5"
+        path = "/data"
+
+        with h5py.File(server.served_directory / filename, mode="w") as h5file:
+            h5file[path] = "I am not numeric"
+
+        server.assert_error_code(
+            f"/data/?file={filename}&path={path}&format={format_arg}", 422
+        )
+
     def test_422_on_invalid_query_arg(self, server):
         filename = "test.h5"
         path = "/data"

test/utils.py

@@ -3,6 +3,8 @@
 import numpy as np
 from typing import List, NamedTuple, Tuple
 
+import tifffile
+
 from h5grove.utils import hdf_path_join
 
 
@@ -40,14 +42,20 @@ def decode_response(response: Response, format: str = "json"):
     if format == "npy":
         assert content_type == "application/octet-stream"
         return np.load(io.BytesIO(response.content))
+    if format == "csv":
+        assert content_type == "text/csv"
+        return np.genfromtxt(response.content.splitlines(), delimiter=",")
+    if format == "tiff":
+        assert content_type == "image/tiff"
+        return tifffile.imread(io.BytesIO(response.content))
     raise ValueError(f"Unsupported format: {format}")
 
 
 def decode_array_response(
     response: Response,
     format: str,
     dtype: str,
-    shape: Tuple[int],
+    shape: Tuple[int, ...],
 ) -> np.ndarray:
     """Decode data array response content according to given information"""
     content_type = response.find_header_value("content-type")