Make process filtering reusable

- Move logic to `lib` and plugin, with better (internal) test coverage
- Replace fixture based approach in tests/processes/processing with filtering at test collection phase

src/openeo_test_suite/lib/internal-tests/test_process_registry.py

@@ -4,6 +4,12 @@
 
 
 class TestProcessRegistry:
+    # Some example processes for some levels
+    PROCESS_EXAMPLES_L1 = ["add", "divide", "apply_dimension", "reduce_dimension"]
+    PROCESS_EXAMPLES_L2 = ["aggregate_temporal", "if"]
+    PROCESS_EXAMPLES_L3 = ["apply_neighborhood", "merge_cubes"]
+    PROCESS_EXAMPLES_EXPERIMENTAL = ["apply_polygon"]
+
     @pytest.fixture(scope="class")
     def process_registry(self) -> ProcessRegistry:
         return ProcessRegistry()
@@ -41,3 +47,76 @@ def test_get_all_processes_divide(self, process_registry):
             "throws": "DivisionByZero",
         }
         assert divide0 in divide.tests
+
+    def test_get_processes_filtered_default(self, process_registry):
+        pids = [p.process_id for p in process_registry.get_processes_filtered()]
+        assert len(pids) > 100
+        for pid in (
+            self.PROCESS_EXAMPLES_L1
+            + self.PROCESS_EXAMPLES_L2
+            + self.PROCESS_EXAMPLES_L3
+        ):
+            assert pid in pids
+        for pid in self.PROCESS_EXAMPLES_EXPERIMENTAL:
+            assert pid not in pids
+
+    def test_get_processes_filtered_with_process_ids(self, process_registry):
+        pids = [
+            p.process_id
+            for p in process_registry.get_processes_filtered(
+                process_ids=["add", "divide"]
+            )
+        ]
+        assert sorted(pids) == ["add", "divide"]
+
+    def test_get_processes_filtered_with_process_levels(self, process_registry):
+        pids_l1 = [
+            p.process_id
+            for p in process_registry.get_processes_filtered(process_levels=["L1"])
+        ]
+        pids_l23 = [
+            p.process_id
+            for p in process_registry.get_processes_filtered(
+                process_levels=["L2", "L3"]
+            )
+        ]
+        for pid in self.PROCESS_EXAMPLES_L1:
+            assert pid in pids_l1
+            assert pid not in pids_l23
+        for pid in self.PROCESS_EXAMPLES_L2:
+            assert pid not in pids_l1
+            assert pid in pids_l23
+        for pid in self.PROCESS_EXAMPLES_L3:
+            assert pid not in pids_l1
+            assert pid in pids_l23
+        for pid in self.PROCESS_EXAMPLES_EXPERIMENTAL:
+            assert pid not in pids_l1
+            assert pid not in pids_l23
+
+    def test_get_processes_filtered_with_process_ids_and_levels(self, process_registry):
+        pids = [
+            p.process_id
+            for p in process_registry.get_processes_filtered(
+                process_ids=["min", "max"], process_levels=["L2"]
+            )
+        ]
+        for pid in ["min", "max"] + self.PROCESS_EXAMPLES_L2:
+            assert pid in pids
+        for pid in (
+            self.PROCESS_EXAMPLES_L1
+            + self.PROCESS_EXAMPLES_L3
+            + self.PROCESS_EXAMPLES_EXPERIMENTAL
+        ):
+            assert pid not in pids
+
+    def test_get_processes_filtered_with_experimental(self, process_registry):
+        pids = [
+            p.process_id
+            for p in process_registry.get_processes_filtered(
+                process_ids=["min", "max"], process_levels=["L3"], experimental=True
+            )
+        ]
+        for pid in ["min", "max"] + self.PROCESS_EXAMPLES_L3:
+            assert pid in pids
+        for pid in self.PROCESS_EXAMPLES_EXPERIMENTAL:
+            assert pid in pids

src/openeo_test_suite/lib/internal-tests/test_process_selection.py

