Integration of OGC API benchmarks

algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json

@@ -0,0 +1,62 @@
+[
+  {
+    "id": "gep_api_ost",
+    "type": "ogc_api_process",
+    "description": "Benchmark for the GEP API OST process",
+    "endpoint": "https://processing.geohazards-tep.eu",
+    "namespace": "44454d3235313539",
+    "application": "opensartoolkit-v2-2-1",
+    "auth": {
+      "url": "https://iam.terradue.com/",
+      "realm": "master"
+    },
+    "parameters": {
+      "resolution": 100,
+      "ard-type": "OST_GTC",
+      "with-speckle-filter": "NO-FILTER",
+      "resampling-method": "BILINEAR_INTERPOLATION",
+      "target_datetime": "2024-11-12T00:00:00Z",
+      "bbox": {
+        "class": "https://raw.githubusercontent.com/eoap/schemas/main/geojson.yaml#Polygon",
+        "type": "Polygon",
+        "bbox": [
+          12.146,
+          41.701,
+          12.629,
+          41.967
+        ],
+        "coordinates": [
+          [
+            [
+              12.146,
+              41.701
+            ],
+            [
+              12.629,
+              41.701
+            ],
+            [
+              12.629,
+              41.967
+            ],
+            [
+              12.146,
+              41.967
+            ],
+            [
+              12.146,
+              41.701
+            ]
+          ]
+        ]
+      }
+    },
+    "reference_data": {
+      "job-results.json": "file:///Users/bramjanssen/projects/apex/apex_algorithms/qa/benchmarks/data/job-results.json",
+      "ost-ard-cog.tif": "file:///Users/bramjanssen/projects/apex/apex_algorithms/qa/benchmarks/data/ost-ard-cog.tif"
+    },
+    "reference_options": {
+      "atol": 1
+    }
+  }
+]

qa/benchmarks/README.md

