[test] add test cases, add command for test coverage

.coveragerc

@@ -0,0 +1,9 @@
+[run]
+source = src
+omit = ./venv/*,*tests*,*apps.py,*manage.py,*__init__.py,*migrations*,*asgi*,*wsgi*,*admin.py,*urls.py,./tests/*
+
+[report]
+omit = ./venv/*,*tests*,*apps.py,*manage.py,*__init__.py,*migrations*,*asgi*,*wsgi*,*admin.py,*urls.py,./tests/*
+
+exclude_lines =
+     if __name__ == .__main__.:

.gitignore

@@ -7,3 +7,4 @@ tmp/
 *.onnx
 .DS_Store
 .pytest_cache
+.coverage

README.md

@@ -53,4 +53,13 @@ curl -X 'POST' \
     docker push 686901913580.dkr.ecr.eu-west-1.amazonaws.com/lambda-gdal-runner:latest
     docker push 686901913580.dkr.ecr.eu-west-1.amazonaws.com/lambda-fastsam-api:latest
     ```
-3. It's possible to publish a new aws lambda version from cmd or from lambda page
+3. It's possible to publish a new aws lambda version from cmd or from lambda page
+
+
+## Tests
+
+Tests are defined in the `tests` folder in this project. Use PIP to install the test dependencies and run tests.
+
+```bash
+python -m pytest --cov=src --cov-report=term-missing && coverage html
+```

pytest.ini

@@ -0,0 +1,7 @@
+[pytest]
+env_override_existing_values = 1
+env_files =
+    test/.test.env
+[path]
+source = src
+omit = ./venv/*,*tests*,*apps.py,*manage.py,*__init__.py,*migrations*,*asgi*,*wsgi*,*admin.py,*urls.py,./tests/*

requirements.txt

@@ -8,5 +8,6 @@ onnxruntime
 opencv-python
 pillow
 pydantic>=2.0.3
+python-dotenv
 rasterio
 requests

requirements_dev.txt

@@ -8,5 +8,9 @@ onnxruntime
 opencv-python
 pillow
 pydantic>=2.0.3
+pytest
+pytest-cov
+pytest-dotenv
+python-dotenv
 rasterio
 requests

src/io/geo_helpers.py

@@ -20,6 +20,7 @@ def load_affine_transformation_from_matrix(matrix_source_coeffs: List):
         return center * Affine.translation(-0.5, -0.5)
     except Exception as e:
         app_logger.error(f"exception:{e}, check https://github.com/rasterio/affine project for updates")
+        raise e
 
 
 def get_affine_transform_from_gdal(matrix):

tests/io/test_geo_helpers.py

@@ -1,23 +1,80 @@
 import json
+import unittest
+
 import numpy as np
 import shapely
 
+from src.io.geo_helpers import load_affine_transformation_from_matrix
 from tests import TEST_EVENTS_FOLDER
 
 
-def test_get_vectorized_raster_as_geojson():
-    from src.io.geo_helpers import get_vectorized_raster_as_geojson
+class TestGeoHelpers(unittest.TestCase):
+    def test_load_affine_transformation_from_matrix(self):
+        name_fn = "samexporter_predict"
+
+        expected_output = {
+            'europe': (
+                1524458.6551710723, 0.0, 152.87405657035242, 4713262.318571913, -762229.3275855362, -2356860.470370812
+            ),
+            'north_america': (
+                -13855281.495084189, 0.0, 1222.9924525628194, 6732573.451358326, 6927640.747542094, -3368121.214358007
+            ),
+            'oceania': (
+                7269467.138033403, 0.0, 9783.93962050256, -166326.9735485418, -3634733.5690167015, 68487.57734351706
+            ),
+            'south_america': (
+                -7922544.351904369, 0.0, 305.74811314070394, -5432228.234830927, 3961272.1759521845, 2715655.4952457524
+            )}
+
+        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
+            inputs_outputs = json.load(tst_json)
+            for k, input_output in inputs_outputs.items():
+                print(f"k:{k}.")
+
+                output = load_affine_transformation_from_matrix(input_output["input"]["matrix"])
+                assert output.to_shapely() == expected_output[k]
+
+    def test_load_affine_transformation_from_matrix_value_error(self):
+        name_fn = "samexporter_predict"
+        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
+            inputs_outputs = json.load(tst_json)
+            with self.assertRaises(ValueError):
+                try:
+                    io_value_error = inputs_outputs["europe"]["input"]["matrix"][:5]
+                    load_affine_transformation_from_matrix(io_value_error)
+                except ValueError as ve:
+                    print(f"ve:{ve}.")
+                    self.assertEqual(str(ve), "Expected 6 coefficients, found 5; argument type: <class 'list'>.")
+                    raise ve
+
+    def test_load_affine_transformation_from_matrix_exception(self):
+        name_fn = "samexporter_predict"
+        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
+            inputs_outputs = json.load(tst_json)
+            with self.assertRaises(Exception):
+                try:
+                    io_exception = inputs_outputs["europe"]["input"]["matrix"]
+                    io_exception[0] = "ciao"
+                    load_affine_transformation_from_matrix(io_exception)
+                except Exception as e:
+                    print(f"e:{e}.")
+                    self.assertEqual(str(e), "exception:could not convert string to float: 'ciao', "
+                                             "check https://github.com/rasterio/affine project for updates")
+                    raise e
+
+    def test_get_vectorized_raster_as_geojson(self):
+        from src.io.geo_helpers import get_vectorized_raster_as_geojson
 
-    name_fn = "samexporter_predict"
+        name_fn = "samexporter_predict"
 
-    with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-        inputs_outputs = json.load(tst_json)
-        for k, input_output in inputs_outputs.items():
-            print(f"k:{k}.")
-            mask = np.load(TEST_EVENTS_FOLDER / name_fn / k / "mask.npy")
+        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
+            inputs_outputs = json.load(tst_json)
+            for k, input_output in inputs_outputs.items():
+                print(f"k:{k}.")
+                mask = np.load(TEST_EVENTS_FOLDER / name_fn / k / "mask.npy")
 
-            output = get_vectorized_raster_as_geojson(mask=mask, matrix=input_output["input"]["matrix"])
-            assert output["n_shapes_geojson"] == input_output["output"]["n_shapes_geojson"]
-            output_geojson = shapely.from_geojson(output["geojson"])
-            expected_output_geojson = shapely.from_geojson(input_output["output"]["geojson"])
-            assert shapely.equals_exact(output_geojson, expected_output_geojson, tolerance=0.000006)
+                output = get_vectorized_raster_as_geojson(mask=mask, matrix=input_output["input"]["matrix"])
+                assert output["n_shapes_geojson"] == input_output["output"]["n_shapes_geojson"]
+                output_geojson = shapely.from_geojson(output["geojson"])
+                expected_output_geojson = shapely.from_geojson(input_output["output"]["geojson"])
+                assert shapely.equals_exact(output_geojson, expected_output_geojson, tolerance=0.000006)

tests/utilities/test_serialize.py

@@ -0,0 +1,96 @@
+import unittest
+
+import numpy as np
+
+from src.utilities.serialize import serialize
+
+test_dict_list_dict = {
+    "type": "FeatureCollection",
+    "name": "volcanoes",
+    "crs": {"type": "name", "properties": {"name": "urn:ogc:def:crs:OGC:1.3:CRS84"}},
+    "features": [
+        {"type": "Feature", "properties": {"Volcano_Number": 283010, "Volcano_Name": "Izu-Tobu", "prop_none": None},
+         "geometry": {"type": "Point", "coordinates": [139.098, 34.9]}},
+        {"type": "Feature",
+         "properties": {"Volcano_Number": 283020, "Volcano_Name": "Hakoneyama", "ndarray": np.array([1])},
+         "geometry": {"type": "Point", "coordinates": [139.021, 35.233]}}
+    ]
+}
+
+
+class TestSerialize(unittest.TestCase):
+    def test_serialize(self):
+        from bson import ObjectId
+
+        # remove keys with values as bson.ObjectId
+        d1 = {"_id": ObjectId()}
+        self.assertDictEqual(serialize(d1), dict())
+
+        # test: serialize nd.float*, number as key => str
+        np_int_4 = np.asarray([87], dtype=np.integer)[0]
+        d2 = {"b": np.float32(45.0), 3: 33, 1.56: np_int_4, 3.5: 44.0, "d": "b", "tuple": (1, 2)}
+        expected_d2 = {
+            'b': 45.0,
+            3: 33,
+            1.56: 87,
+            3.5: 44.0,
+            'd': 'b',
+            "tuple": [1, 2]
+        }
+        serialized_d2 = serialize(d2)
+        self.assertDictEqual(serialized_d2, expected_d2)
+
+        # # nested dict of list of dict, serialize nd.array
+        d3 = {"e": [{"q": 123}, {"q": 456}], "a": np.arange(1.1, 16.88).reshape(4, 4)}
+        expected_d3 = {
+            "e": [{"q": 123}, {"q": 456}],
+            'a': [[1.1, 2.1, 3.1, 4.1], [5.1, 6.1, 7.1, 8.1], [9.1, 10.1, 11.1, 12.1], [13.1, 14.1, 15.1, 16.1]]
+        }
+        self.assertDictEqual(serialize(d3), expected_d3)
+
+    def test_serialize_dict_exception(self):
+        from json import JSONDecodeError
+
+        e = JSONDecodeError(msg="x", doc="what we are?", pos=111)
+        exception = serialize({"k": e})
+        self.assertDictEqual(
+            exception,
+            {'k': {'msg': 'x', 'type': "<class 'json.decoder.JSONDecodeError'>", 'doc': 'what we are?', 'pos': 111,
+                   'lineno': 1, 'colno': 112}}
+        )
+
+    def test_serialize_bytes(self):
+        self.assertDictEqual(
+            serialize({"k": b"x"}),
+            {'k': {'value': 'eA==', 'type': 'bytes'}}
+        )
+
+    def test_serialize_dict_list_dict(self):
+        serialized_dict_no_none = serialize(test_dict_list_dict, include_none=False)
+        self.assertDictEqual(serialized_dict_no_none, {
+            'type': 'FeatureCollection',
+            'name': 'volcanoes',
+            'crs': {'type': 'name', 'properties': {'name': 'urn:ogc:def:crs:OGC:1.3:CRS84'}},
+            'features': [
+                {'type': 'Feature', 'properties': {'Volcano_Number': 283010, 'Volcano_Name': 'Izu-Tobu'},
+                 'geometry': {'type': 'Point', 'coordinates': [139.098, 34.9]}},
+                {'type': 'Feature',
+                 'properties': {'Volcano_Number': 283020, 'Volcano_Name': 'Hakoneyama', 'ndarray': [1]},
+                 'geometry': {'type': 'Point', 'coordinates': [139.021, 35.233]}}
+            ]
+        })
+
+        serialized_dict_wiht_none = serialize(test_dict_list_dict, include_none=True)
+        self.assertDictEqual(serialized_dict_wiht_none, {
+            'type': 'FeatureCollection',
+            'name': 'volcanoes',
+            'crs': {'type': 'name', 'properties': {'name': 'urn:ogc:def:crs:OGC:1.3:CRS84'}},
+            'features': [
+                {'type': 'Feature',
+                 'properties': {'Volcano_Number': 283010, 'Volcano_Name': 'Izu-Tobu', 'prop_none': None},
+                 'geometry': {'type': 'Point', 'coordinates': [139.098, 34.9]}},
+                {'type': 'Feature',
+                 'properties': {'Volcano_Number': 283020, 'Volcano_Name': 'Hakoneyama', 'ndarray': [1]},
+                 'geometry': {'type': 'Point', 'coordinates': [139.021, 35.233]}}
+            ]
+        })