@@ -0,0 +1,20 @@
+from openeo_test_suite.lib.process_selection import csv_to_list
+
+
+def test_csv_to_list():
+    assert csv_to_list() == []
+    assert csv_to_list("") == []
+    assert csv_to_list("   ") == []
+    assert csv_to_list(" ,  ") == []
+    assert csv_to_list("foo") == ["foo"]
+    assert csv_to_list("foo,bar,baz") == ["foo", "bar", "baz"]
+    assert csv_to_list(",foo,bar,baz,") == ["foo", "bar", "baz"]
+    assert csv_to_list("  ,foo , bar,  baz , ") == ["foo", "bar", "baz"]
+    assert csv_to_list("  ,foo ,,, bar, , baz , ") == ["foo", "bar", "baz"]
+
+
+def test_csv_to_list_none_on_empty():
+    assert csv_to_list(none_on_empty=True) is None
+    assert csv_to_list("", none_on_empty=True) is None
+    assert csv_to_list("   ", none_on_empty=True) is None
+    assert csv_to_list(" ,  ", none_on_empty=True) is None

src/openeo_test_suite/lib/process_registry.py

@@ -2,7 +2,7 @@
 import os
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Iterator, List, Optional
+from typing import Any, Iterable, Iterator, List, Optional, Union
 
 import json5
 
@@ -24,7 +24,8 @@ class ProcessData:
 
 class ProcessRegistry:
     """
-    Registry of processes and related tests defined in openeo-processes project
+    Registry of processes, metadata (level, experimental flag)
+    and related tests defined in openeo-processes project
     """
 
     def __init__(self, root: Optional[Path] = None):
@@ -37,6 +38,8 @@ def __init__(self, root: Optional[Path] = None):
             or os.environ.get("OPENEO_TEST_SUITE_PROCESSES_TEST_ROOT")
             or self._guess_root()
         )
+        # Lazy load cache
+        self._processes: Union[None, List[ProcessData]] = None
 
     def _guess_root(self):
         # TODO: avoid need for guessing and properly include assets in (installed) package
@@ -53,9 +56,9 @@ def _guess_root(self):
             f"Could not find valid processes test root directory (tried {candidates})"
         )
 
-    def get_all_processes(self) -> Iterator[ProcessData]:
+    def _load(self) -> Iterator[ProcessData]:
         """Collect all processes"""
-        # TODO: cache or preload this in __init__?
+        # TODO: cache or preload this in __init__? Or even reuse across instances?
         if not self._root.is_dir():
             raise ValueError(f"Invalid process test root directory: {self._root}")
         _log.info(f"Loading process definitions from {self._root}")
@@ -74,3 +77,36 @@ def get_all_processes(self) -> Iterator[ProcessData]:
             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}")
+
+    def get_all_processes(self) -> Iterable[ProcessData]:
+        if self._processes is None:
+            self._processes = list(self._load())
+        return iter(self._processes)
+
+    def get_processes_filtered(
+        self,
+        process_ids: Optional[List[str]] = None,
+        process_levels: Optional[List[str]] = None,
+        experimental: bool = False,
+    ) -> Iterable[ProcessData]:
+        """
+        Collect processes matching with additional filtering:
+
+        :param process_ids: allow list of process ids (empty/None means allow all)
+        :param process_levels: allow list of process levels (empty/None means allow all)
+        :param experimental: allow experimental processes or not?
+        """
+        for process_data in self.get_all_processes():
+            pid = process_data.process_id
+            level = process_data.level
+
+            if process_data.experimental and not experimental:
+                continue
+
+            if process_ids and pid in process_ids:
+                yield process_data
+            elif process_levels and level in process_levels:
+                yield process_data
+            elif not process_ids and not process_levels:
+                # No id or level allow lists: no filtering
+                yield process_data

src/openeo_test_suite/lib/process_selection.py