@@ -41,12 +41,15 @@ with increased verbosity and live logging at DEBUG level:
 pytest -vv --log-cli-level=DEBUG -k '[max_ndvi]'
 ```
 
-## `openeo` client library
-
-The test suite heavily relies on the
-[`openeo` Python client library](https://open-eo.github.io/openeo-python-client/)
-to interact with the openEO backends, e.g. to create openEO jobs,
-wait for their completion, retrieve their results, ...
+## Client libraries
+
+The test suite supports two scenario types:
+- `openeo`: uses the
+    [`openeo` Python client library](https://open-eo.github.io/openeo-python-client/)
+    to create jobs, wait for completion and download results.
+- `ogcapi-processes`: uses the
+    `ogc-api-processes-client` Python library
+    to execute OGC API Processes, poll job status and download result assets.
 
 
 ## Authentication

qa/benchmarks/data/job-results.json

@@ -0,0 +1,9 @@
+{
+  "assets": {
+    "test_output": {
+      "href": "s3://apex-benchmarks/test-output.tif",
+      "type": "image/tiff; application=geotiff",
+      "title": "Test Output"
+    }
+  }
+}

qa/benchmarks/requirements.txt

@@ -1,5 +1,6 @@
 apex-algorithm-qa-tools
 openeo>=0.48.0.dev
+ogc-api-processes-client>=0.6.0
 pytest>=8.2.0
 requests>=2.32.0
 xarray>=2024.6.0

qa/benchmarks/tests/conftest.py

@@ -92,92 +92,4 @@ def pytest_collection_modifyitems(session, config, items):
             selected_ids = set(item.nodeid for item in selected)
             deselected = [item for item in items if item.nodeid not in selected_ids]
             config.hook.pytest_deselected(items=deselected)
-            items[:] = selected
-
-
-def _get_client_credentials_env_var(url: str) -> str:
-    """
-    Get client credentials env var name for a given backend URL.
-    """
-    if not re.match(r"https?://", url):
-        url = f"https://{url}"
-    parsed = urllib.parse.urlparse(url)
-    hostname = parsed.hostname
-    if hostname in {
-        "openeo.dataspace.copernicus.eu",
-        "openeofed.dataspace.copernicus.eu",
-        "openeo.terrascope.be",
-    }:
-        return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSE"
-    elif hostname in {
-        "openeo-staging.dataspace.copernicus.eu",
-        "openeo-staging.terrascope.be",
-        "openeo-dev.terrascope.be",
-    }:
-        return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG"
-    elif re.fullmatch(r"openeo\.dev\.([a-z0-9-]+)\.openeo-int\.v1\.dataspace\.copernicus\.eu", hostname):
-        return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG"
-    elif hostname in {"openeo.cloud", "openeo.eodc.eu"}:
-        return "OPENEO_AUTH_CLIENT_CREDENTIALS_EGI"
-    elif hostname in {
-        "openeo-dev.vito.be",
-        "openeo.vito.be",
-    }:
-        return "OPENEO_AUTH_CLIENT_CREDENTIALS_TERRASCOPE"
-    else:
-        raise ValueError(f"Unsupported backend: {url=} ({hostname=})")
-
-
-@pytest.fixture
-def connection_factory(request, capfd) -> Callable[[], openeo.Connection]:
-    """
-    Fixture for a function that sets up an authenticated connection to an openEO backend.
-
-    This is implemented as a fixture to have access to other fixtures that allow
-    deeper integration with the pytest framework.
-    For example, the `request` fixture allows to identify the currently running test/benchmark.
-    """
-
-    # Identifier for the current test/benchmark, to be injected automatically
-    # into requests to the backend for tracking/cross-referencing purposes
-    origin = f"apex-algorithms/benchmarks/{request.session.name}/{request.node.name}"
-
-    def get_connection(url: str) -> openeo.Connection:
-        session = requests.Session()
-        session.params["_origin"] = origin
-
-        _log.info(f"Connecting to {url!r}")
-        connection = openeo.connect(url, auto_validate=False, session=session)
-        connection.default_headers["X-OpenEO-Client-Context"] = "APEx Algorithm Benchmarks"
-
-        # Authentication:
-        # In production CI context, we want to extract client credentials
-        # from environment variables (based on backend url).
-        # In absence of such environment variables, to allow local development,
-        # we fall back on a traditional `authenticate_oidc()`
-        # which automatically supports various authentication flows (device code, refresh token, client credentials, etc.)
-        auth_env_var = _get_client_credentials_env_var(url)
-        _log.info(f"Checking for {auth_env_var=} to drive auth against {url=}.")
-        if auth_env_var in os.environ:
-            client_credentials = os.environ[auth_env_var]
-            provider_id, client_id, client_secret = client_credentials.split("/", 2)
-            _log.info(f"Extracted {provider_id=} {client_id=} from {auth_env_var=}")
-            connection.authenticate_oidc_client_credentials(
-                provider_id=provider_id,
-                client_id=client_id,
-                client_secret=client_secret,
-            )
-        else:
-            # Temporarily disable output capturing,
-            # to make sure that the OIDC device code instructions are shown
-            # to the user running interactively.
-            with capfd.disabled():
-                # Use a shorter max poll time by default
-                # to alleviate the default impression that the test seem to hang
-                # because of the OIDC device code poll loop.
-                max_poll_time = int(os.environ.get("OPENEO_OIDC_DEVICE_CODE_MAX_POLL_TIME") or 30)
-                connection.authenticate_oidc(max_poll_time=max_poll_time)
-
-        return connection
-
-    return get_connection
+            items[:] = selected

qa/benchmarks/tests/test_benchmarks.py

@@ -1,23 +1,25 @@
 import json
 import logging
-import re
-import signal
 from pathlib import Path
 
-import openeo
 import pytest
-from apex_algorithm_qa_tools.benchmarks import (
-    analyse_results_comparison_exception,
-    collect_metrics_from_job_metadata,
-    collect_metrics_from_results_metadata,
-)
 from apex_algorithm_qa_tools.scenarios import (
     BenchmarkScenario,
     download_reference_data,
     get_benchmark_scenarios,
 )
 from openeo.testing.results import assert_job_results_allclose
 
+from apex_algorithm_qa_tools.benchmarks.runners.factory import (
+    create_benchmark_runner
+)
+
+from apex_algorithm_qa_tools.benchmarks.common import (
+    analyse_results_comparison_exception, 
+    collect_metrics_from_job_metadata, 
+    collect_metrics_from_results_metadata
+)
+
 _log = logging.getLogger(__name__)
 
 
@@ -31,42 +33,43 @@
 )
 def test_run_benchmark(
     scenario: BenchmarkScenario,
-    connection_factory,
     tmp_path: Path,
     track_metric,
     track_phase,
     upload_assets_on_fail,
     request,
 ):
     track_metric("scenario_id", scenario.id)
+    track_metric("scenario_type", scenario.type)
 
     with track_phase(phase="connect"):
-        # Check if a backend override has been provided via cli options.
-        override_backend = request.config.getoption("--override-backend")
-        backend_filter = request.config.getoption("--backend-filter")
-        if backend_filter and not re.match(backend_filter, scenario.backend):
-            # TODO apply filter during scenario retrieval, but seems to be hard to retrieve cli param
-            pytest.skip(
-                f"skipping scenario {scenario.id} because backend {scenario.backend} does not match filter {backend_filter!r}"
-            )
-        backend = scenario.backend
-        if override_backend:
-            _log.info(f"Overriding backend URL with {override_backend!r}")
-            backend = override_backend
-
-        connection: openeo.Connection = connection_factory(url=backend)
+        runner = create_benchmark_runner(
+            request=request,
+            scenario=scenario,
+        )
 
     report_path = None
     if request.config.getoption("--upload-benchmark-report"):
-        report_path = tmp_path / "benchmark_report.json"
-        report_path.write_text(json.dumps({
+        report_data = {
             "scenario_id": scenario.id,
+            "scenario_type": scenario.type,
             "scenario_description": scenario.description,
-            "scenario_backend": scenario.backend,
             "scenario_source": str(scenario.source) if scenario.source else None,
             "reference_data": scenario.reference_data,
             "reference_options": scenario.reference_options,
-        }, indent=2))
+        }
+        # Add scenario type-specific fields if they exist
+        if hasattr(scenario, 'backend'):
+            report_data["scenario_backend"] = scenario.backend
+        if hasattr(scenario, 'process_id'):
+            report_data["process_id"] = scenario.process_id
+        elif hasattr(scenario, 'application'):
+            report_data["application"] = scenario.application
+        if hasattr(scenario, 'endpoint'):
+            report_data["endpoint"] = scenario.endpoint
+
+        report_path = tmp_path / "benchmark_report.json"
+        report_path.write_text(json.dumps(report_data, indent=2))
         upload_assets_on_fail(report_path)
 
     def _on_phase_exception(phase: str, exc: Exception):
@@ -87,50 +90,36 @@ def _on_phase_exception(phase: str, exc: Exception):
 
     track_phase.on_exception = _on_phase_exception
 
-    with track_phase(phase="create-job"):
-        # TODO #14 scenario option to use synchronous instead of batch job mode?
-        job = connection.create_job(
-            process_graph=scenario.process_graph,
-            title=f"APEx benchmark {scenario.id}",
-            additional=scenario.job_options,
-        )
-        track_metric("job_id", job.job_id)
+    artifacts = None
 
-        if report_path is not None:
-            report = json.loads(report_path.read_text())
-            report["job_id"] = job.job_id
-            report_path.write_text(json.dumps(report, indent=2))
+    with track_phase(phase="create-job"):
+        runner.create_job()
 
     with track_phase(phase="run-job"):
-        # TODO: monitor timing and progress
-        # TODO: separate "job started" and run phases?
         max_minutes = request.config.getoption("--maximum-job-time-in-minutes")
-        if max_minutes:
-            def _timeout_handler(signum, frame):
-                raise TimeoutError(
-                    f"Batch job {job.job_id} exceeded maximum allowed time of {max_minutes} minutes"
-                )
-
-            old_handler = signal.signal(signal.SIGALRM, _timeout_handler)
-            signal.alarm(max_minutes * 60)
-        try:
-            job.start_and_wait()
-        finally:
-            if max_minutes:
-                signal.alarm(0)
-                signal.signal(signal.SIGALRM, old_handler)
+        runner.run_job(max_minutes=max_minutes)
 
     with track_phase(phase="collect-metadata"):
-        collect_metrics_from_job_metadata(job, track_metric=track_metric)
-
-        results = job.get_results()
-        collect_metrics_from_results_metadata(results, track_metric=track_metric)
+        artifacts = runner.collect_artifacts()
+        if artifacts.job_id:
+            track_metric("job_id", artifacts.job_id)
+            if report_path is not None:
+                report = json.loads(report_path.read_text())
+                report["job_id"] = artifacts.job_id
+                report_path.write_text(json.dumps(report, indent=2))
+
+        collect_metrics_from_job_metadata(
+            artifacts.job_metadata,
+            track_metric=track_metric,
+        )
+        collect_metrics_from_results_metadata(
+            artifacts.results_metadata,
+            track_metric=track_metric,
+        )
 
     with track_phase(phase="download-actual"):
-        # Download actual results
         actual_dir = tmp_path / "actual"
-        paths = results.download_files(target=actual_dir, include_stac_metadata=True)
-
+        paths = runner.download_actual(actual_dir=actual_dir)
         # Upload assets on failure
         upload_assets_on_fail(*paths)