Refactor process test example parsing/listing to openeo_test_suite.li…

…b.processes.registry

related to #10 #24

src/openeo_test_suite/lib/internal-tests/processes/test_registry.py

@@ -0,0 +1,43 @@
+import pytest
+
+from openeo_test_suite.lib.processes.registry import ProcessRegistry
+
+
+class TestProcessRegistry:
+    @pytest.fixture(scope="class")
+    def process_registry(self) -> ProcessRegistry:
+        return ProcessRegistry()
+
+    def test_get_all_processes_basic(self, process_registry):
+        processes = list(process_registry.get_all_processes())
+        assert len(processes) > 0
+
+    def test_get_all_processes_add(self, process_registry):
+        (add,) = [
+            p for p in process_registry.get_all_processes() if p.process_id == "add"
+        ]
+
+        assert add.level == "L1"
+        assert add.experimental is False
+        assert add.path.name == "add.json5"
+        assert len(add.tests)
+
+        add00 = {"arguments": {"x": 0, "y": 0}, "returns": 0}
+        assert add00 in add.tests
+
+    def test_get_all_processes_divide(self, process_registry):
+        (divide,) = [
+            p for p in process_registry.get_all_processes() if p.process_id == "divide"
+        ]
+
+        assert divide.level == "L1"
+        assert divide.experimental is False
+        assert divide.path.name == "divide.json5"
+        assert len(divide.tests)
+
+        divide0 = {
+            "arguments": {"x": 1, "y": 0},
+            "returns": float("inf"),
+            "throws": "DivisionByZero",
+        }
+        assert divide0 in divide.tests

src/openeo_test_suite/lib/processes/registry.py

@@ -0,0 +1,56 @@
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterator, List, Optional
+
+import json5
+
+import openeo_test_suite
+
+_log = logging.getLogger(__name__)
+
+DEFAULT_EXAMPLES_ROOT = (
+    Path(openeo_test_suite.__file__).parents[2] / "assets/processes/tests"
+)
+
+
+@dataclass(frozen=True)
+class ProcessData:
+    """Process data, including profile level and list of tests"""
+
+    process_id: str
+    level: str
+    tests: List[dict]  # TODO: also make dataclass for each test?
+    experimental: bool
+    path: Path
+
+
+class ProcessRegistry:
+    """
+    Registry of processes and related tests defined in openeo-processes project
+    """
+
+    def __init__(self, root: Path = DEFAULT_EXAMPLES_ROOT):
+        """
+        :param root: Root directory of the tests folder in  openeo-processes project
+        """
+        self._root = root
+
+    def get_all_processes(self) -> Iterator[ProcessData]:
+        """Collect all processes"""
+        # TODO: cache or preload this in __init__?
+        for path in self._root.glob("*.json5"):
+            try:
+                with path.open() as f:
+                    data = json5.load(f)
+                    assert data["id"] == path.stem
+                    yield ProcessData(
+                        process_id=data["id"],
+                        level=data.get("level"),
+                        tests=data.get("tests", []),
+                        experimental=data.get("experimental", False),
+                        path=path,
+                    )
+            except Exception as e:
+                # TODO: good idea to skip broken definitions? Why not just fail hard?
+                _log.error(f"Failed to load process data from {path}: {e!r}")

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

@@ -10,50 +10,26 @@
 import pytest
 from deepdiff import DeepDiff
 
-import openeo_test_suite
 from openeo_test_suite.lib.process_runner.base import ProcessTestRunner
 from openeo_test_suite.lib.process_runner.util import isostr_to_datetime
+from openeo_test_suite.lib.processes.registry import ProcessRegistry
 
 _log = logging.getLogger(__name__)
 
 
-DEFAULT_EXAMPLES_ROOT = (
-    Path(openeo_test_suite.__file__).parents[2] / "assets/processes/tests"
-)
-
-
-def _get_prop(prop: str, data: dict, test: dict, *, default=None):
-    """Get property from example data, first trying test data, then full process data, then general fallback value."""
-    # TODO make this function more generic (e.g. `multidict_get(key, *dicts, default=None)`)
-    if prop in test:
-        level = test[prop]
-    elif prop in data:
-        level = data[prop]
-    else:
-        level = default
-    return level
-
-
-def get_examples(
-    root: Union[str, Path] = DEFAULT_EXAMPLES_ROOT
-) -> List[Tuple[str, dict, Path, str, bool]]:
+def get_examples() -> List[Tuple[str, dict, Path, str, bool]]:
     """Collect process examples/tests from examples root folder containing JSON5 files."""
-    # TODO return a list of NamedTuples?
-    examples = []
-    # TODO: it's not recommended use `file` (a built-in) as variable name. `path` would be better.
-    for file in root.glob("*.json5"):
-        process_id = file.stem
-        try:
-            with file.open() as f:
-                data = json5.load(f)
-            for test in data["tests"]:
-                level = _get_prop("level", data, test, default="L4")
-                experimental = _get_prop("experimental", data, test, default=False)
-                examples.append((process_id, test, file, level, experimental))
-        except Exception as e:
-            _log.warning(f"Failed to load {file}: {e}")
-
-    return examples
+    return [
+        (
+            process.process_id,
+            test,
+            process.path,
+            test.get("level", process.level),
+            test.get("experimental", process.experimental),
+        )
+        for process in ProcessRegistry().get_all_processes()
+        for test in process.tests
+    ]
 
 
 @pytest.mark.parametrize(