@@ -0,0 +1,69 @@
+import logging
+from dataclasses import dataclass
+from typing import Iterable, List, Optional, Union
+
+import pytest
+
+from openeo_test_suite.lib.process_registry import ProcessData, ProcessRegistry
+
+_log = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ProcessFilters:
+    """
+    Container for process filters as specified through command line options
+    `--processes`,`--process-level`, `--experimental`
+    """
+
+    process_ids: Optional[List[str]] = None
+    process_levels: Optional[List[str]] = None
+    experimental: bool = False
+
+
+# Internal singleton pointing to active set of process filters
+# setup happens in `pytest_configure` hook
+_process_filters: Union[ProcessFilters, None] = None
+
+
+def set_process_selection_from_config(config: pytest.Config):
+    """Set up process selection from pytest config (CLI options)."""
+    global _process_filters
+    assert _process_filters is None
+    _process_filters = ProcessFilters(
+        process_ids=csv_to_list(config.getoption("--processes"), none_on_empty=True),
+        process_levels=csv_to_list(
+            config.getoption("--process-levels"), none_on_empty=True
+        ),
+        experimental=config.getoption("--experimental"),
+    )
+
+
+def get_selected_processes() -> Iterable[ProcessData]:
+    """
+    Get effective list of processes extracted from the process registry
+    with filtering based on command line options
+    `--processes`,`--process-level`, `--experimental`
+    """
+    global _process_filters
+    assert isinstance(_process_filters, ProcessFilters)
+
+    return ProcessRegistry().get_processes_filtered(
+        process_ids=_process_filters.process_ids,
+        process_levels=_process_filters.process_levels,
+        experimental=_process_filters.experimental,
+    )
+
+
+def csv_to_list(
+    csv: Union[str, None] = None, *, separator: str = ",", none_on_empty: bool = False
+) -> Union[List[str], None]:
+    """
+    Convert comma-separated string to list of strings,
+    properly taking care of trailing whitespace, empty items, ...
+    """
+    # TODO: options to disable stripping, or to allow empty items?
+    items = [item.strip() for item in (csv or "").split(separator) if item.strip()]
+    if not items and none_on_empty:
+        return None
+    return items

src/openeo_test_suite/lib/pytest_plugin.py

@@ -8,6 +8,7 @@
     get_backend_url,
     set_backend_under_test,
 )
+from openeo_test_suite.lib.process_selection import set_process_selection_from_config
 
 
 def pytest_addoption(parser):
@@ -65,3 +66,5 @@ def pytest_configure(config):
         connection = openeo.connect(url=backend_url, auto_validate=False)
         backend = HttpBackend(connection=connection)
     set_backend_under_test(backend)
+
+    set_process_selection_from_config(config)

src/openeo_test_suite/tests/conftest.py

@@ -13,21 +13,12 @@
 _log = logging.getLogger(__name__)
 
 
-@pytest.fixture(scope="session")
-def skip_experimental(request) -> bool:
-    """
-    Fixture to determine whether experimental functionality should be tested or not.
-    """
-    skip = not request.config.getoption("--experimental")
-    _log.info(f"Skip experimental functionality {skip=}")
-    return skip
-
-
 @pytest.fixture(scope="session")
 def process_levels(request) -> List[str]:
     """
     Fixture to get the desired openEO profiles levels.
     """
+    # TODO: eliminate this fixture?
     levels_str = request.config.getoption("--process-levels")
 
     if isinstance(levels_str, str) and len(levels_str) > 0:
@@ -37,20 +28,6 @@ def process_levels(request) -> List[str]:
         return []
 
 
-@pytest.fixture(scope="session")
-def processes(request) -> List[str]:
-    """
-    Fixture to get the desired processes to test against.
-    """
-    processes_str = request.config.getoption("--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(scope="module")
 def auto_authenticate() -> bool:
     """

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

@@ -12,7 +12,7 @@
 
 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.process_registry import ProcessRegistry
+from openeo_test_suite.lib.process_selection import get_selected_processes
 
 _log = logging.getLogger(__name__)
 
@@ -27,7 +27,7 @@ def get_examples() -> List[Tuple[str, dict, Path, str, bool]]:
             test.get("level", process.level),
             test.get("experimental", process.experimental),
         )
-        for process in ProcessRegistry().get_all_processes()
+        for process in get_selected_processes()
         for test in process.tests
     ]
 
@@ -37,31 +37,15 @@ def get_examples() -> List[Tuple[str, dict, Path, str, bool]]:
 )
 def test_process(
     connection,
-    skip_experimental,
-    process_levels,
-    processes,
     process_id,
     example,
     file,
     level,
     experimental,
     skipper,
 ):
-    skipper.skip_if_unmatching_process_level(level)
-    if len(processes) > 0 and process_id not in processes:
-        pytest.skip(
-            f"Skipping process {process_id!r} because it is not in the specified processes"
-        )
-
-    # check whether the process is available
-    skipper.skip_if_unsupported_process([process_id])
-
-    if skip_experimental and experimental:
-        pytest.skip("Skipping experimental process {}".format(id))
-
-    # check whether any additionally required processes are available
-    if "required" in example:
-        skipper.skip_if_unsupported_process(example["required"])
+    # Check whether the process (and additional extra required ones, if any) is supported on the backend
+    skipper.skip_if_unsupported_process([process_id] + example.get("required", []))
 
     # prepare the arguments from test JSON encoding to internal backend representations
     # or skip if not supported by the test runner