issue693 move adapt search_list_for_dict_key to get_enum_options for Schema and add testing

Author: ElienVandermaesenVITO

openeo/internal/processes/parse.py

@@ -43,6 +43,20 @@ def is_geojson_schema(schema) -> bool:
             return any(is_geojson_schema(s) for s in self.schema)
         return False
 
+    def get_enum_options(self,name):
+        result = None
+        if isinstance(self.schema,list):
+            for item in self.schema:
+                if name in item:
+                    if result is None:
+                        result = item[name]
+                    else:
+                        raise ValueError("Multiple entries found with name {v}.".format(v=name))
+        elif isinstance(self.schema,dict):
+            if name in self.schema:
+                result = self.schema[name]
+        return result
+
 
 _NO_DEFAULT = object()
 
@@ -127,11 +141,11 @@ def from_json_file(cls, path: Union[str, Path]) -> Process:
         with Path(path).open("r") as f:
             return cls.from_json(f.read())
 
-    def get_parameter(self, parameter_id: str) -> Union[dict, list]:
+    def get_parameter(self, name: str) -> Parameter:
         for parameter in self.parameters:
-            if parameter.name == parameter_id:
-                return parameter.schema.schema
-        raise LookupError(f"Expected parameter {parameter_id} not found.")
+            if parameter.name == name:
+                return parameter
+        raise LookupError(f"Parameter {name!r} not found.")
 
 
 def parse_all_from_dir(path: Union[str, Path], pattern="*.json") -> Iterator[Process]:

openeo/rest/connection.py

@@ -38,7 +38,6 @@
 from openeo.internal.graph_building import FlatGraphableMixin, PGNode, as_flat_graph
 from openeo.internal.jupyter import VisualDict, VisualList
 from openeo.internal.processes.builder import ProcessBuilderBase
-from openeo.internal.processes.parse import Process
 from openeo.internal.warnings import deprecated, legacy_alias
 from openeo.metadata import (
     Band,

openeo/rest/datacube.py

@@ -66,7 +66,6 @@
     load_json,
     normalize_crs,
     rfc3339,
-    search_list_for_dict_key,
 )
 
 if typing.TYPE_CHECKING:
