PR #255 add example unit test

openeo/extra/spectral_indices/spectral_indices.py

@@ -1,6 +1,11 @@
+from pathlib import Path
+import json
+
+import numpy as np
+
 from openeo.processes import ProcessBuilder, array_modify
 from openeo.rest.datacube import DataCube
-import numpy as np
+
 
 def _get_expression_map(cube: DataCube, x: ProcessBuilder):
     collection_id = cube.metadata.get("id")
@@ -68,8 +73,9 @@ def compute_indices(datacube: DataCube, index_list: list, uplim_rescale: int = N
     return: the datacube with the indices attached as bands
 
     """
-    with open("resources/spectral-indices-dict.json") as f:
-        index_specs = eval(f.read())["SpectralIndices"]
+    # TODO: use pkg_resources here instead of direct file reading
+    with (Path(__file__).parent / "resources/spectral-indices-dict.json").open() as f:
+        index_specs = json.load(f)["SpectralIndices"]
     return datacube.apply_dimension(dimension="bands",
                                     process=lambda x: _callback(x, index_list, datacube, uplim_rescale,
                                                                 index_specs)).rename_labels('bands',

tests/extra/spectral_indices/conftest.py

@@ -0,0 +1,41 @@
+import pytest
+
+import openeo
+from openeo.rest.connection import Connection
+from openeo.util import dict_no_none
+
+API_URL = "https://oeo.test"
+
+
+def _setup_connection(api_version, requests_mock) -> Connection:
+    requests_mock.get(API_URL + "/", json={"api_version": api_version})
+
+    # Classic Sentinel2 collection
+    sentinel2_bands = [
+        ("B01", "coastal aerosol"), ("B02", "blue"), ("B03", "green"), ("B04", "red"), ("B05", "nir"),
+        ("B06", None), ("B07", None), ("B08", "nir"), ("B8A", "nir08"), ("B09", "nir09"),
+        ("B11", "swir16"), ("B12", "swir22"),
+    ]
+    requests_mock.get(API_URL + "/collections/SENTINEL2", json={
+        "id": "SENTINEL2",
+        "cube:dimensions": {
+            "x": {"type": "spatial"},
+            "y": {"type": "spatial"},
+            "t": {"type": "temporal"},
+            "bands": {"type": "bands", "values": [n for n, _ in sentinel2_bands]}
+        },
+        "summaries": {
+            "eo:bands": [dict_no_none(name=n, common_name=c) for n, c in sentinel2_bands]
+        },
+    })
+
+    # TODO: add other collections: Landsat, Modis, ProbaV, ...
+
+    return openeo.connect(API_URL)
+
+
+@pytest.fixture
+def con(requests_mock) -> Connection:
+    """Connection fixture to a 1.0.0 backend with some image collections."""
+    return _setup_connection("1.0.0", requests_mock)
+

tests/extra/spectral_indices/test_spectral_indices.py

@@ -0,0 +1,44 @@
+from typing import List, Union
+
+from openeo.extra.spectral_indices.spectral_indices import compute_indices
+from openeo.rest.datacube import DataCube
+
+
+def _extract_process_nodes(cube: Union[dict, DataCube], process_id: str) -> List[dict]:
+    """Extract process node(s) from a data cube or flat graph presentation by process_id"""
+    if isinstance(cube, DataCube):
+        cube = cube.flat_graph()
+    return [d for d in cube.values() if d["process_id"] == process_id]
+
+
+def test_simple_ndvi(con):
+    cube = con.load_collection("Sentinel2")
+    indices = compute_indices(cube, ["NDVI"])
+    apply_dim, = _extract_process_nodes(indices, "apply_dimension")
+    assert apply_dim["arguments"]["process"]["process_graph"] == {
+        "arrayelement1": {
+            "process_id": "array_element",
+            "arguments": {"data": {"from_parameter": "data"}, "index": 7},
+        },
+        "arrayelement2": {
+            "process_id": "array_element",
+            "arguments": {"data": {"from_parameter": "data"}, "index": 3},
+        },
+        "subtract1": {
+            "process_id": "subtract",
+            "arguments": {"x": {"from_node": "arrayelement1"}, "y": {"from_node": "arrayelement2"}},
+        },
+        "add1": {
+            "process_id": "add",
+            "arguments": {"x": {"from_node": "arrayelement1"}, "y": {"from_node": "arrayelement2"}},
+        },
+        "divide1": {
+            "process_id": "divide",
+            "arguments": {"x": {"from_node": "subtract1"}, "y": {"from_node": "add1"}},
+        },
+        "arraymodify1": {
+            "process_id": "array_modify",
+            "arguments": {"data": {"from_parameter": "data"}, "index": 12, "values": [{"from_node": "divide1"}]},
+            "result": True,
+        },
+    }