Merge pull request #7 from Open-EO/indiv-processes

Individual process testing updates

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "xarray>=2023.11.0",
     "numpy>=1.26.2",
     "deepdiff>=6.7.1",
+    "python-dateutil>=2.8.2",
 ]
 classifiers = [
     "Programming Language :: Python :: 3",
@@ -32,3 +33,6 @@ Dask = [
 testpaths = [
     "src/openeo_test_suite/tests",
 ]
+filterwarnings = [
+    "ignore:(pkg_resources|jsonschema.RefResolver):DeprecationWarning",
+]

src/openeo_test_suite/lib/process_runner/base.py

@@ -48,9 +48,18 @@ def encode_datacube(self, data: Dict) -> Any:
         """
         raise Exception("datacubes not implemented yet")
 
-    def decode_data(self, data: Any) -> Any:
+    def encode_data(self, data: Any) -> Any:
         """
-        Converts data from the internal backend representation to the process test/JSON5 representation
+        Converts data from the process test/JSON5 representation to the internal backend representation,
+        excluding datacubes and labeled arrays.
+        For example: JSON data types to numpy arrays.
+        openEO process tests specification -> backend
+        """
+        return data
+
+    def decode_data(self, data: Any, expected: Any) -> Any:
+        """
+        Converts data from the internal backend representation to the process test/JSON5 representation.
         For example: numpy values to JSON data types, labeled-array or datacube to
         JSON object representation.
         backend -> openEO process tests specification
@@ -63,3 +72,9 @@ def is_json_only(self) -> bool:
         If True, the runner will skip all tests that contain non JSON values such as infinity and NaN.
         """
         return False
+
+    def get_nodata_value(self) -> Any:
+        """
+        Returns the nodata value of the backend.
+        """
+        return None

src/openeo_test_suite/lib/process_runner/dask.py

@@ -1,6 +1,7 @@
 import importlib
 import inspect
 
+import dask
 from openeo_pg_parser_networkx import OpenEOProcessGraph, ProcessRegistry
 from openeo_pg_parser_networkx.process_registry import Process
 from openeo_processes_dask.process_implementations.core import process
@@ -24,6 +25,11 @@ def create_process_registry():
         )
     ]
 
+    # not sure why this is needed
+    from openeo_processes_dask.process_implementations.math import e
+
+    processes_from_module.append(e)
+
     specs_module = importlib.import_module("openeo_processes_dask.specs")
     specs = {
         func.__name__: getattr(specs_module, func.__name__)
@@ -61,7 +67,14 @@ def encode_process_graph(
     def encode_datacube(self, data):
         return datacube_to_xarray(data)
 
-    def decode_data(self, data):
-        data = numpy_to_native(data)
+    def decode_data(self, data, expected):
+        if isinstance(data, dask.array.core.Array):
+            data = data.compute()
+
+        data = numpy_to_native(data, expected)
         data = xarray_to_datacube(data)
+
         return data
+
+    def get_nodata_value(self):
+        return float("nan")

src/openeo_test_suite/lib/process_runner/util.py

@@ -2,23 +2,30 @@
 
 import numpy as np
 import xarray as xr
+from dateutil.parser import parse
 
 
-def numpy_to_native(data):
+def numpy_to_native(data, expected):
     # Converting numpy dtypes to native python types
     if isinstance(data, np.ndarray) or isinstance(data, np.generic):
-        if data.size == 1:
-            return data.item()
-        elif data.size > 1:
+        if isinstance(expected, list):
             return data.tolist()
+        else:
+            if data.size == 0:
+                return None
+            if data.size == 1:
+                return data.item()
+            elif data.size > 1:
+                return data.tolist()
 
     return data
 
 
 def datacube_to_xarray(cube):
     coords = []
     crs = None
-    for dim in cube["dimensions"]:
+    for name in cube["order"]:
+        dim = cube["dimensions"][name]
         if dim["type"] == "temporal":
             # date replace for older Python versions that don't support ISO parsing (only available since 3.11)
             values = [
@@ -31,7 +38,7 @@ def datacube_to_xarray(cube):
         else:
             values = dim["values"]
 
-        coords.append((dim["name"], values))
+        coords.append((name, values))
 
     da = xr.DataArray(cube["data"], coords=coords)
     if crs is not None:
@@ -45,14 +52,16 @@ def xarray_to_datacube(data):
     if not isinstance(data, xr.DataArray):
         return data
 
-    dims = []
+    order = list(data.dims)
+
+    dims = {}
     for c in data.coords:
         type = "bands"
         values = []
         axis = None
-        if isinstance(data.coords[c].values[0], np.datetime64):
+        if np.issubdtype(data.coords[c].dtype, np.datetime64):
             type = "temporal"
-            values = [iso_datetime(date) for date in data.coords[c].values]
+            values = [datetime_to_isostr(date) for date in data.coords[c].values]
         else:
             values = data.coords[c].values.tolist()
             if c == "x":  # todo: non-standardized
@@ -62,22 +71,32 @@ def xarray_to_datacube(data):
                 type = "spatial"
                 axis = "y"
 
-        dim = {"name": c, "type": type, "values": values}
+        dim = {"type": type, "values": values}
         if axis is not None:
             dim["axis"] = axis
         if "crs" in data.attrs:
             dim["reference_system"] = data.attrs["crs"]  # todo: non-standardized
-        dims.append(dim)
 
-    cube = {"type": "datacube", "dimensions": dims, "data": data.values.tolist()}
+        dims[c] = dim
+
+    cube = {
+        "type": "datacube",
+        "order": order,
+        "dimensions": dims,
+        "data": data.values.tolist(),
+    }
 
     if "nodata" in data.attrs:
         cube["nodata"] = data.attrs["nodata"]  # todo: non-standardized
 
     return cube
 
 
-def iso_datetime(dt):
+def isostr_to_datetime(dt):
+    return parse(dt)
+
+
+def datetime_to_isostr(dt):
     # Convert numpy.datetime64 to timestamp (in seconds)
     timestamp = dt.astype("datetime64[s]").astype(int)
     # Create a datetime object from the timestamp

src/openeo_test_suite/lib/process_runner/vito.py

@@ -22,7 +22,10 @@ def execute(self, id, arguments):
     def encode_datacube(self, data):
         return datacube_to_xarray(data)
 
-    def decode_data(self, data):
-        data = numpy_to_native(data)
+    def decode_data(self, data, expected):
+        data = numpy_to_native(data, expected)
         data = xarray_to_datacube(data)
         return data
+
+    def get_nodata_value(self):
+        return float("nan")

src/openeo_test_suite/tests/conftest.py

@@ -1,5 +1,7 @@
+import argparse
 import logging
 import os
+from distutils.util import strtobool
 
 import openeo
 import pytest
@@ -14,6 +16,13 @@ def pytest_addoption(parser):
         default=None,
         help="The openEO backend URL to connect to.",
     )
+    parser.addoption(
+        "--experimental",
+        type=bool,
+        action=argparse.BooleanOptionalAction,
+        default=False,
+        help="Run tests for experimental functionality or not. By default the tests will be skipped.",
+    )
     parser.addoption(
         "--process-levels",
         action="store",
@@ -61,6 +70,62 @@ def backend_url(request) -> str:
     return url
 
 
+@pytest.fixture(scope="session")
+def skip_experimental(request) -> str:
+    """
+    Fixture to determine whether experimental functionality should be tested or not.
+    """
+    # TODO: also support getting it from a config file?
+    if request.config.getoption("--experimental"):
+        skip = False
+    elif "OPENEO_EXPERIMENTAL" in os.environ:
+        skip = bool(strtobool(os.environ["OPENEO_EXPERIMENTAL"]))
+    else:
+        skip = True
+
+    _log.info(f"Skip experimental functionality {skip!r}")
+
+    return skip
+
+
+@pytest.fixture(scope="session")
+def process_levels(request):
+    """
+    Fixture to get the desired openEO profiles levels.
+    """
+    levels_str = ""
+    # TODO: also support getting it from a config file?
+    if request.config.getoption("--process-levels"):
+        levels_str = request.config.getoption("--process-levels")
+    elif "OPENEO_PROCESS_LEVELS" in os.environ:
+        levels_str = os.environ["OPENEO_PROCESS_LEVELS"]
+
+    if isinstance(levels_str, str) and len(levels_str) > 0:
+        _log.info(f"Testing process levels {levels_str!r}")
+        return list(map(lambda l: l.strip(), levels_str.split(",")))
+    else:
+        return []
+
+
+@pytest.fixture(scope="session")
+def processes(request):
+    """
+    Fixture to get the desired profiles to test against.
+    """
+    processes_str = ""
+    # TODO: also support getting it from a config file?
+    if request.config.getoption("--processes"):
+        processes_str = request.config.getoption("--processes")
+    elif "OPENEO_PROCESSES" in os.environ:
+        processes_str = os.environ["OPENEO_PROCESSES"]
+
+    if isinstance(processes_str, str) and len(processes_str) > 0:
+        _log.info(f"Testing processes {processes_str!r}")
+        return list(map(lambda p: p.strip(), processes_str.split(",")))
+    else:
+        return []
+
+
 @pytest.fixture
 def auto_authenticate() -> bool:
     """

src/openeo_test_suite/tests/processes/README.md

@@ -9,13 +9,16 @@
 
 ## Individual Process Testing
 
-Examples:
+### Examples
 
 - `pytest --openeo-backend-url=https://openeo.cloud --processes=min,max`
 - `pytest --runner=vito --process-levels=L1,L2,L2A`
 - `pytest --runner=dask`
+- `pytest src/openeo_test_suite/tests/processes/processing/test_example.py --runner=dask`
 
-Parameters:
+### Parameters
+
+Specify `src/openeo_test_suite/tests/processes/processing/test_example.py` to only run individual process tests.
 
 - `--runner`: The execution engine. One of:
   - `vito` (needs <https://github.com/Open-EO/openeo-python-driver> being installed)

src/openeo_test_suite/tests/processes/processing/conftest.py

@@ -27,44 +27,6 @@ def runner(request) -> str:
     return runner
 
 
-@pytest.fixture(scope="session")
-def process_levels(request):
-    """
-    Fixture to get the desired openEO profiles levels.
-    """
-    levels_str = ""
-    # TODO: also support getting it from a config file?
-    if request.config.getoption("--process-levels"):
-        levels_str = request.config.getoption("--process-levels")
-    elif "OPENEO_PROCESS_LEVELS" in os.environ:
-        levels_str = os.environ["OPENEO_PROCESS_LEVELS"]
-
-    if isinstance(levels_str, str) and len(levels_str) > 0:
-        _log.info(f"Testing process levels {levels_str!r}")
-        return list(map(lambda l: l.strip(), levels_str.split(",")))
-    else:
-        return []
-
-
-@pytest.fixture(scope="session")
-def processes(request):
-    """
-    Fixture to get the desired profiles to test against.
-    """
-    processes_str = ""
-    # TODO: also support getting it from a config file?
-    if request.config.getoption("--processes"):
-        processes_str = request.config.getoption("--processes")
-    elif "OPENEO_PROCESSES" in os.environ:
-        processes_str = os.environ["OPENEO_PROCESSES"]
-
-    if isinstance(processes_str, str) and len(processes_str) > 0:
-        _log.info(f"Testing processes {processes_str!r}")
-        return list(map(lambda p: p.strip(), processes_str.split(",")))
-    else:
-        return []
-
-
 @pytest.fixture
 def connection(
     backend_url: str, runner: str, auto_authenticate: bool, capfd