@@ -2741,10 +2740,10 @@ def sar_backscatter(
         .. versionchanged:: 0.4.10 replace `orthorectify` and `rtc` arguments with `coefficient`.
         """
         try:
-            schema = Process.from_dict(self.connection.describe_process("sar_backscatter")).get_parameter("coefficient")
-            coefficient_options = search_list_for_dict_key(schema, "enum") + [None]
-        except Exception:
-            log.warning(f"Failed to extract option for coefficient in sar_backscatter")
+            schema = Process.from_dict(self.connection.describe_process("sar_backscatter")).get_parameter("coefficient").schema
+            coefficient_options = schema.get_enum_options("enum") + [None]
+        except Exception as e:
+            log.warning(f"Failed to extract coefficient options for process `sar_backscatter`: {e}")
             coefficient_options = [
                 "beta0",
                 "sigma0-ellipsoid",

openeo/util.py

@@ -687,22 +687,3 @@ def normalize_crs(crs: Any, *, use_pyproj: bool = True) -> Union[None, int, str]
             raise ValueError(f"Can not normalize CRS data {type(crs)}")
 
     return crs
-
-
-def search_list_for_dict_key(lst: list, key: str) -> Union[dict, list]:
-    """
-    Searches a value of the dict that matches with the key in the list.
-
-    :param lst: list with dictionaries.
-    :param key: The key for which the value is searched for.
-
-    :return: value that matches key.
-    """
-    result = None
-    for item in lst:
-        if key in item:
-            if result is None:
-                result = item[key]
-            else:
-                raise ValueError("Multiple keys found with value {v}.".format(v=key))
-    return result

tests/internal/processes/test_parse.py

@@ -36,6 +36,35 @@ def test_schema_accepts_geojson(schema, expected):
     assert Schema([{"type": "number"}, schema]).accepts_geojson() == expected
 
 
+@pytest.mark.parametrize(
+        "schema",
+        [
+            Schema([
+                {"x1": "y1"},
+                {"x2": "y2"},
+                {"x3": "y3"},
+                {"x5": "y3"},
+            ]),
+            Schema({"x1":"y1","x2":"y2","x3": "y3","x5": "y3"})
+        ],
+    )
+@pytest.mark.parametrize(("key", "expected"), [("x1", "y1"), ("x2", "y2"), ("x4", None)])
+def test_get_enum_options(schema, key, expected):
+    assert schema.get_enum_options(key) == expected
+
+@pytest.mark.parametrize("key",["x2","x3",])
+def test_get_enum_options_error(key):
+    schema = Schema([
+                {"x1": "y1"},
+                {"x2": "y2"},
+                {"x3": "y3"},
+                {"x3": "y4"},
+                {"x2": "y1"},
+            ])
+    with pytest.raises(ValueError):
+        schema.get_enum_options(key)
+
+
 def test_parameter():
     p = Parameter.from_dict({
         "name": "foo",
@@ -138,6 +167,51 @@ def test_process_from_json():
     assert p.returns.schema.schema == {"type": ["number", "null"], "minimum": 0}
 
 
+@pytest.mark.parametrize(
+    ("key", "expected"), [
+        ("x", Parameter.from_dict({"name": "x", "description": "A number.", "schema": {"type": ["number", "null"]}})),
+        ("y", Parameter.from_dict({"name": "y", "description": "A number.", "schema": {"type": ["number", "null"]}})),
+    ]
+)
+def test_get_parameter(key, expected):
+    p = Process.from_dict({
+        "id": "absolute",
+        "summary": "Absolute value",
+        "description": "Computes the absolute value of a real number.",
+        "categories": ["math"],
+        "parameters": [
+            {"name": "x", "description": "A number.", "schema": {"type": ["number", "null"]}},
+            {"name": "y", "description": "A number.", "schema": {"type": ["number", "null"]}},
+        ],
+        "returns": {
+            "description": "The computed absolute value.",
+            "schema": {"type": ["number", "null"], "minimum": 0}
+        },
+        "links": [{"rel": "about", "href": "http://example.com/abs.html"}],
+    })
+    assert p.get_parameter(key) == expected
+
+
+def test_get_parameter_error():
+    p = Process.from_dict({
+        "id": "absolute",
+        "summary": "Absolute value",
+        "description": "Computes the absolute value of a real number.",
+        "categories": ["math"],
+        "parameters": [
+            {"name": "x", "description": "A number.", "schema": {"type": ["number", "null"]}},
+            {"name": "y", "description": "A number.", "schema": {"type": ["number", "null"]}},
+        ],
+        "returns": {
+            "description": "The computed absolute value.",
+            "schema": {"type": ["number", "null"], "minimum": 0}
+        },
+        "links": [{"rel": "about", "href": "http://example.com/abs.html"}],
+    })
+    with pytest.raises(LookupError):
+        p.get_parameter("z")
+
+
 def test_parse_remote_process_definition_minimal(requests_mock):
     url = "https://example.com/ndvi.json"
     requests_mock.get(url, json={"id": "ndvi"})

tests/test_util.py

@@ -33,7 +33,6 @@
     normalize_crs,
     repr_truncate,
     rfc3339,
-    search_list_for_dict_key,
     str_truncate,
     to_bbox_dict,
     url_join,
@@ -1056,41 +1055,3 @@ def test_normalize_crs_succeeds_with_correct_projstring(self, epsg_input, expect
     def test_normalize_crs_handles_incorrect_crs(self, epsg_input, use_pyproj):
         with pytest.raises(ValueError):
             normalize_crs(epsg_input, use_pyproj=use_pyproj)
-
-    @pytest.mark.parametrize(
-        "list_with_dicts",
-        [
-            [
-                {"x1": "y1"},
-                {"x2": "y2"},
-                {"x3": "y3"},
-                {"x3": "y3"},
-            ]
-        ],
-    )
-    @pytest.mark.parametrize(("key", "expected"), [("x1", "y1"), ("x2", "y2"), ("x4", None)])
-    def test_search_list_for_dict_key(self, list_with_dicts, key, expected):
-        assert search_list_for_dict_key(list_with_dicts, key) == expected
-
-    @pytest.mark.parametrize(
-        "list_with_dicts",
-        [
-            [
-                {"x1": "y1"},
-                {"x2": "y2"},
-                {"x3": "y3"},
-                {"x3": "y4"},
-                {"x2": "y1"},
-            ]
-        ],
-    )
-    @pytest.mark.parametrize(
-        "key",
-        [
-            "x2",
-            "x3",
-        ],
-    )
-    def test_search_list_for_dict_key_value_error(self, list_with_dicts, key):
-        with pytest.raises(ValueError):
-            search_list_for_dict_key(list_with_dicts, key)