✅ review unittests for cube creation

Author: raphaelrpl

cube_builder/alembic/f639d83487b6_v0_4_0.py

@@ -30,7 +30,7 @@
 
 # revision identifiers, used by Alembic.
 revision = 'f639d83487b6'
-down_revision = None
+down_revision = 'c68b17b1860b'
 branch_labels = ('cube_builder',)
 depends_on = None
 

cube_builder/forms.py

@@ -105,7 +105,7 @@ class CustomMaskDefinition(Schema):
     nodata = fields.Integer(required=True, allow_none=False)
     saturated_data = fields.List(fields.Integer, required=False, allow_none=False)
     saturated_band = fields.String(required=False, allow_none=False)
-    bits = fields.Boolean(required=False, allow_none=False, default=False)
+    bits = fields.Boolean(required=False, allow_none=False, dump_default=False)
 
 
 class CubeParametersSchema(Schema):

cube_builder/utils/image.py

@@ -22,7 +22,7 @@
 import os
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
-from typing import Iterable, List, Optional, Union, Tuple
+from typing import Iterable, List, Optional, Tuple, Union
 from urllib.parse import urlparse
 
 import numpy
@@ -401,7 +401,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     def close(self):
         """Close rasterio data set."""
-        if not self.dataset.closed:
+        if hasattr(self, 'dataset') and not self.dataset.closed:
             logging.debug('Closing dataset {}'.format(str(self.path)))
 
             if self.mode == 'w' and self.tags:
@@ -613,9 +613,14 @@ def rescale(array: ArrayType, multiplier: float, new_scale: float,
     Tip:
         When dealing with negative ``origin_additive`` factor or values which may be negative,
         make sure to use right numpy dtype and
-        `Numpy Masked Arrays <https://numpy.org/doc/stable/reference/maskedarray.html>`_
+        `Numpy Masked Arrays`_
         to mask ``nodata`` values to avoid value limit coercion.
 
+    Note:
+        When the result value overflow the data type,
+        the value is coerced to the data type limits.
+        See more in `numpy.iinfo <https://numpy.org/doc/stable/reference/generated/numpy.iinfo.html>`_.
+
     Args:
         array: Input array
         multiplier: Origin array scale multiplier
@@ -626,6 +631,10 @@ def rescale(array: ArrayType, multiplier: float, new_scale: float,
     Examples:
         This example covers the rescaling Landsat Collection 2 arrays
         (1-65535, scale=0.0000275 - 0.2) into 0-10000 values.
+
+        .. doctest::
+           :skipif: True
+
             >>> import numpy
             >>> from cube_builder.utils.image import rescale
             >>> arr3d = numpy.random.randint(1, 65535, (3, 3), dtype=numpy.uint16)

cube_builder/utils/processing.py

@@ -50,8 +50,8 @@
                          TOTAL_OBSERVATION_NAME)
 # Builder
 from . import get_srid_column
-from .image import (SmartDataSet, generate_cogs, linear_raster_scale, raster_convexhull,
-                    raster_extent, rescale, save_as_cog)
+from .image import (SmartDataSet, generate_cogs, linear_raster_scale, raster_convexhull, raster_extent, rescale,
+                    save_as_cog)
 from .index_generator import generate_band_indexes
 from .strings import StringFormatter
 

tests/test_api.py

@@ -20,9 +20,13 @@
 
 import datetime
 
+from click.testing import CliRunner
 from dateutil.relativedelta import relativedelta
 from flask import Response
 
+from cube_builder import __version__
+from cube_builder.cli import cli
+
 
 def _assert_json_request(response: Response, status_code=200):
     assert response.status_code == status_code
@@ -31,6 +35,12 @@ def _assert_json_request(response: Response, status_code=200):
     return response.json
 
 
+def test_index_api(client):
+    response = client.get('/')
+    assert response.status_code == 200
+    assert response.json['description'] == 'Cube Builder' and response.json['version'] == __version__
+
+
 def test_create_grid(client, json_data):
     response = client.post('/create-grids', json=json_data['grid-bdc-md.json'])
     assert response.status_code == 201
@@ -43,6 +53,11 @@ def test_get_grid(client, json_data):
     assert len(grids) > 0
 
 
+def test_load_initial_data():
+    res = CliRunner().invoke(cli, args=['load-data'])
+    assert res.exit_code == 0
+
+
 def test_create_cube(client, json_data):
     json_cube = json_data['lc8-16d-stk.json']
     response = client.post('/cubes', json=json_cube)
@@ -81,3 +96,57 @@ def test_list_periods_continuous_month(client):
         assert end == (ref_date + offset)
 
         ref_date += offset + relativedelta(days=1)
+
+
+def test_datacube_status(client, json_data):
+    json_cube = json_data['lc8-16d-stk.json']
+    identifier = f"{json_cube['datacube']}-{json_cube['version']}"
+    response = client.get('/cube-status', query_string={'cube_name': identifier})
+    _assert_json_request(response, status_code=200)
+
+    # Test invalid request
+    response = client.get('/cube-status')
+    _assert_json_request(response, status_code=400)
+
+
+def test_list_cubes(client):
+    cube_info = _get_first_cube(client)
+
+    response = client.get(f'/cubes/{cube_info["id"]}')
+    cube = _assert_json_request(response, 200)
+    assert cube['name'] == cube_info['name']
+
+
+def test_update_cube_meta(client):
+    cube = _get_first_cube(client)
+
+    props = dict(
+        title="New Cube - Updated",
+        public=True
+    )
+    response = client.put(f'/cubes/{cube["id"]}', json=props)
+    res = _assert_json_request(response, 200)
+    assert res['message'] == 'Updated cube!'
+    updated_cube = _get_first_cube(client)
+
+    assert updated_cube['title'] == props['title']
+    assert updated_cube['is_public'] == props['public']
+
+    # invalid parameter
+    props['public'] = 'invalid'
+    response = client.put(f'/cubes/{cube["id"]}', json=props)
+    _assert_json_request(response, 400)
+
+
+def test_list_cube_tiles(client):
+    cube = _get_first_cube(client)
+
+    response = client.get(f'/cubes/{cube["id"]}/tiles')
+    _assert_json_request(response, 200)
+
+
+def _get_first_cube(client):
+    response = client.get('/cubes')
+    cubes = _assert_json_request(response, 200)
+    assert len(cubes) > 0
+    return cubes[0]

tests/test_image.py

@@ -67,3 +67,58 @@ def test_create_empty_raster():
             data = ds.read(1)
 
             assert data.min() == data.max() == nodata
+
+
+def test_rescale_raster():
+    """Test scale values of array into a new range.
+
+    This test consists in adapt the values of collections like
+    Landsat Collection 2 Level 2 transforming
+    the range 8000~40000 with additive -0.2 into range 0-10000."""
+    # Landsat C2 L2 values
+    arr2d = numpy.array([
+        [17834, 16269, 8275, 27369],
+        [15755, 19081, 21684, 20865],
+        [24655, 21416, 8932, 32578],
+        [27168, 21827, 11382, 37001],
+    ], dtype=numpy.uint16)
+
+    res = image.rescale(arr2d, 0.0000275, new_scale=0.0001, origin_additive=-0.2)
+
+    assert res.max() < 10000
+    expected = numpy.array([[2904, 2473,  275, 5526],
+                            [2332, 3247, 3963, 3737],
+                            [4780, 3889,  456, 6958],
+                            [5471, 4002, 1130, 8175]], dtype=numpy.uint16)
+    assert numpy.array_equal(res, expected)
+
+
+def test_linear_raster_scale():
+    """Test linear raster scaling to transform band values int16 into byte."""
+    arr2d = numpy.random.randint(0, 10000, (10, 10), dtype=numpy.uint16)
+
+    # Cast to uint8
+    rescaled_arr2d = image.linear_raster_scale(arr2d, (0, arr2d.max(),), (0, 255))
+
+    assert 0 <= rescaled_arr2d.min() <= rescaled_arr2d.max()
+
+    # Using masked array
+    arr2d_ma = numpy.ma.array(arr2d, mask=11000)
+    rescaled_arr2d_ma = image.linear_raster_scale(arr2d_ma, (0, arr2d_ma.max(),), (0, 255))
+    assert 0 <= rescaled_arr2d.min() <= rescaled_arr2d.max()
+
+
+def test_bit_extraction():
+    """Test bit extraction from any value (used for Landsat sensors)."""
+    # 43 => 0010 1011
+    for position in [0, 1, 3, 5]:
+        bit_pos_value = image.extract_qa_bits(43, position)
+        expected = 2 ** position
+        assert bit_pos_value == expected
+
+
+def test_radsat_bit():
+    data = numpy.array([1026, 1037], dtype=numpy.uint16)
+    saturated_values = image.radsat_extract_bits(data, 1, 7)
+    expected = numpy.array([1, 6], dtype=numpy.uint16)
+    assert numpy.array_equal(saturated_values, expected)