From 678899a03ee68bd8f905b5c47532df6c353e43f3 Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Fri, 29 May 2026 15:45:30 +0200 Subject: [PATCH 1/6] benchmarks: started implementation of openeo/ogc split --- qa/benchmarks/README.md | 15 +- qa/benchmarks/requirements.txt | 1 + qa/benchmarks/tests/conftest.py | 88 +--------- qa/benchmarks/tests/test_benchmarks.py | 95 ++++------ .../benchmarks/__init__.py | 0 .../{benchmarks.py => benchmarks/common.py} | 55 ++++-- .../apex_algorithm_qa_tools/benchmarks/ogc.py | 165 ++++++++++++++++++ .../benchmarks/openeo.py | 124 +++++++++++++ .../benchmarks/runners/__init__.py | 0 .../benchmarks/runners/base.py | 43 +++++ .../benchmarks/runners/factory.py | 13 ++ .../benchmarks/runners/ogc.py | 82 +++++++++ .../benchmarks/runners/openeo.py | 60 +++++++ qa/tools/apex_algorithm_qa_tools/scenarios.py | 156 ----------------- .../scenarios/__init__.py | 0 .../scenarios/common.py | 76 ++++++++ .../scenarios/factory.py | 27 +++ .../scenarios/openeo.py | 89 ++++++++++ .../scenarios/scenario.py | 33 ++++ qa/tools/pyproject.toml | 1 + .../tests/pytest/test_pytest_track_metrics.py | 2 +- qa/unittests/tests/test_benchmarks.py | 2 +- qa/unittests/tests/test_scenarios.py | 58 ++++-- schemas/benchmark_scenario.json | 108 +++++++----- 24 files changed, 919 insertions(+), 374 deletions(-) create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/__init__.py rename qa/tools/apex_algorithm_qa_tools/{benchmarks.py => benchmarks/common.py} (59%) create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/runners/__init__.py create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py delete mode 100644 qa/tools/apex_algorithm_qa_tools/scenarios.py create mode 100644 qa/tools/apex_algorithm_qa_tools/scenarios/__init__.py create mode 100644 qa/tools/apex_algorithm_qa_tools/scenarios/common.py create mode 100644 qa/tools/apex_algorithm_qa_tools/scenarios/factory.py create mode 100644 qa/tools/apex_algorithm_qa_tools/scenarios/openeo.py create mode 100644 qa/tools/apex_algorithm_qa_tools/scenarios/scenario.py diff --git a/qa/benchmarks/README.md b/qa/benchmarks/README.md index 67a2ceccd..282254359 100644 --- a/qa/benchmarks/README.md +++ b/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 diff --git a/qa/benchmarks/requirements.txt b/qa/benchmarks/requirements.txt index d6e199406..5bf1fd7bd 100644 --- a/qa/benchmarks/requirements.txt +++ b/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 diff --git a/qa/benchmarks/tests/conftest.py b/qa/benchmarks/tests/conftest.py index 35ead9edb..1715d7b1e 100644 --- a/qa/benchmarks/tests/conftest.py +++ b/qa/benchmarks/tests/conftest.py @@ -94,90 +94,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", - }: - # TODO: env var could just be OPENEO_AUTH_CLIENT_CREDENTIALS_CDSE - # (which should work on both classic CDSE and CDSEfed) - return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED" - elif hostname == "openeo-staging.dataspace.copernicus.eu": - return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG" - elif hostname == "openeo.dev.warsaw.openeo.dataspace.copernicus.eu": - 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", "openeo.terrascope.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 \ No newline at end of file diff --git a/qa/benchmarks/tests/test_benchmarks.py b/qa/benchmarks/tests/test_benchmarks.py index e480bc91a..f059dd7f9 100644 --- a/qa/benchmarks/tests/test_benchmarks.py +++ b/qa/benchmarks/tests/test_benchmarks.py @@ -1,16 +1,9 @@ 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, @@ -18,6 +11,16 @@ ) 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,7 +34,6 @@ ) def test_run_benchmark( scenario: BenchmarkScenario, - connection_factory, tmp_path: Path, track_metric, track_phase, @@ -39,31 +41,24 @@ def test_run_benchmark( 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({ "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, + "process_id": scenario.process_id, "reference_data": scenario.reference_data, "reference_options": scenario.reference_options, }, indent=2)) @@ -87,50 +82,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) diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/__init__.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/common.py similarity index 59% rename from qa/tools/apex_algorithm_qa_tools/benchmarks.py rename to qa/tools/apex_algorithm_qa_tools/benchmarks/common.py index aada509f0..e7ee8a4c1 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/common.py @@ -1,20 +1,49 @@ -""" -Reusable utilities to use in benchmarking -""" +from __future__ import annotations +import dataclasses +import logging +from pathlib import Path +from typing import Any from typing import Union from apex_algorithm_qa_tools.pytest.pytest_track_metrics import MetricsTracker -from openeo.rest.job import BatchJob, JobResults + +_log = logging.getLogger(__name__) + +@dataclasses.dataclass +class BenchmarkExecutionArtifacts: + """Backend-agnostic benchmark execution artifacts.""" + + job_id: str | None + job_metadata: dict | Any + results_metadata: dict | Any + paths: list[Path] + + +def ensure_safe_relative_target(base_dir: Path, relative_path: str) -> Path: + """Resolve a relative path under base_dir and reject path traversal.""" + + root = base_dir.resolve() + target = (base_dir / relative_path).resolve() + if not target.is_relative_to(root): + raise ValueError(f"Unsafe result path: {relative_path!r}") + return target + + +def to_jsonable(value: Any) -> Any: + """Convert pydantic/generated client values to plain JSON-serializable data.""" + + if hasattr(value, "actual_instance"): + value = value.actual_instance + if hasattr(value, "to_dict"): + return value.to_dict() + return value def collect_metrics_from_job_metadata( - job_metadata: Union[BatchJob, dict], + job_metadata: dict, track_metric: MetricsTracker, ): - if isinstance(job_metadata, BatchJob): - job_metadata = job_metadata.describe() - track_metric("costs", job_metadata.get("costs")) for usage_metric, usage_data in job_metadata.get("usage", {}).items(): if "unit" in usage_data and "value" in usage_data: @@ -24,13 +53,9 @@ def collect_metrics_from_job_metadata( def collect_metrics_from_results_metadata( - results_metadata: Union[BatchJob, JobResults, dict], track_metric: MetricsTracker + results_metadata: dict, + track_metric: MetricsTracker ): - if isinstance(results_metadata, BatchJob): - results_metadata = results_metadata.get_results().get_metadata() - elif isinstance(results_metadata, JobResults): - results_metadata = results_metadata.get_metadata() - proj_shape_area_mpx = [] proj_shape_area_km2 = [] for asset_name, asset_data in results_metadata.get("assets", {}).items(): @@ -55,4 +80,4 @@ def collect_metrics_from_results_metadata( def analyse_results_comparison_exception(exc: Exception) -> Union[str, None]: if isinstance(exc, AssertionError): if "Differing 'derived_from' links" in str(exc): - return "derived_from-change" + return "derived_from-change" \ No newline at end of file diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py new file mode 100644 index 000000000..aff1be1ef --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py @@ -0,0 +1,165 @@ +# from __future__ import annotations + +# import importlib +# import json +# import time +# from pathlib import Path +# from typing import Any + +# import requests +# from apex_algorithm_qa_tools.benchmark_execution_shared import ( +# ensure_safe_relative_target, +# to_jsonable, +# ) + + +# def load_ogc_api_components() -> dict[str, Any]: +# try: +# package = importlib.import_module("ogc_api_processes_client") +# execute_api = importlib.import_module( +# "ogc_api_processes_client.api.execute_api" +# ) +# status_api = importlib.import_module( +# "ogc_api_processes_client.api.status_api" +# ) +# result_api = importlib.import_module( +# "ogc_api_processes_client.api.result_api" +# ) +# execute_model = importlib.import_module( +# "ogc_api_processes_client.models.execute" +# ) +# except ModuleNotFoundError as e: +# raise RuntimeError( +# "OGC API processes benchmark requires dependency " +# "'ogc-api-processes-client'" +# ) from e + +# return { +# "ApiClient": package.ApiClient, +# "Configuration": package.Configuration, +# "ExecuteApi": execute_api.ExecuteApi, +# "StatusApi": status_api.StatusApi, +# "ResultApi": result_api.ResultApi, +# "Execute": execute_model.Execute, +# } + + +# def create_ogc_api_client( +# *, +# backend: str, +# request_headers: dict | None, +# ): +# components = load_ogc_api_components() +# configuration = components["Configuration"](host=backend) +# api_client = components["ApiClient"](configuration=configuration) + +# if request_headers: +# api_client.default_headers.update(request_headers) + +# return components, api_client + + +# def execute_ogc_process(*, scenario, components: dict[str, Any], api_client): +# execute_api = components["ExecuteApi"](api_client) +# execute_request = {"inputs": scenario.inputs or {}} +# if scenario.outputs is not None: +# execute_request["outputs"] = scenario.outputs +# if scenario.response is not None: +# execute_request["response"] = scenario.response + +# execute_response = execute_api.execute_with_http_info( +# process_id=scenario.process_id, +# execute=components["Execute"].from_dict(execute_request), +# ) + +# if execute_response.status_code == 201: +# status_info = execute_response.data +# return status_info.job_id, status_info, None +# if execute_response.status_code == 200: +# return None, None, execute_response.data + +# raise RuntimeError( +# "Unexpected HTTP status from OGC execute call: " +# f"{execute_response.status_code}" +# ) + + +# def wait_for_ogc_job(*, components: dict[str, Any], api_client, job_id: str, max_minutes: int | None): +# status_api = components["StatusApi"](api_client) +# deadline = None +# if max_minutes: +# deadline = time.monotonic() + max_minutes * 60 + +# while True: +# status_info = status_api.get_status(job_id=job_id) +# status = str(status_info.status) + +# if status == "successful": +# return status_info +# if status in {"failed", "dismissed"}: +# raise RuntimeError( +# "OGC API process job " +# f"{job_id} ended in status {status!r}: " +# f"{status_info.message!r}" +# ) +# if deadline is not None and time.monotonic() >= deadline: +# raise TimeoutError( +# "OGC API process job " +# f"{job_id} exceeded maximum allowed time " +# f"of {max_minutes} minutes" +# ) + +# time.sleep(5) + + +# def get_ogc_results(*, components: dict[str, Any], api_client, job_id: str): +# result_api = components["ResultApi"](api_client) +# return result_api.get_result(job_id=job_id) + + +# def serialize_ogc_results(results: dict[str, Any]) -> dict[str, Any]: +# return {name: to_jsonable(item) for name, item in results.items()} + + +# def _extract_result_href(item: Any) -> str | None: +# value = to_jsonable(item) +# if not isinstance(value, dict): +# return None +# if isinstance(value.get("href"), str): +# return value["href"] +# nested = value.get("value") +# if isinstance(nested, dict) and isinstance(nested.get("href"), str): +# return nested["href"] +# return None + + +# def download_ogc_results( +# *, +# results: dict[str, Any], +# actual_dir: Path, +# request_headers: dict | None, +# ): +# actual_dir.mkdir(parents=True, exist_ok=True) +# serialized_results = serialize_ogc_results(results) + +# paths = [] +# metadata_path = actual_dir / "job-results.json" +# metadata_path.write_text(json.dumps(serialized_results, indent=2)) +# paths.append(metadata_path) + +# for name, item in results.items(): +# href = _extract_result_href(item) +# if not href: +# continue + +# target = ensure_safe_relative_target(actual_dir, name) +# target.parent.mkdir(parents=True, exist_ok=True) + +# response = requests.get(href, stream=True, headers=request_headers) +# response.raise_for_status() +# with target.open("wb") as f: +# for chunk in response.iter_content(chunk_size=128 * 1024): +# f.write(chunk) +# paths.append(target) + +# return paths diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py new file mode 100644 index 000000000..f3d2fb21b --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py @@ -0,0 +1,124 @@ +import logging +import os +import re + +import openeo +import pytest +import requests +import urllib +from pathlib import Path +import signal + +_log = logging.getLogger(__name__) + + +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", + }: + # TODO: env var could just be OPENEO_AUTH_CLIENT_CREDENTIALS_CDSE + # (which should work on both classic CDSE and CDSEfed) + return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED" + elif hostname == "openeo-staging.dataspace.copernicus.eu": + return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG" + elif hostname == "openeo.dev.warsaw.openeo.dataspace.copernicus.eu": + 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", "openeo.terrascope.be"}: + return "OPENEO_AUTH_CLIENT_CREDENTIALS_TERRASCOPE" + else: + raise ValueError(f"Unsupported backend: {url=} ({hostname=})") + + +def get_openeo_backend(scenario, request): + # 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 " + f"{scenario.backend} does not match filter {backend_filter!r}" + ) + if override_backend: + _log.info(f"Overriding backend URL with {override_backend!r}") + return override_backend + return scenario.backend + + +def create_openeo_connection(*, backend: str, origin: str | None = None) -> openeo.Connection: + session = requests.Session() + session.params["_origin"] = origin + + _log.info(f"Connecting to {backend!r}") + connection = openeo.connect(backend, 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(backend) + _log.info(f"Checking for {auth_env_var=} to drive auth against {backend=}.") + 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: + 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 + + + +def create_openeo_job(*, connection, scenario): + return connection.create_job( + process_graph=scenario.process_graph, + title=f"APEx benchmark {scenario.id}", + additional=scenario.job_options, + ) + + +def run_openeo_job(*, job, max_minutes: int | None): + if max_minutes: + + def _timeout_handler(signum, frame): + raise TimeoutError( + "Batch job " + f"{job.job_id} exceeded maximum allowed time " + f"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) + + +def collect_openeo_metadata(*, job): + return job.get_results() + + +def download_openeo_results(*, results, actual_dir: Path): + return results.download_files(target=actual_dir, include_stac_metadata=True) \ No newline at end of file diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/__init__.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py new file mode 100644 index 000000000..4a3105936 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import dataclasses +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Any + +from apex_algorithm_qa_tools.scenarios import BenchmarkScenario + +@dataclasses.dataclass +class BenchmarkRunnerArtifacts: + job_id: str | None + job_metadata: Any + results_metadata: Any + + +class BenchmarkRunner(ABC): + def __init__(self, *, scenario: BenchmarkScenario, request): + self.scenario = scenario + self.request = request + self.origin = ( + f"apex-algorithms/benchmarks/" + f"{request.session.name}/{request.node.name}" + ) + + @abstractmethod + def create_job(self): + pass + + @abstractmethod + def run_job(self, *, max_minutes: int | None): + pass + + @abstractmethod + def collect_artifacts(self) -> BenchmarkRunnerArtifacts: + pass + + @abstractmethod + def download_actual(self, *, actual_dir: Path) -> list[Path]: + pass + + + diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py new file mode 100644 index 000000000..7c5b35a96 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py @@ -0,0 +1,13 @@ + + +from apex_algorithm_qa_tools.benchmarks.runners.openeo import OpenEOBenchmarkRunner + + +def create_benchmark_runner(*, scenario, request): + if scenario.type == "openeo": + return OpenEOBenchmarkRunner( + scenario=scenario, + request=request, + ) + + raise ValueError(f"Unsupported benchmark scenario type {scenario.type!r}") diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py new file mode 100644 index 000000000..250dd7e30 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py @@ -0,0 +1,82 @@ +# from __future__ import annotations + +# import dataclasses +# from abc import ABC, abstractmethod +# from pathlib import Path +# from typing import Any + +# from apex_algorithm_qa_tools.benchmark_execution_ogcapi import ( +# create_ogc_api_client, +# download_ogc_results, +# execute_ogc_process, +# get_ogc_results, +# serialize_ogc_results, +# wait_for_ogc_job, +# ) +# from apex_algorithm_qa_tools.benchmark_execution_openeo import ( +# collect_openeo_metadata, +# create_openeo_connection, +# create_openeo_job, +# download_openeo_results, +# run_openeo_job, +# ) + + + + +# class OGCAPIBenchmarkRunner(BenchmarkRunner): +# def __init__(self, *, scenario, backend: str): +# super().__init__(scenario=scenario, backend=backend) +# self._components, self._api_client = create_ogc_api_client( +# backend=backend, +# request_headers=scenario.request_headers, +# ) +# self._job_id = None +# self._status_info = None +# self._results = None + +# def create_job(self): +# self._job_id, self._status_info, self._results = execute_ogc_process( +# scenario=self.scenario, +# components=self._components, +# api_client=self._api_client, +# ) + +# def run_job(self, *, max_minutes: int | None): +# if self._job_id is None: +# return + +# self._status_info = wait_for_ogc_job( +# components=self._components, +# api_client=self._api_client, +# job_id=self._job_id, +# max_minutes=max_minutes, +# ) + +# def collect_artifacts(self) -> BenchmarkRunnerArtifacts: +# if self._results is None and self._job_id is not None: +# self._results = get_ogc_results( +# components=self._components, +# api_client=self._api_client, +# job_id=self._job_id, +# ) + +# if not isinstance(self._results, dict): +# raise RuntimeError("OGC API execution did not return a result mapping") + +# job_metadata = self._status_info.to_dict() if self._status_info else {} +# return BenchmarkRunnerArtifacts( +# job_id=self._job_id, +# job_metadata=job_metadata, +# results_metadata=serialize_ogc_results(self._results), +# ) + +# def download_actual(self, *, actual_dir: Path) -> list[Path]: +# if not isinstance(self._results, dict): +# raise RuntimeError("OGC API execution did not return a result mapping") + +# return download_ogc_results( +# results=self._results, +# actual_dir=actual_dir, +# request_headers=self._api_client.default_headers, +# ) \ No newline at end of file diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py new file mode 100644 index 000000000..ac48f88c2 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import logging +from pathlib import Path + +from apex_algorithm_qa_tools.benchmarks.openeo import ( + collect_openeo_metadata, + create_openeo_connection, + create_openeo_job, + download_openeo_results, + get_openeo_backend, + run_openeo_job, +) +from apex_algorithm_qa_tools.benchmarks.runners.base import ( + BenchmarkRunner, + BenchmarkRunnerArtifacts, +) + +from apex_algorithm_qa_tools.scenarios import BenchmarkScenario + +_log = logging.getLogger(__name__) + + +class OpenEOBenchmarkRunner(BenchmarkRunner): + def __init__(self, *, scenario: BenchmarkScenario, request): + super().__init__(scenario=scenario, request=request) + self.backend = get_openeo_backend(scenario, request) + self._connection = create_openeo_connection( + backend=self.backend, + origin=self.origin, + ) + self._job = None + self._results = None + + def create_job(self): + self._job = create_openeo_job( + connection=self._connection, + scenario=self.scenario, + ) + + def run_job(self, *, max_minutes: int | None): + if self._job is None: + raise RuntimeError("Cannot run openEO job before create_job().") + run_openeo_job(job=self._job, max_minutes=max_minutes) + + def collect_artifacts(self) -> BenchmarkRunnerArtifacts: + if self._job is None: + raise RuntimeError("Cannot collect openEO metadata before create_job().") + + self._results = collect_openeo_metadata(job=self._job) + return BenchmarkRunnerArtifacts( + job_id=self._job.job_id, + job_metadata=self._job.describe(), + results_metadata=self._results.get_metadata(), + ) + + def download_actual(self, *, actual_dir: Path) -> list[Path]: + if self._results is None: + raise RuntimeError("Cannot download openEO results before collect_artifacts().") + return download_openeo_results(results=self._results, actual_dir=actual_dir) diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios.py b/qa/tools/apex_algorithm_qa_tools/scenarios.py deleted file mode 100644 index 2f0557a51..000000000 --- a/qa/tools/apex_algorithm_qa_tools/scenarios.py +++ /dev/null @@ -1,156 +0,0 @@ -from __future__ import annotations - -import dataclasses -import glob -import json -import logging -import re -from pathlib import Path -from typing import List - -import jsonschema -import requests -from apex_algorithm_qa_tools.common import ( - assert_no_github_feature_branch_refs, - get_project_root, -) -from openeo.util import TimingLogger - -_log = logging.getLogger(__name__) - - -# TODO #15 Flatten apex_algorithm_qa_tools to a single module and push as much functionality to https://github.com/ESA-APEx/esa-apex-toolbox-python - - -def _get_benchmark_scenario_schema() -> dict: - with open(get_project_root() / "schemas/benchmark_scenario.json") as f: - return json.load(f) - - -@dataclasses.dataclass(kw_only=True) -class BenchmarkScenario: - # TODO #14 need for differentiation between different types of scenarios? - id: str - description: str | None = None - backend: str - process_graph: dict - job_options: dict | None = None - reference_data: dict = dataclasses.field(default_factory=dict) - reference_options: dict = dataclasses.field(default_factory=dict) - source: str | Path | None = None - - @classmethod - def from_dict( - cls, data: dict, *, source: str | Path | None = None - ) -> BenchmarkScenario: - jsonschema.validate(instance=data, schema=_get_benchmark_scenario_schema()) - # TODO: also include the `lint_benchmark_scenario` stuff here (maybe with option to toggle deep URL inspection)? - - # TODO #14 standardization of these types? What about other types and how to support them? - assert data["type"] == "openeo" - return cls( - id=data["id"], - description=data.get("description"), - backend=data["backend"], - process_graph=data["process_graph"], - reference_data=data.get("reference_data", {}), - job_options=data.get("job_options"), - reference_options=data.get("reference_options", {}), - source=source, - ) - - @classmethod - def read_scenarios_file(cls, path: str | Path) -> List[BenchmarkScenario]: - """ - Load list of benchmark scenarios from a JSON file. - """ - path = Path(path) - with path.open("r", encoding="utf8") as f: - data = json.load(f) - # TODO: support single scenario files in addition to listings? - assert isinstance(data, list) - return [cls.from_dict(item, source=path) for item in data] - - -def get_benchmark_scenarios(root=None) -> List[BenchmarkScenario]: - # TODO: instead of flat list, keep original grouping/structure of benchmark scenario files? - # TODO: check for uniqueness of scenario IDs? Also make this a pre-commit lint tool? - scenarios = [] - # old style glob is used to support symlinks - for path in glob.glob( - str((root or get_project_root()) / "algorithm_catalog") - + "/**/*benchmark_scenarios*/*.json", - recursive=True, - ): - scenarios.extend(BenchmarkScenario.read_scenarios_file(path)) - return scenarios - - -def lint_benchmark_scenario(scenario: BenchmarkScenario): - """ - Various sanity checks for scenario data. - To be used in unit tests and pre-commit hooks. - """ - # TODO #17 use JSON Schema based validation instead of ad-hoc checks? - # TODO integrate this as a pre-commit hook - # TODO #17 raise descriptive exceptions instead of asserts? - assert re.match(r"^[a-zA-Z0-9_-]+$", scenario.id) - # TODO proper allow-list of backends or leave this freeform? - backend_patterns = [ - r"^(https:\/\/)?openeo\.dataspace\.copernicus\.eu(\/.*)?$", - r"^(https:\/\/)?openeofed\.dataspace\.copernicus\.eu(\/.*)?$", - r"^(https:\/\/)?openeo\.cloud(\/.*)?$", - r"^(https:\/\/)?openeo\.vito\.be(\/.*)?$", - r"^(https:\/\/)?openeo\.eodc\.eu(\/.*)?$", - ] - assert any(re.fullmatch(p, scenario.backend) for p in backend_patterns), f"Unsupported backend: {scenario.backend!r}" - # TODO: more advanced process graph validation? - assert isinstance(scenario.process_graph, dict) - for node_id, node in scenario.process_graph.items(): - assert isinstance(node, dict) - assert re.match(r"^[a-zA-Z0-9_-]+$", node["process_id"]) - assert "arguments" in node - assert isinstance(node["arguments"], dict) - - if "namespace" in node: - namespace = node["namespace"] - if re.match("^https://", namespace): - if re.match( - "https://github.com/.*/blob/.*", namespace, flags=re.IGNORECASE - ): - raise ValueError( - f"Invalid github.com based namespace {namespace!r}: should be a raw URL" - ) - assert_no_github_feature_branch_refs(namespace) - # Inspect openEO process definition URL - resp = requests.get(namespace) - resp.raise_for_status() - # TODO: also support process listings? - assert resp.json()["id"] == node["process_id"] - # TODO: check that github URL is a "pinned" reference - # TODO: check that provided parameters match expected process parameters - - -def download_reference_data(scenario: BenchmarkScenario, reference_dir: Path) -> Path: - with TimingLogger( - title=f"Downloading reference data for {scenario.id=} to {reference_dir=}", - logger=_log.info, - ): - for path, source in scenario.reference_data.items(): - path = reference_dir / path - if not path.is_relative_to(reference_dir): - raise ValueError( - f"Resolved {path=} is not relative to {reference_dir=} ({scenario.id=})" - ) - path.parent.mkdir(parents=True, exist_ok=True) - - with TimingLogger( - title=f"Downloading {source=} to {path=}", logger=_log.info - ): - # TODO: support other sources than HTTP? - resp = requests.get(source, stream=True) - with path.open("wb") as f: - for chunk in resp.iter_content(chunk_size=128): - f.write(chunk) - - return reference_dir diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/__init__.py b/qa/tools/apex_algorithm_qa_tools/scenarios/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/common.py b/qa/tools/apex_algorithm_qa_tools/scenarios/common.py new file mode 100644 index 000000000..6472507ff --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/common.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import dataclasses +import glob +import json +import logging +import re +from pathlib import Path +from typing import List + +import jsonschema +import requests +from apex_algorithm_qa_tools.common import ( + get_project_root, +) +from openeo.util import TimingLogger + +from apex_algorithm_qa_tools.scenarios.openeo import lint_openeo_fields +from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario +from apex_algorithm_qa_tools.scenarios.factory import read_scenarios_file + + + +def get_benchmark_scenarios(root=None) -> List[BenchmarkScenario]: + # TODO: instead of flat list, keep original grouping/structure of benchmark scenario files? + # TODO: check for uniqueness of scenario IDs? Also make this a pre-commit lint tool? + scenarios = [] + # old style glob is used to support symlinks + for path in glob.glob( + str((root or get_project_root()) / "algorithm_catalog") + "/**/*benchmark_scenarios*/*.json", + recursive=True, + ): + scenarios.extend(read_scenarios_file(path)) + return scenarios + + +def lint_common_fields(scenario: BenchmarkScenario): + """ + Various sanity checks for scenario data. + To be used in unit tests and pre-commit hooks. + """ + assert re.match(r"^[a-zA-Z0-9_-]+$", scenario.id) + + +def lint_benchmark_scenario(scenario: BenchmarkScenario): + """ + Various sanity checks for scenario data. + To be used in unit tests and pre-commit hooks. + """ + lint_common_fields(scenario=scenario) + + if scenario.type == "openeo": + lint_openeo_fields(scenario=scenario) + else: + raise ValueError(f"Unsupported benchmark scenario type: {scenario.type!r}") + + +def download_reference_data(scenario: BenchmarkScenario, reference_dir: Path) -> Path: + with TimingLogger( + title=f"Downloading reference data for {scenario.id=} to {reference_dir=}", + logger=_log.info, + ): + for path, source in scenario.reference_data.items(): + path = reference_dir / path + if not path.is_relative_to(reference_dir): + raise ValueError(f"Resolved {path=} is not relative to {reference_dir=} ({scenario.id=})") + path.parent.mkdir(parents=True, exist_ok=True) + + with TimingLogger(title=f"Downloading {source=} to {path=}", logger=_log.info): + # TODO: support other sources than HTTP? + resp = requests.get(source, stream=True) + with path.open("wb") as f: + for chunk in resp.iter_content(chunk_size=128): + f.write(chunk) + + return reference_dir \ No newline at end of file diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py b/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py new file mode 100644 index 000000000..c7755432b --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py @@ -0,0 +1,27 @@ +import json +from pathlib import Path +from typing import List + +from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario +from apex_algorithm_qa_tools.scenarios.openeo import openEOBenchmarkScenario + + +def read_scenarios_file(path: str | Path) -> List[BenchmarkScenario]: + """ + Load list of benchmark scenarios from a JSON file. + """ + path = Path(path) + with path.open("r", encoding="utf8") as f: + data = json.load(f) + # TODO: support single scenario files in addition to listings? + assert isinstance(data, list) + return [_benchmark_factory(item, path) for item in data] + + +def _benchmark_factory(item: dict, path: str | Path) -> BenchmarkScenario: + # TODO #14 need for differentiation between different types of scenarios? + assert item["type"] is not None, "Missing required 'type' field in benchmark scenario" + if item["type"] == "openeo": + return openEOBenchmarkScenario.from_dict(data=item, source=path) + else: + raise ValueError(f"Unsupported benchmark scenario type: {item['type']}") diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/openeo.py b/qa/tools/apex_algorithm_qa_tools/scenarios/openeo.py new file mode 100644 index 000000000..abc74af60 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/openeo.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +import dataclasses +import json +from pathlib import Path +import re +from typing import List + +import jsonschema +import requests + +from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario,get_benchmark_scenario_schema +from apex_algorithm_qa_tools.common import assert_no_github_feature_branch_refs + + +BACKEND_PATTERNS = [ + r"^(https://)?openeo\.dataspace\.copernicus\.eu(/.*)?$", + r"^(https://)?openeofed\.dataspace\.copernicus\.eu(/.*)?$", + r"^(https://)?openeo\.cloud(/.*)?$", + r"^(https://)?openeo\.vito\.be(/.*)?$", + r"^(https://)?openeo\.eodc\.eu(/.*)?$", +] + + +@dataclasses.dataclass(kw_only=True) +class openEOBenchmarkScenario(BenchmarkScenario): + backend: str + process_graph: dict + job_options: dict | None = None + + @classmethod + def from_dict(cls, data: dict, source: str | Path | None = None) -> BenchmarkScenario: + jsonschema.validate(instance=data, schema=get_benchmark_scenario_schema()) + # TODO: also include the `lint_benchmark_scenario` stuff here (maybe with option to toggle deep URL inspection)? + + # TODO #14 standardization of these types? What about other types and how to support them? + assert data["type"] == "openeo" + return cls( + id=data["id"], + type=data["type"], + description=data.get("description"), + backend=data["backend"], + process_graph=data["process_graph"], + job_options=data.get("job_options"), + reference_data=data.get("reference_data", {}), + reference_options=data.get("reference_options", {}), + source=source, + ) + +def lint_openeo_fields(scenario: BenchmarkScenario): + lint_backend(scenario=scenario) + lint_openeo_process_graph(scenario=scenario) + + +def lint_backend(scenario: BenchmarkScenario): + assert any(re.fullmatch(pattern, scenario.backend) for pattern in BACKEND_PATTERNS), ( + f"Unsupported backend: {scenario.backend!r}" + ) + + +def lint_namespace_reference(namespace: str): + if not re.match(r"^https://", namespace): + return + + if re.match(r"https://github.com/.*/blob/.*", namespace, flags=re.IGNORECASE): + raise ValueError(f"Invalid github.com based namespace {namespace!r}: should be a raw URL") + + assert_no_github_feature_branch_refs(namespace) + + +def lint_openeo_process_graph(scenario: BenchmarkScenario): + assert isinstance(scenario.process_graph, dict) + + for node in scenario.process_graph.values(): + assert isinstance(node, dict) + assert re.match(r"^[a-zA-Z0-9_-]+$", node["process_id"]) + assert "arguments" in node + assert isinstance(node["arguments"], dict) + + namespace = node.get("namespace") + if namespace is None: + continue + + lint_namespace_reference(namespace) + + if re.match(r"^https://", namespace): + response = requests.get(namespace) + response.raise_for_status() + assert response.json()["id"] == node["process_id"] diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/scenario.py b/qa/tools/apex_algorithm_qa_tools/scenarios/scenario.py new file mode 100644 index 000000000..fc487feab --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/scenario.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from abc import ABC +import dataclasses +import json +from pathlib import Path +from typing import List + + +from apex_algorithm_qa_tools.common import ( + get_project_root, +) + + +# TODO #15 Flatten apex_algorithm_qa_tools to a single module and push as much functionality to https://github.com/ESA-APEx/esa-apex-toolbox-python +def get_benchmark_scenario_schema() -> dict: + with open(get_project_root() / "schemas/benchmark_scenario.json") as f: + return json.load(f) + + + +@dataclasses.dataclass(kw_only=True) +class BenchmarkScenario(ABC): + id: str + type: str + description: str | None = None + reference_data: dict = dataclasses.field(default_factory=dict) + reference_options: dict = dataclasses.field(default_factory=dict) + source: str | Path | None = None + + @classmethod + def from_dict(cls, data: dict, source: str | Path | None = None) -> BenchmarkScenario: + pass \ No newline at end of file diff --git a/qa/tools/pyproject.toml b/qa/tools/pyproject.toml index b3ec25a89..92855ec6d 100644 --- a/qa/tools/pyproject.toml +++ b/qa/tools/pyproject.toml @@ -21,6 +21,7 @@ dependencies = [ "requests>=2.32.0", "jsonschema>=4.0.0", "openeo>=0.30.0", + "ogc-api-processes-client>=0.6.0", # TODO: make some of these dependencies optional "boto3>=1.36.5", "pyarrow>=17.0.0", diff --git a/qa/unittests/tests/pytest/test_pytest_track_metrics.py b/qa/unittests/tests/pytest/test_pytest_track_metrics.py index 064221f3b..072907556 100644 --- a/qa/unittests/tests/pytest/test_pytest_track_metrics.py +++ b/qa/unittests/tests/pytest/test_pytest_track_metrics.py @@ -332,7 +332,7 @@ def test_track_phase_describe_derived_from_change(pytester: pytest.Pytester, tmp pytester.makeconftest(CONTENT_CONFTEST) src = f""" - from apex_algorithm_qa_tools.benchmarks import analyse_results_comparison_exception + from apex_algorithm_qa_tools.benchmarks.common import analyse_results_comparison_exception import openeo.testing.results def test_derived_from(track_phase): diff --git a/qa/unittests/tests/test_benchmarks.py b/qa/unittests/tests/test_benchmarks.py index 6ba0f045e..9d7b7881b 100644 --- a/qa/unittests/tests/test_benchmarks.py +++ b/qa/unittests/tests/test_benchmarks.py @@ -5,7 +5,7 @@ import openeo.rest.job import openeo.testing.results import pytest -from apex_algorithm_qa_tools.benchmarks import ( +from apex_algorithm_qa_tools.benchmarks.common import ( analyse_results_comparison_exception, collect_metrics_from_job_metadata, collect_metrics_from_results_metadata, diff --git a/qa/unittests/tests/test_scenarios.py b/qa/unittests/tests/test_scenarios.py index c36477263..61dd65d66 100644 --- a/qa/unittests/tests/test_scenarios.py +++ b/qa/unittests/tests/test_scenarios.py @@ -2,12 +2,11 @@ import jsonschema import pytest -from apex_algorithm_qa_tools.scenarios import ( - BenchmarkScenario, - get_benchmark_scenarios, - lint_benchmark_scenario, -) +from apex_algorithm_qa_tools.scenarios.common import get_benchmark_scenarios, lint_benchmark_scenario +from apex_algorithm_qa_tools.scenarios.factory import _benchmark_factory +from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario +from apex_algorithm_qa_tools.scenarios.openeo import openEOBenchmarkScenario def test_get_benchmark_scenarios(): scenarios = get_benchmark_scenarios() @@ -31,10 +30,11 @@ def test_lint_scenario(scenario: BenchmarkScenario): assert isinstance(scenario.source, Path) and scenario.source.exists() -class TestBenchmarkScenario: - def test_init_minimal(self): - bs = BenchmarkScenario( +class TestOpenEOBenchmarkScenario: + def test_openeo_init_minimal(self): + bs = openEOBenchmarkScenario( id="foo", + type="openeo", backend="openeo.test", process_graph={}, ) @@ -46,8 +46,8 @@ def test_init_minimal(self): assert bs.reference_data == {} assert bs.reference_options == {} - def test_validation_minimal(self): - bs = BenchmarkScenario.from_dict( + def test_openeo_validation_minimal(self): + bs = openEOBenchmarkScenario.from_dict( { "id": "foo", "type": "openeo", @@ -65,4 +65,40 @@ def test_validation_minimal(self): def test_validation_missing_essentials(self): with pytest.raises(jsonschema.ValidationError): - BenchmarkScenario.from_dict({}) + openEOBenchmarkScenario.from_dict({}) + + +class TestBenchmarkFactory: + def test_openeo_factory(self): + path = Path("dummy_scenarios.json") + scenario = _benchmark_factory( + item={ + "id": "foo", + "type": "openeo", + "backend": "openeo.test", + "process_graph": {}, + }, + path=path, + ) + + assert isinstance(scenario, openEOBenchmarkScenario) + assert scenario.id == "foo" + assert scenario.type == "openeo" + assert scenario.source == path + + def test_factory_missing_type(self): + with pytest.raises( + AssertionError, + match="Missing required 'type' field in benchmark scenario", + ): + _benchmark_factory(item={"type": None}, path=Path("dummy.json")) + + def test_factory_unsupported_type(self): + with pytest.raises( + ValueError, + match="Unsupported benchmark scenario type: ogcapi-processes", + ): + _benchmark_factory( + item={"type": "ogcapi-processes"}, + path=Path("dummy.json"), + ) diff --git a/schemas/benchmark_scenario.json b/schemas/benchmark_scenario.json index 832c694e1..5ac759e37 100644 --- a/schemas/benchmark_scenario.json +++ b/schemas/benchmark_scenario.json @@ -1,48 +1,76 @@ { "$schema": "https://json-schema.org/draft/2020-12/schema", "title": "Benchmark Scenario", - "type": "object", - "properties": { - "id": { - "type": "string", - "description": "The unique identifier for the benchmark scenario." - }, - "type": { - "type": "string", - "description": "The type of the benchmark scenario.", - "enum": [ - "openeo" - ] - }, - "description": { - "type": "string", - "description": "A description of the benchmark scenario." - }, - "backend": { - "type": "string", - "description": "The openEO backend URL to connect to." - }, - "process_graph": { - "type": "object", - "description": "The openEO process graph to execute." - }, - "job_options": { + "$defs": { + "common": { "type": "object", - "description": "Batch job options to use when creating an openEO batch job." - }, - "reference_data": { - "type": "object", - "description": "Reference data of the benchmark, to compare actual results with." + "properties": { + "id": { + "type": "string", + "description": "The unique identifier for the benchmark scenario." + }, + "type": { + "type": "string", + "description": "The type of the benchmark scenario.", + "enum": [ + "openeo", + "ogcapi-processes" + ] + }, + "description": { + "type": "string", + "description": "A description of the benchmark scenario." + }, + "reference_data": { + "type": "object", + "description": "Reference data of the benchmark, to compare actual results with." + }, + "reference_options": { + "type": "object", + "description": "Options to fine-tune how actual and reference results should be compared." + } + }, + "required": [ + "id", + "type", + "backend" + ] }, - "reference_options": { + "openeo": { "type": "object", - "description": "Options to fine-tune how actual and reference results should be compared." + "properties": { + "type": { + "const": "openeo" + }, + "backend": { + "type": "string", + "description": "The backend URL to connect to." + }, + "process_graph": { + "type": "object", + "description": "The openEO process graph to execute." + }, + "job_options": { + "type": "object", + "description": "Batch job options to use when creating an openEO batch job." + } + }, + "required": [ + "process_graph" + ] } }, - "required": [ - "id", - "type", - "backend", - "process_graph" - ] -} + "oneOf": [ + { + "allOf": [ + { + "$ref": "#/$defs/common" + }, + { + "$ref": "#/$defs/openeo" + } + ] + } + ], + "unevaluatedProperties": false +} \ No newline at end of file From 797b1192db8e43e44183fdf3ba37346af611655f Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Fri, 29 May 2026 16:56:23 +0200 Subject: [PATCH 2/6] test: fixed ref in tests --- qa/unittests/tests/test_github_issue_handler.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/qa/unittests/tests/test_github_issue_handler.py b/qa/unittests/tests/test_github_issue_handler.py index cd74ea698..0cfb72187 100644 --- a/qa/unittests/tests/test_github_issue_handler.py +++ b/qa/unittests/tests/test_github_issue_handler.py @@ -12,7 +12,9 @@ TerminalReportSection, TestMetricsData, ) -from apex_algorithm_qa_tools.scenarios import BenchmarkScenario + +from apex_algorithm_qa_tools.scenarios.factory import read_scenarios_file +from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario class TestGithubApi: @@ -323,7 +325,7 @@ def benchmark_scenario(self, test_data_root) -> BenchmarkScenario: / "benchmark_scenarios" / "add3x.json" ) - return BenchmarkScenario.read_scenarios_file(path)[0] + return read_scenarios_file(path)[0] @pytest.fixture def metrics_data(self) -> TestMetricsData: From 0affc9f30f14857b5037cc5f2c716041beda476b Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Fri, 29 May 2026 17:13:11 +0200 Subject: [PATCH 3/6] fix: fixed refs in code --- qa/tools/apex_algorithm_qa_tools/github_issue_handler.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/qa/tools/apex_algorithm_qa_tools/github_issue_handler.py b/qa/tools/apex_algorithm_qa_tools/github_issue_handler.py index 4e1a01f52..4058b0722 100644 --- a/qa/tools/apex_algorithm_qa_tools/github_issue_handler.py +++ b/qa/tools/apex_algorithm_qa_tools/github_issue_handler.py @@ -12,11 +12,10 @@ from typing import Any, Dict, List, Optional, Union import requests -from apex_algorithm_qa_tools.scenarios import ( - BenchmarkScenario, - get_benchmark_scenarios, - get_project_root, -) + +from apex_algorithm_qa_tools.common import get_project_root +from apex_algorithm_qa_tools.scenarios.common import get_benchmark_scenarios +from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario logger = logging.getLogger(__name__) From 644ce796abe3c57a60e2be1878eb9ac3863b38cf Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Tue, 2 Jun 2026 12:12:31 +0200 Subject: [PATCH 4/6] feat: integrated ogc benchmarks --- .../benchmarks/auth.py | 155 ++++++ .../benchmarks/common.py | 41 +- .../apex_algorithm_qa_tools/benchmarks/ogc.py | 455 +++++++++++------- .../benchmarks/openeo.py | 9 +- .../benchmarks/runners/base.py | 31 +- .../benchmarks/runners/factory.py | 12 +- .../benchmarks/runners/ogc.py | 143 +++--- .../benchmarks/runners/openeo.py | 7 +- .../scenarios/__init__.py | 13 + .../scenarios/common.py | 17 +- .../scenarios/factory.py | 3 + .../apex_algorithm_qa_tools/scenarios/ogc.py | 63 +++ qa/tools/pyproject.toml | 1 + schemas/benchmark_scenario.json | 89 +++- 14 files changed, 755 insertions(+), 284 deletions(-) create mode 100644 qa/tools/apex_algorithm_qa_tools/benchmarks/auth.py create mode 100644 qa/tools/apex_algorithm_qa_tools/scenarios/ogc.py diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/auth.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/auth.py new file mode 100644 index 000000000..d633baffd --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/auth.py @@ -0,0 +1,155 @@ +import time + +import requests + + +def _discover_oidc_endpoints(keycloak_url: str, realm: str) -> tuple[str, list[str]]: + """Resolve token and device authorization endpoints from OIDC discovery. + + Falls back to well-known Keycloak endpoint patterns when discovery is unavailable. + """ + + base = keycloak_url.rstrip("/") + fallback_token = f"{base}/realms/{realm}/protocol/openid-connect/token" + fallback_devices = [ + f"{base}/realms/{realm}/protocol/openid-connect/auth/device", + f"{base}/realms/{realm}/protocol/openid-connect/device/auth", + ] + + discovery_url = f"{base}/realms/{realm}/.well-known/openid-configuration" + try: + response = requests.get(discovery_url, timeout=30) + response.raise_for_status() + discovery = response.json() + token_url = discovery.get("token_endpoint") or fallback_token + device_url = discovery.get("device_authorization_endpoint") + device_urls = [device_url] if device_url else [] + device_urls.extend(fallback_devices) + # Keep ordering stable while avoiding duplicates. + device_urls = list(dict.fromkeys(device_urls)) + return token_url, device_urls + except Exception: + return fallback_token, fallback_devices + + +def _safe_json(response: requests.Response) -> dict: + try: + data = response.json() + return data if isinstance(data, dict) else {} + except ValueError: + return {} + + +def get_token_with_client_credentials( + keycloak_url: str, + realm: str, + client_id: str, + client_secret: str, +) -> str: + """Retrieve an access token from a Keycloak instance using client credentials (client ID + secret). + + :param keycloak_url: Base URL of the Keycloak instance (e.g. https://keycloak.example.com). + :param realm: Keycloak realm name. + :param client_id: Client ID registered in Keycloak. + :param client_secret: Client secret associated with the client ID. + :return: Access token string. + """ + token_url = f"{keycloak_url.rstrip('/')}/realms/{realm}/protocol/openid-connect/token" + response = requests.post( + token_url, + data={ + "grant_type": "client_credentials", + "client_id": client_id, + "client_secret": client_secret, + }, + timeout=30, + ) + response.raise_for_status() + return response.json()["access_token"] + + +def get_token_with_device_flow( + keycloak_url: str, + realm: str, + client_id: str, + poll_interval: int = 5, + timeout: int = 300, +) -> str: + """Retrieve an access token from a Keycloak instance using the OAuth2 device authorization flow. + + This will print a user code and verification URL to the console, then poll until + the user completes authentication or the timeout is reached. + + :param keycloak_url: Base URL of the Keycloak instance (e.g. https://keycloak.example.com). + :param realm: Keycloak realm name. + :param client_id: Client ID registered in Keycloak. + :param poll_interval: Seconds to wait between polling attempts (default: 5). + :param timeout: Maximum seconds to wait for the user to authenticate (default: 300). + :return: Access token string. + """ + token_url, device_urls = _discover_oidc_endpoints(keycloak_url, realm) + + device_response = None + for device_url in device_urls: + candidate = requests.post(device_url, data={"client_id": client_id}, timeout=30) + if candidate.status_code in {404, 405}: + continue + device_response = candidate + break + + if device_response is None: + raise RuntimeError(f"Could not find a working device authorization endpoint. Tried: {device_urls!r}") + + if device_response.status_code == 401: + error_payload = _safe_json(device_response) + error = error_payload.get("error") + error_description = error_payload.get("error_description") + raise RuntimeError( + "Device-flow authorization request was rejected (401). " + "This usually means client or realm configuration mismatch. " + "Verify: 1) realm is correct, 2) client_id exists in that realm, " + "3) client allows device authorization grant, 4) client is public " + "or otherwise configured for device flow. " + f"endpoint={device_response.url!r} error={error!r} " + f"error_description={error_description!r}" + ) + + device_response.raise_for_status() + device_data = device_response.json() + + device_code = device_data["device_code"] + user_code = device_data["user_code"] + verification_uri = device_data.get("verification_uri_complete") or device_data["verification_uri"] + interval = device_data.get("interval", poll_interval) + + print(f"To authenticate, visit: {verification_uri}") + print(f"And enter the code: {user_code}") + + elapsed = 0 + while elapsed < timeout: + time.sleep(interval) + elapsed += interval + + token_response = requests.post( + token_url, + data={ + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + "client_id": client_id, + "device_code": device_code, + }, + timeout=30, + ) + + if token_response.status_code == 200: + return token_response.json()["access_token"] + + error_data = _safe_json(token_response) + error = error_data.get("error") + if error == "authorization_pending": + continue + elif error == "slow_down": + interval += 5 + else: + token_response.raise_for_status() + + raise TimeoutError(f"Device flow authentication timed out after {timeout} seconds.") diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/common.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/common.py index e7ee8a4c1..9fcee450a 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/common.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/common.py @@ -1,14 +1,15 @@ from __future__ import annotations import dataclasses -import logging from pathlib import Path from typing import Any from typing import Union +from urllib.parse import urlparse from apex_algorithm_qa_tools.pytest.pytest_track_metrics import MetricsTracker +from apex_algorithm_qa_tools.benchmarks.runners.base import BenchmarkJobMetadata, BenchmarkResults +import requests -_log = logging.getLogger(__name__) @dataclasses.dataclass class BenchmarkExecutionArtifacts: @@ -41,24 +42,19 @@ def to_jsonable(value: Any) -> Any: def collect_metrics_from_job_metadata( - job_metadata: dict, + job_metadata: BenchmarkJobMetadata, track_metric: MetricsTracker, ): - track_metric("costs", job_metadata.get("costs")) - for usage_metric, usage_data in job_metadata.get("usage", {}).items(): - if "unit" in usage_data and "value" in usage_data: - track_metric( - f"usage:{usage_metric}:{usage_data['unit']}", usage_data["value"] - ) + track_metric("costs", job_metadata.cost) + for usage_metric in job_metadata.usage or []: + if usage_metric.unit and usage_metric.value is not None: + track_metric(f"usage:{usage_metric.name}:{usage_metric.unit}", usage_metric.value) -def collect_metrics_from_results_metadata( - results_metadata: dict, - track_metric: MetricsTracker -): +def collect_metrics_from_results_metadata(results_metadata: BenchmarkResults, track_metric: MetricsTracker): proj_shape_area_mpx = [] proj_shape_area_km2 = [] - for asset_name, asset_data in results_metadata.get("assets", {}).items(): + for asset_name, asset_data in results_metadata.assets.items(): if "proj:shape" in asset_data: y, x = asset_data["proj:shape"][:2] proj_shape_area_mpx.append(y * x / 1e6) @@ -80,4 +76,19 @@ def collect_metrics_from_results_metadata( def analyse_results_comparison_exception(exc: Exception) -> Union[str, None]: if isinstance(exc, AssertionError): if "Differing 'derived_from' links" in str(exc): - return "derived_from-change" \ No newline at end of file + return "derived_from-change" + + +def download_file(href: str, target: Path, user_token: str | None = None): + parsed = urlparse(href) + target.parent.mkdir(parents=True, exist_ok=True) + + if parsed.scheme in {"http", "https"}: + headers = {"Authorization": f"Bearer {user_token}"} if user_token else None + with requests.get(href, headers=headers, stream=True) as response: + response.raise_for_status() + with target.open("wb") as fh: + for chunk in response.iter_content(chunk_size=1024 * 64): + fh.write(chunk) + else: + raise ValueError(f"Unsupported URL scheme in href: {href!r}") diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py index aff1be1ef..7510a1455 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py @@ -1,165 +1,290 @@ -# from __future__ import annotations - -# import importlib -# import json -# import time -# from pathlib import Path -# from typing import Any - -# import requests -# from apex_algorithm_qa_tools.benchmark_execution_shared import ( -# ensure_safe_relative_target, -# to_jsonable, -# ) - - -# def load_ogc_api_components() -> dict[str, Any]: -# try: -# package = importlib.import_module("ogc_api_processes_client") -# execute_api = importlib.import_module( -# "ogc_api_processes_client.api.execute_api" -# ) -# status_api = importlib.import_module( -# "ogc_api_processes_client.api.status_api" -# ) -# result_api = importlib.import_module( -# "ogc_api_processes_client.api.result_api" -# ) -# execute_model = importlib.import_module( -# "ogc_api_processes_client.models.execute" -# ) -# except ModuleNotFoundError as e: -# raise RuntimeError( -# "OGC API processes benchmark requires dependency " -# "'ogc-api-processes-client'" -# ) from e - -# return { -# "ApiClient": package.ApiClient, -# "Configuration": package.Configuration, -# "ExecuteApi": execute_api.ExecuteApi, -# "StatusApi": status_api.StatusApi, -# "ResultApi": result_api.ResultApi, -# "Execute": execute_model.Execute, -# } - - -# def create_ogc_api_client( -# *, -# backend: str, -# request_headers: dict | None, -# ): -# components = load_ogc_api_components() -# configuration = components["Configuration"](host=backend) -# api_client = components["ApiClient"](configuration=configuration) - -# if request_headers: -# api_client.default_headers.update(request_headers) - -# return components, api_client - - -# def execute_ogc_process(*, scenario, components: dict[str, Any], api_client): -# execute_api = components["ExecuteApi"](api_client) -# execute_request = {"inputs": scenario.inputs or {}} -# if scenario.outputs is not None: -# execute_request["outputs"] = scenario.outputs -# if scenario.response is not None: -# execute_request["response"] = scenario.response - -# execute_response = execute_api.execute_with_http_info( -# process_id=scenario.process_id, -# execute=components["Execute"].from_dict(execute_request), -# ) - -# if execute_response.status_code == 201: -# status_info = execute_response.data -# return status_info.job_id, status_info, None -# if execute_response.status_code == 200: -# return None, None, execute_response.data - -# raise RuntimeError( -# "Unexpected HTTP status from OGC execute call: " -# f"{execute_response.status_code}" -# ) - - -# def wait_for_ogc_job(*, components: dict[str, Any], api_client, job_id: str, max_minutes: int | None): -# status_api = components["StatusApi"](api_client) -# deadline = None -# if max_minutes: -# deadline = time.monotonic() + max_minutes * 60 - -# while True: -# status_info = status_api.get_status(job_id=job_id) -# status = str(status_info.status) - -# if status == "successful": -# return status_info -# if status in {"failed", "dismissed"}: -# raise RuntimeError( -# "OGC API process job " -# f"{job_id} ended in status {status!r}: " -# f"{status_info.message!r}" -# ) -# if deadline is not None and time.monotonic() >= deadline: -# raise TimeoutError( -# "OGC API process job " -# f"{job_id} exceeded maximum allowed time " -# f"of {max_minutes} minutes" -# ) - -# time.sleep(5) - - -# def get_ogc_results(*, components: dict[str, Any], api_client, job_id: str): -# result_api = components["ResultApi"](api_client) -# return result_api.get_result(job_id=job_id) - - -# def serialize_ogc_results(results: dict[str, Any]) -> dict[str, Any]: -# return {name: to_jsonable(item) for name, item in results.items()} - - -# def _extract_result_href(item: Any) -> str | None: -# value = to_jsonable(item) -# if not isinstance(value, dict): -# return None -# if isinstance(value.get("href"), str): -# return value["href"] -# nested = value.get("value") -# if isinstance(nested, dict) and isinstance(nested.get("href"), str): -# return nested["href"] -# return None - - -# def download_ogc_results( -# *, -# results: dict[str, Any], -# actual_dir: Path, -# request_headers: dict | None, -# ): -# actual_dir.mkdir(parents=True, exist_ok=True) -# serialized_results = serialize_ogc_results(results) - -# paths = [] -# metadata_path = actual_dir / "job-results.json" -# metadata_path.write_text(json.dumps(serialized_results, indent=2)) -# paths.append(metadata_path) - -# for name, item in results.items(): -# href = _extract_result_href(item) -# if not href: -# continue - -# target = ensure_safe_relative_target(actual_dir, name) -# target.parent.mkdir(parents=True, exist_ok=True) - -# response = requests.get(href, stream=True, headers=request_headers) -# response.raise_for_status() -# with target.open("wb") as f: -# for chunk in response.iter_content(chunk_size=128 * 1024): -# f.write(chunk) -# paths.append(target) - -# return paths +from __future__ import annotations + +import dataclasses +import json +import logging +import os +import re +import time +import urllib +from pathlib import Path +from urllib.parse import urlparse + +from ogc_api_processes_client.api.result_api import ResultApi +from ogc_api_processes_client.api_client_wrapper import ApiClientWrapper +from ogc_api_processes_client.configuration import Configuration +from ogc_api_processes_client.models import StatusCode +from ogc_api_processes_client.models.link import Link as OgcLink +from ogc_api_processes_client.models.input_value_no_object import InputValueNoObject + +from stac_pydantic.collection import Collection + +from apex_algorithm_qa_tools.benchmarks.auth import get_token_with_client_credentials, get_token_with_device_flow +from apex_algorithm_qa_tools.benchmarks.common import ( + BenchmarkJobMetadata, + download_file, + ensure_safe_relative_target, + to_jsonable, +) +from apex_algorithm_qa_tools.benchmarks.runners.base import BenchmarkResults +from apex_algorithm_qa_tools.scenarios.ogc import OGCAPIBenchmarkScenario, OGCAPIResults + +from httpx import get as http_get, Response as HTTPXResponse + + +_log = logging.getLogger(__name__) + +_TERMINAL_JOB_STATUSES = {StatusCode.SUCCESSFUL, StatusCode.FAILED, StatusCode.DISMISSED} + +STAC_COLLECTION_SCHEMA = "https://schemas.stacspec.org/v1.0.0/collection-spec/json-schema/collection.json" + +GEOJSON_FEATURECOLLECTION_SCHEMA = ( + "https://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/schemas/featureCollectionGeoJSON.yaml" +) + + +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 { + "processing.geohazards-tep.eu", + }: + return "OGC_AUTH_CLIENT_CREDENTIALS_GEOHAZARDS_TEP" + else: + raise ValueError(f"Unsupported backend: {url=} ({hostname=})") + + +def get_auth_token(endpoint: str, keycloak_url: str, keycloak_realm: str) -> str: + auth_env_var = get_client_credentials_env_var(endpoint) + if auth_env_var in os.environ: + client_credentials = os.environ[auth_env_var] + client_id, sep, client_secret = client_credentials.partition("/") + if not sep: + raise ValueError( + f"Invalid client credentials format in env var {auth_env_var!r}. " + "Expected 'client_id/client_secret' or 'client_id/'." + ) + + if client_id and client_secret: + return get_token_with_client_credentials( + keycloak_url=keycloak_url, + realm=keycloak_realm, + client_id=client_id, + client_secret=client_secret, + ) + elif client_id and not client_secret: + return get_token_with_device_flow( + keycloak_url=keycloak_url, + realm=keycloak_realm, + client_id=client_id, + ) + else: + raise ValueError( + f"Invalid client credentials format in env var {auth_env_var!r}. " + f"Expected 'client_id/client_secret' or 'client_id/'. Got: {client_credentials!r}" + ) + else: + raise ValueError( + f"No authentication credentials found for endpoint {endpoint!r}. " + "Please set the environment variable " + f"{auth_env_var!r} with value 'client_id/client_secret' or 'client_id/'." + ) + + +def create_ogc_api_client(*, endpoint: str, namespace: str, user_token: str) -> ApiClientWrapper: + config = Configuration(host=f"{endpoint}/{namespace}" if namespace else endpoint) + + additional_args = {} + additional_args["header_name"] = "Authorization" + additional_args["header_value"] = f"Bearer {user_token}" + + return ApiClientWrapper(configuration=config, **additional_args) + + +def create_ogc_job(*, scenario) -> dict: + return {"inputs": {key: value for key, value in scenario.parameters.items()}} + + +def _get_job_headers(user_token: str) -> dict: + headers = { + "accept": "*/*", + # "Prefer": "respond-async;return=representation", + "Content-Type": "application/json", + } + + headers["Authorization"] = f"Bearer {user_token}" + return headers + + +def _split_job_id(job_id: str) -> tuple[str, ...]: + parts = job_id.split(":", 1) + if len(parts) != 2: + return ("", job_id) + return tuple(parts) + + +def run_ogc_job( + *, + api_client: ApiClientWrapper, + scenario: OGCAPIBenchmarkScenario, + user_token: str, + job: dict, + max_minutes: int | None, +) -> str: + headers = _get_job_headers(user_token=user_token) + deadline = time.monotonic() + max_minutes * 60 if max_minutes else None + + # Execute the job + content = api_client.execute_simple(process_id=scenario.application, execute=job, _headers=headers) + namespace, job_id = _split_job_id(job_id=content.job_id) + + # Check the status + status = StatusCode.ACCEPTED + while status not in _TERMINAL_JOB_STATUSES: + if deadline is not None and time.monotonic() > deadline: + raise TimeoutError(f"Batch job {job_id} exceeded maximum allowed time of {max_minutes} minutes") + time.sleep(5) + status = api_client.get_status(job_id=job_id).status + _log.info(f"Job {job_id} is still running with status {status}...") + return job_id + + +def collect_ogc_job_metadata(*, api_client: ApiClientWrapper, job_id: str) -> BenchmarkJobMetadata: + # status_api = StatusApi(api_client.api_client) + # status = status_api.get_status(job_id=job_id) + # Currently the status API is not returning any relevant metrics + return BenchmarkJobMetadata(cost=None, usage=[]) + + +def collect_ogc_results(*, api_client: ApiClientWrapper, job_id: str, user_token: str) -> BenchmarkResults: + result_api = ResultApi(api_client.api_client) + results = result_api.get_result(job_id=job_id) + assets = {} + for result_name, result_value in results.items(): + if not result_value.actual_instance: + _log.debug(f"Ignoring result '{result_name}' with None value") + continue + + if isinstance(result_value.actual_instance, InputValueNoObject) or isinstance( + result_value.actual_instance, OgcLink + ): + _log.debug(f"Ignoring result '{result_name}' of unmanaged type {type(result_value)}") + continue + + qualified_value = result_value.actual_instance + if qualified_value.var_schema and qualified_value.var_schema.actual_instance: + schema_reference = qualified_value.var_schema.actual_instance + _log.debug( + f"Processing result\n* Name: '{result_name}'\n" + f"* media type: {qualified_value.media_type}\n" + f"* Python type: {type(qualified_value.value)}\n" + "* schema {qualified_value.var_schema}..." + ) + + if not isinstance(schema_reference, str): + _log.warning( + f"Processing result name: '{result_name}' can not be processed, " + f"schema of type {type(schema_reference)} not recognized" + ) + continue + + if STAC_COLLECTION_SCHEMA == schema_reference: + _log.info(f"STAC Collection found in results: '{result_name}'") + collection = Collection.model_validate(qualified_value.value.actual_instance) + _log.debug(f"Extracted collection '{collection.id}' with assets: {list(collection.assets.keys())}") + assets.update(collection.assets.to_dict()) + elif GEOJSON_FEATURECOLLECTION_SCHEMA == schema_reference: + _log.info(f"GeoJSON FeatureCollection found in results: '{result_name}'") + feature_collection = qualified_value.value.oneof_schema_2_validator or {} + for feature in feature_collection.get("features", []): + for link in feature.get("links", []): + if "collection" == link.get("rel") and link.get("href"): + collection_link: str = link.get("href") + _log.debug( + f"GeoJSON FeatureCollection results: '{result_name}' " + "points to a valid collection URL: {collection_link}" + ) + + response: HTTPXResponse = http_get( + collection_link, + follow_redirects=True, + headers={"Authorization": f"Bearer {user_token}"}, + ) + response.raise_for_status() + collection = Collection.model_validate(response.json()) + _log.debug( + f"Extracted collection '{collection.id}' with assets: {list(collection.assets.keys())}" + ) + assets.update(collection.to_dict().get("assets", {})) + else: + _log.warning( + f"Processing result: '{result_name}' can not be processed, " + f"schema {schema_reference} not yet managed" + ) + _log.debug(f"Collected {len(assets)} assets from OGC API results: {list(assets.keys())}") + return BenchmarkResults(assets=assets) + + +def download_ogc_results( + *, + results_metadata: BenchmarkResults, + actual_dir: Path, + user_token: str | None = None, + details: OGCAPIResults | None = None, +) -> list[Path]: + actual_dir.mkdir(parents=True, exist_ok=True) + paths: list[Path] = [] + + results_path = actual_dir / "job-results.json" + results_path.write_text(json.dumps(dataclasses.asdict(results_metadata), indent=2), encoding="utf8") + paths.append(results_path) + + for output_name, output_data in sorted(results_metadata.assets.items()): + output_data = to_jsonable(output_data) + if isinstance(output_data, dict) and output_data.get("href"): + ref = output_data["href"] + file_name = _resolve_output_filename(output_name=output_name, href=ref) + target = ensure_safe_relative_target(actual_dir, file_name) + target.parent.mkdir(parents=True, exist_ok=True) + download_file( + _convert_s3_href_to_https(ref, s3_endpoint=details.s3_endpoint if details else None), + target, + user_token=user_token, + ) + paths.append(target) + else: + target = ensure_safe_relative_target(actual_dir, f"{output_name}.json") + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(json.dumps(output_data, indent=2), encoding="utf8") + paths.append(target) + + return paths + + +def _convert_s3_href_to_https(href: str, s3_endpoint: str | None) -> str: + parsed = urlparse(href) + if not s3_endpoint: + raise ValueError(f"Cannot convert s3 href to https without s3_endpoint: {href=}, {s3_endpoint=}") + if parsed.scheme != "s3": + return href + bucket = parsed.netloc + key = parsed.path.lstrip("/") + if not bucket or not key: + raise ValueError(f"Invalid s3 href: {href!r}") + url = f"https://{bucket}.{s3_endpoint}/{key}" + _log.debug(f"Converted s3 href to https: {href} -> {url}") + return url + + +def _resolve_output_filename(*, output_name: str, href: str) -> str: + parsed = urlparse(href) + name = Path(parsed.path).name + if name: + return name + return f"{output_name}.bin" diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py index 4acce69f0..50764266f 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py @@ -87,7 +87,6 @@ def create_openeo_connection(*, backend: str, origin: str | None = None) -> open return connection - def create_openeo_job(*, connection, scenario): return connection.create_job( process_graph=scenario.process_graph, @@ -100,11 +99,7 @@ def run_openeo_job(*, job, max_minutes: int | None): if max_minutes: def _timeout_handler(signum, frame): - raise TimeoutError( - "Batch job " - f"{job.job_id} exceeded maximum allowed time " - f"of {max_minutes} minutes" - ) + 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) @@ -121,4 +116,4 @@ def collect_openeo_metadata(*, job): def download_openeo_results(*, results, actual_dir: Path): - return results.download_files(target=actual_dir, include_stac_metadata=True) \ No newline at end of file + return results.download_files(target=actual_dir, include_stac_metadata=True) diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py index 4a3105936..e0ce4e697 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py @@ -7,21 +7,37 @@ from apex_algorithm_qa_tools.scenarios import BenchmarkScenario + +@dataclasses.dataclass +class BenchmarkResults: + assets: dict + + +@dataclasses.dataclass +class BenchmarkMetric: + name: str + value: Any + unit: str | None = None + + +@dataclasses.dataclass +class BenchmarkJobMetadata: + cost: float | None + usage: list[BenchmarkMetric] | None + + @dataclasses.dataclass class BenchmarkRunnerArtifacts: job_id: str | None - job_metadata: Any - results_metadata: Any + job_metadata: BenchmarkJobMetadata + results_metadata: BenchmarkResults class BenchmarkRunner(ABC): def __init__(self, *, scenario: BenchmarkScenario, request): self.scenario = scenario self.request = request - self.origin = ( - f"apex-algorithms/benchmarks/" - f"{request.session.name}/{request.node.name}" - ) + self.origin = f"apex-algorithms/benchmarks/{request.session.name}/{request.node.name}" @abstractmethod def create_job(self): @@ -38,6 +54,3 @@ def collect_artifacts(self) -> BenchmarkRunnerArtifacts: @abstractmethod def download_actual(self, *, actual_dir: Path) -> list[Path]: pass - - - diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py index 7c5b35a96..1a3647010 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py @@ -1,6 +1,5 @@ - - from apex_algorithm_qa_tools.benchmarks.runners.openeo import OpenEOBenchmarkRunner +from apex_algorithm_qa_tools.benchmarks.runners.ogc import OGCBenchmarkRunner def create_benchmark_runner(*, scenario, request): @@ -9,5 +8,10 @@ def create_benchmark_runner(*, scenario, request): scenario=scenario, request=request, ) - - raise ValueError(f"Unsupported benchmark scenario type {scenario.type!r}") + elif scenario.type == "ogc_api_process": + return OGCBenchmarkRunner( + scenario=scenario, + request=request, + ) + else: + raise ValueError(f"Unsupported benchmark scenario type {scenario.type!r}") diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py index 250dd7e30..0e8835c1e 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py @@ -1,82 +1,81 @@ -# from __future__ import annotations +from __future__ import annotations -# import dataclasses -# from abc import ABC, abstractmethod -# from pathlib import Path -# from typing import Any +from pathlib import Path -# from apex_algorithm_qa_tools.benchmark_execution_ogcapi import ( -# create_ogc_api_client, -# download_ogc_results, -# execute_ogc_process, -# get_ogc_results, -# serialize_ogc_results, -# wait_for_ogc_job, -# ) -# from apex_algorithm_qa_tools.benchmark_execution_openeo import ( -# collect_openeo_metadata, -# create_openeo_connection, -# create_openeo_job, -# download_openeo_results, -# run_openeo_job, -# ) +from apex_algorithm_qa_tools.benchmarks.ogc import ( + collect_ogc_results, + collect_ogc_job_metadata, + create_ogc_api_client, + create_ogc_job, + download_ogc_results, + get_auth_token, + run_ogc_job, +) +from apex_algorithm_qa_tools.benchmarks.runners.base import ( + BenchmarkRunner, + BenchmarkRunnerArtifacts, +) +from apex_algorithm_qa_tools.scenarios.ogc import OGCAPIBenchmarkScenario +class OGCBenchmarkRunner(BenchmarkRunner): + def __init__(self, *, scenario: OGCAPIBenchmarkScenario, request): + super().__init__(scenario=scenario, request=request) + self.endpoint = scenario.endpoint + self.namespace = scenario.namespace + self.application = scenario.application + self.result_details = scenario.results + self.user_token = get_auth_token( + endpoint=self.endpoint, keycloak_url=scenario.auth.url, keycloak_realm=scenario.auth.realm + ) + self._api_client = create_ogc_api_client( + user_token=self.user_token, + endpoint=self.endpoint, + namespace=self.namespace, + ) + self._job = None + self._results = None -# class OGCAPIBenchmarkRunner(BenchmarkRunner): -# def __init__(self, *, scenario, backend: str): -# super().__init__(scenario=scenario, backend=backend) -# self._components, self._api_client = create_ogc_api_client( -# backend=backend, -# request_headers=scenario.request_headers, -# ) -# self._job_id = None -# self._status_info = None -# self._results = None + def create_job(self): + self._job = create_ogc_job( + scenario=self.scenario, + ) -# def create_job(self): -# self._job_id, self._status_info, self._results = execute_ogc_process( -# scenario=self.scenario, -# components=self._components, -# api_client=self._api_client, -# ) + def run_job(self, *, max_minutes: int | None): + if self._job is None: + raise RuntimeError("Cannot run OGC API job before create_job().") + self._job_id = run_ogc_job( + api_client=self._api_client, + scenario=self.scenario, + user_token=self.user_token, + job=self._job, + max_minutes=max_minutes, + ) -# def run_job(self, *, max_minutes: int | None): -# if self._job_id is None: -# return + def collect_artifacts(self) -> BenchmarkRunnerArtifacts: + self._job_id = "opensartoolkit-v2-2-1-k6wp5" + if self._job_id is None: + raise RuntimeError("Cannot collect OGC API metadata before run_job().") -# self._status_info = wait_for_ogc_job( -# components=self._components, -# api_client=self._api_client, -# job_id=self._job_id, -# max_minutes=max_minutes, -# ) + self._results = collect_ogc_results( + api_client=self._api_client, job_id=self._job_id, user_token=self.user_token + ) + return BenchmarkRunnerArtifacts( + job_id=self._job_id, + job_metadata=collect_ogc_job_metadata( + api_client=self._api_client, + job_id=self._job_id, + ), + results_metadata=self._results, + ) -# def collect_artifacts(self) -> BenchmarkRunnerArtifacts: -# if self._results is None and self._job_id is not None: -# self._results = get_ogc_results( -# components=self._components, -# api_client=self._api_client, -# job_id=self._job_id, -# ) - -# if not isinstance(self._results, dict): -# raise RuntimeError("OGC API execution did not return a result mapping") - -# job_metadata = self._status_info.to_dict() if self._status_info else {} -# return BenchmarkRunnerArtifacts( -# job_id=self._job_id, -# job_metadata=job_metadata, -# results_metadata=serialize_ogc_results(self._results), -# ) - -# def download_actual(self, *, actual_dir: Path) -> list[Path]: -# if not isinstance(self._results, dict): -# raise RuntimeError("OGC API execution did not return a result mapping") - -# return download_ogc_results( -# results=self._results, -# actual_dir=actual_dir, -# request_headers=self._api_client.default_headers, -# ) \ No newline at end of file + def download_actual(self, *, actual_dir: Path) -> list[Path]: + if self._results is None: + raise RuntimeError("Cannot download OGC API results before collect_artifacts().") + return download_ogc_results( + results_metadata=self._results, + actual_dir=actual_dir, + user_token=self.user_token, + details=self.result_details, + ) diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py index ac48f88c2..f1e9502e7 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py @@ -1,6 +1,5 @@ from __future__ import annotations -import logging from pathlib import Path from apex_algorithm_qa_tools.benchmarks.openeo import ( @@ -16,13 +15,11 @@ BenchmarkRunnerArtifacts, ) -from apex_algorithm_qa_tools.scenarios import BenchmarkScenario - -_log = logging.getLogger(__name__) +from apex_algorithm_qa_tools.scenarios.openeo import openEOBenchmarkScenario class OpenEOBenchmarkRunner(BenchmarkRunner): - def __init__(self, *, scenario: BenchmarkScenario, request): + def __init__(self, *, scenario: openEOBenchmarkScenario, request): super().__init__(scenario=scenario, request=request) self.backend = get_openeo_backend(scenario, request) self._connection = create_openeo_connection( diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/__init__.py b/qa/tools/apex_algorithm_qa_tools/scenarios/__init__.py index e69de29bb..3b7f6555d 100644 --- a/qa/tools/apex_algorithm_qa_tools/scenarios/__init__.py +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/__init__.py @@ -0,0 +1,13 @@ +from apex_algorithm_qa_tools.scenarios.common import ( + download_reference_data, + get_benchmark_scenarios, + lint_benchmark_scenario, +) +from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario + +__all__ = [ + "BenchmarkScenario", + "download_reference_data", + "get_benchmark_scenarios", + "lint_benchmark_scenario", +] diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/common.py b/qa/tools/apex_algorithm_qa_tools/scenarios/common.py index 6472507ff..d769029d0 100644 --- a/qa/tools/apex_algorithm_qa_tools/scenarios/common.py +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/common.py @@ -1,14 +1,11 @@ from __future__ import annotations -import dataclasses import glob -import json import logging import re from pathlib import Path from typing import List -import jsonschema import requests from apex_algorithm_qa_tools.common import ( get_project_root, @@ -16,9 +13,11 @@ from openeo.util import TimingLogger from apex_algorithm_qa_tools.scenarios.openeo import lint_openeo_fields +from apex_algorithm_qa_tools.scenarios.ogc import OGCAPIBenchmarkScenario from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario from apex_algorithm_qa_tools.scenarios.factory import read_scenarios_file +_log = logging.getLogger(__name__) def get_benchmark_scenarios(root=None) -> List[BenchmarkScenario]: @@ -51,10 +50,20 @@ def lint_benchmark_scenario(scenario: BenchmarkScenario): if scenario.type == "openeo": lint_openeo_fields(scenario=scenario) + elif scenario.type in {"ogc_api_process", "ogcapi-processes"}: + lint_ogc_fields(scenario=scenario) else: raise ValueError(f"Unsupported benchmark scenario type: {scenario.type!r}") +def lint_ogc_fields(scenario: OGCAPIBenchmarkScenario): + assert scenario.backend.startswith(("http://", "https://")), ( + f"Unsupported OGC endpoint: {scenario.backend!r}" + ) + assert scenario.process_id + assert isinstance(scenario.parameters, dict) + + def download_reference_data(scenario: BenchmarkScenario, reference_dir: Path) -> Path: with TimingLogger( title=f"Downloading reference data for {scenario.id=} to {reference_dir=}", @@ -73,4 +82,4 @@ def download_reference_data(scenario: BenchmarkScenario, reference_dir: Path) -> for chunk in resp.iter_content(chunk_size=128): f.write(chunk) - return reference_dir \ No newline at end of file + return reference_dir diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py b/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py index c7755432b..dafe1ae7c 100644 --- a/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py @@ -4,6 +4,7 @@ from apex_algorithm_qa_tools.scenarios.scenario import BenchmarkScenario from apex_algorithm_qa_tools.scenarios.openeo import openEOBenchmarkScenario +from apex_algorithm_qa_tools.scenarios.ogc import OGCAPIBenchmarkScenario def read_scenarios_file(path: str | Path) -> List[BenchmarkScenario]: @@ -23,5 +24,7 @@ def _benchmark_factory(item: dict, path: str | Path) -> BenchmarkScenario: assert item["type"] is not None, "Missing required 'type' field in benchmark scenario" if item["type"] == "openeo": return openEOBenchmarkScenario.from_dict(data=item, source=path) + elif item["type"] in {"ogc_api_process", "ogcapi-processes"}: + return OGCAPIBenchmarkScenario.from_dict(data=item, source=path) else: raise ValueError(f"Unsupported benchmark scenario type: {item['type']}") diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/ogc.py b/qa/tools/apex_algorithm_qa_tools/scenarios/ogc.py new file mode 100644 index 000000000..b8af06a9c --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/ogc.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import dataclasses +from pathlib import Path + +import jsonschema + +from apex_algorithm_qa_tools.scenarios.scenario import ( + BenchmarkScenario, + get_benchmark_scenario_schema, +) + + +@dataclasses.dataclass(kw_only=True) +class OGCAPIAuth: + url: str + realm: str + + +@dataclasses.dataclass(kw_only=True) +class OGCAPIResults: + s3_endpoint: str + + +@dataclasses.dataclass(kw_only=True) +class OGCAPIBenchmarkScenario(BenchmarkScenario): + endpoint: str + process_id: str + parameters: dict + namespace: str | None = None + application: str | None = None + auth: OGCAPIAuth + results: OGCAPIResults | None = None + + @classmethod + def from_dict( + cls, + data: dict, + source: str | Path | None = None, + ) -> BenchmarkScenario: + jsonschema.validate(instance=data, schema=get_benchmark_scenario_schema()) + + process_id = data.get("process_id") + application = data.get("application") + namespace = data.get("namespace") + auth = OGCAPIAuth(**data.get("auth", {})) + results = OGCAPIResults(**data.get("results", {})) if data.get("results") else None + + return cls( + id=data["id"], + type=data["type"], + description=data.get("description"), + endpoint=data["endpoint"], + process_id=process_id, + parameters=data.get("parameters", {}), + auth=auth, + namespace=namespace, + application=application, + reference_data=data.get("reference_data", {}), + reference_options=data.get("reference_options", {}), + source=source, + results=results, + ) diff --git a/qa/tools/pyproject.toml b/qa/tools/pyproject.toml index 92855ec6d..080f1228e 100644 --- a/qa/tools/pyproject.toml +++ b/qa/tools/pyproject.toml @@ -25,4 +25,5 @@ dependencies = [ # TODO: make some of these dependencies optional "boto3>=1.36.5", "pyarrow>=17.0.0", + "stac-pydantic" ] diff --git a/schemas/benchmark_scenario.json b/schemas/benchmark_scenario.json index 5ac759e37..94cbe745a 100644 --- a/schemas/benchmark_scenario.json +++ b/schemas/benchmark_scenario.json @@ -14,7 +14,8 @@ "description": "The type of the benchmark scenario.", "enum": [ "openeo", - "ogcapi-processes" + "ogcapi-processes", + "ogc_api_process" ] }, "description": { @@ -32,8 +33,7 @@ }, "required": [ "id", - "type", - "backend" + "type" ] }, "openeo": { @@ -56,8 +56,81 @@ } }, "required": [ + "backend", "process_graph" ] + }, + "ogcapi_process": { + "type": "object", + "properties": { + "type": { + "const": "ogc_api_process" + }, + "endpoint": { + "type": "string", + "description": "The OGC API Processes endpoint." + }, + "process_id": { + "type": "string", + "description": "The process identifier used at execution time." + }, + "namespace": { + "type": "string", + "description": "Optional namespace for process categorization." + }, + "application": { + "type": "string", + "description": "Optional process alias used by some backends." + }, + "auth": { + "type": "object", + "description": "Authentication details for accessing the OGC API endpoint.", + "properties": { + "url": { + "type": "string", + "description": "The URL of the authentication server." + }, + "realm": { + "type": "string", + "description": "The authentication realm." + } + }, + "required": [ + "url", + "realm" + ] + }, + "parameters": { + "type": "object", + "description": "Input parameters passed to the OGC API process." + }, + "results": { + "type": "object", + "description": "Details for accessing the results of the OGC API process.", + "properties": { + "s3_endpoint": { + "type": "string", + "description": "The S3 endpoint where results are stored." + } + } + } + }, + "required": [ + "endpoint", + "parameters" + ], + "anyOf": [ + { + "required": [ + "process_id" + ] + }, + { + "required": [ + "application" + ] + } + ] } }, "oneOf": [ @@ -70,6 +143,16 @@ "$ref": "#/$defs/openeo" } ] + }, + { + "allOf": [ + { + "$ref": "#/$defs/common" + }, + { + "$ref": "#/$defs/ogcapi_process" + } + ] } ], "unevaluatedProperties": false From d91597252bc18075264ad7daf92ae620371f7d09 Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Mon, 22 Jun 2026 16:18:26 +0200 Subject: [PATCH 5/6] feat: start of ogc implementation --- .../benchmark_scenarios/gep_api_ost.json | 64 +++++++ qa/benchmarks/data/job-results.json | 9 + qa/benchmarks/tests/test_benchmarks.py | 159 +++++++++--------- .../apex_algorithm_qa_tools/benchmarks/ogc.py | 126 ++++++++++---- .../benchmarks/runners/ogc.py | 3 + .../scenarios/common.py | 23 ++- 6 files changed, 272 insertions(+), 112 deletions(-) create mode 100644 algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json create mode 100644 qa/benchmarks/data/job-results.json diff --git a/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json b/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json new file mode 100644 index 000000000..2dd382cee --- /dev/null +++ b/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json @@ -0,0 +1,64 @@ +[ + { + "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 + ] + ] + ] + } + }, + "results": { + "s3_endpoint": "s3.waw3-1.cloudferro.com" + }, + "reference_data": { + "job-results.json": "file:///Users/bramjanssen/projects/apex/apex_algorithms/qa/benchmarks/data/job-results.json" + }, + "reference_options": { + "atol": 1 + } + } +] \ No newline at end of file diff --git a/qa/benchmarks/data/job-results.json b/qa/benchmarks/data/job-results.json new file mode 100644 index 000000000..fe22d5b91 --- /dev/null +++ b/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" + } + } +} diff --git a/qa/benchmarks/tests/test_benchmarks.py b/qa/benchmarks/tests/test_benchmarks.py index f059dd7f9..68082bea8 100644 --- a/qa/benchmarks/tests/test_benchmarks.py +++ b/qa/benchmarks/tests/test_benchmarks.py @@ -51,17 +51,26 @@ def test_run_benchmark( 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, - "process_id": scenario.process_id, "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): @@ -84,12 +93,12 @@ def _on_phase_exception(phase: str, exc: Exception): artifacts = None - with track_phase(phase="create-job"): - runner.create_job() + # with track_phase(phase="create-job"): + # runner.create_job() - with track_phase(phase="run-job"): - max_minutes = request.config.getoption("--maximum-job-time-in-minutes") - runner.run_job(max_minutes=max_minutes) + # with track_phase(phase="run-job"): + # max_minutes = request.config.getoption("--maximum-job-time-in-minutes") + # runner.run_job(max_minutes=max_minutes) with track_phase(phase="collect-metadata"): artifacts = runner.collect_artifacts() @@ -115,68 +124,68 @@ def _on_phase_exception(phase: str, exc: Exception): # Upload assets on failure upload_assets_on_fail(*paths) - # Pre-compute S3 URLs for actual files (used in error messages and benchmark reports) - actual_s3_urls = { - str(p.relative_to(actual_dir)): upload_assets_on_fail.get_url(p) - for p in sorted(actual_dir.rglob("*")) if p.is_file() - } - actual_s3_urls = {k: v for k, v in actual_s3_urls.items() if v is not None} - - with track_phase(phase="download-reference"): - reference_dir = download_reference_data( - scenario=scenario, reference_dir=tmp_path / "reference" - ) - - if report_path is not None: - report = json.loads(report_path.read_text()) - report["actual_files"] = { - str(p.relative_to(actual_dir)): f"{p.stat().st_size / 1024:.1f} kb" - for p in sorted(actual_dir.rglob("*")) if p.is_file() - } - ref_files = {} - for p in sorted(reference_dir.rglob("*")): - if not p.is_file(): - continue - rel = p.relative_to(reference_dir) - size_str = f"{p.stat().st_size / 1024:.1f} kb" - actual_counterpart = actual_dir / rel - if not actual_counterpart.exists(): - size_str += " (missing in actual)" - elif actual_counterpart.stat().st_size != p.stat().st_size: - size_str += f" (actual: {actual_counterpart.stat().st_size / 1024:.1f} kb)" - ref_files[str(rel)] = size_str - report["reference_files"] = ref_files - if actual_s3_urls: - report["actual_data"] = actual_s3_urls - report_path.write_text(json.dumps(report, indent=2)) - # Also write to CWD so the report is accessible on Jenkins workspace - cwd_report_dir = Path("benchmark_reports") - cwd_report_dir.mkdir(exist_ok=True) - (cwd_report_dir / f"{scenario.id}_benchmark_report.json").write_text( - json.dumps(report, indent=2) - ) - - with track_phase( - phase="compare", describe_exception=analyse_results_comparison_exception - ): - # Compare actual results with reference data - try: - assert_job_results_allclose( - actual=actual_dir, - expected=reference_dir, - tmp_path=tmp_path, - rtol=scenario.reference_options.get("rtol", 1e-3), - atol=scenario.reference_options.get("atol", 1), - pixel_tolerance=scenario.reference_options.get("pixel_tolerance", 1), - ) - except AssertionError as e: - msg = str(e) - if scenario.reference_data: - msg += "\n\nReference data URLs:" - for name, url in scenario.reference_data.items(): - msg += f"\n {name}: {url}" - if actual_s3_urls: - msg += "\n\nActual data S3 URLs (uploaded on failure):" - for name, url in actual_s3_urls.items(): - msg += f"\n {name}: {url}" - raise AssertionError(msg) from None + # # Pre-compute S3 URLs for actual files (used in error messages and benchmark reports) + # actual_s3_urls = { + # str(p.relative_to(actual_dir)): upload_assets_on_fail.get_url(p) + # for p in sorted(actual_dir.rglob("*")) if p.is_file() + # } + # actual_s3_urls = {k: v for k, v in actual_s3_urls.items() if v is not None} + + # with track_phase(phase="download-reference"): + # reference_dir = download_reference_data( + # scenario=scenario, reference_dir=tmp_path / "reference" + # ) + + # if report_path is not None: + # report = json.loads(report_path.read_text()) + # report["actual_files"] = { + # str(p.relative_to(actual_dir)): f"{p.stat().st_size / 1024:.1f} kb" + # for p in sorted(actual_dir.rglob("*")) if p.is_file() + # } + # ref_files = {} + # for p in sorted(reference_dir.rglob("*")): + # if not p.is_file(): + # continue + # rel = p.relative_to(reference_dir) + # size_str = f"{p.stat().st_size / 1024:.1f} kb" + # actual_counterpart = actual_dir / rel + # if not actual_counterpart.exists(): + # size_str += " (missing in actual)" + # elif actual_counterpart.stat().st_size != p.stat().st_size: + # size_str += f" (actual: {actual_counterpart.stat().st_size / 1024:.1f} kb)" + # ref_files[str(rel)] = size_str + # report["reference_files"] = ref_files + # if actual_s3_urls: + # report["actual_data"] = actual_s3_urls + # report_path.write_text(json.dumps(report, indent=2)) + # # Also write to CWD so the report is accessible on Jenkins workspace + # cwd_report_dir = Path("benchmark_reports") + # cwd_report_dir.mkdir(exist_ok=True) + # (cwd_report_dir / f"{scenario.id}_benchmark_report.json").write_text( + # json.dumps(report, indent=2) + # ) + + # with track_phase( + # phase="compare", describe_exception=analyse_results_comparison_exception + # ): + # # Compare actual results with reference data + # try: + # assert_job_results_allclose( + # actual=actual_dir, + # expected=reference_dir, + # tmp_path=tmp_path, + # rtol=scenario.reference_options.get("rtol", 1e-3), + # atol=scenario.reference_options.get("atol", 1), + # pixel_tolerance=scenario.reference_options.get("pixel_tolerance", 1), + # ) + # except AssertionError as e: + # msg = str(e) + # if scenario.reference_data: + # msg += "\n\nReference data URLs:" + # for name, url in scenario.reference_data.items(): + # msg += f"\n {name}: {url}" + # if actual_s3_urls: + # msg += "\n\nActual data S3 URLs (uploaded on failure):" + # for name, url in actual_s3_urls.items(): + # msg += f"\n {name}: {url}" + # raise AssertionError(msg) from None diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py index 7510a1455..a63be038f 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py @@ -61,16 +61,16 @@ def get_client_credentials_env_var(url: str) -> str: def get_auth_token(endpoint: str, keycloak_url: str, keycloak_realm: str) -> str: + # Support direct token via environment variable + if "OGC_AUTH_TOKEN" in os.environ: + token = os.environ["OGC_AUTH_TOKEN"] + _log.info("Using auth token from OGC_AUTH_TOKEN environment variable") + return token + auth_env_var = get_client_credentials_env_var(endpoint) if auth_env_var in os.environ: client_credentials = os.environ[auth_env_var] - client_id, sep, client_secret = client_credentials.partition("/") - if not sep: - raise ValueError( - f"Invalid client credentials format in env var {auth_env_var!r}. " - "Expected 'client_id/client_secret' or 'client_id/'." - ) - + client_id, _, client_secret = client_credentials.partition("/") if client_id and client_secret: return get_token_with_client_credentials( keycloak_url=keycloak_url, @@ -94,6 +94,7 @@ def get_auth_token(endpoint: str, keycloak_url: str, keycloak_realm: str) -> str f"No authentication credentials found for endpoint {endpoint!r}. " "Please set the environment variable " f"{auth_env_var!r} with value 'client_id/client_secret' or 'client_id/'." + "Or set OGC_AUTH_TOKEN with a valid JWT token." ) @@ -152,6 +153,14 @@ def run_ogc_job( time.sleep(5) status = api_client.get_status(job_id=job_id).status _log.info(f"Job {job_id} is still running with status {status}...") + + # Check for job failure + if status == StatusCode.FAILED: + raise RuntimeError(f"OGC API job {job_id} failed with status {status}") + elif status == StatusCode.DISMISSED: + raise RuntimeError(f"OGC API job {job_id} was dismissed with status {status}") + + _log.info(f"Job {job_id} completed successfully with status {status}") return job_id @@ -162,10 +171,67 @@ def collect_ogc_job_metadata(*, api_client: ApiClientWrapper, job_id: str) -> Be return BenchmarkJobMetadata(cost=None, usage=[]) +def _extract_assets_from_feature_collection(feature_collection: dict, *, result_name: str, user_token: str) -> dict: + assets: dict = {} + for feature in feature_collection.get("features", []): + feature_assets = feature.get("assets") + if isinstance(feature_assets, dict): + assets.update(feature_assets) + continue + + # Some providers expose assets through an item link instead of inlining them in the feature. + for link in feature.get("links", []): + if "collection" == link.get("rel") and link.get("href"): + collection_link: str = link.get("href") + _log.debug( + f"GeoJSON FeatureCollection results: '{result_name}' " + f"points to a valid collection URL: {collection_link}" + ) + + response: HTTPXResponse = http_get( + collection_link, + follow_redirects=True, + headers={"Authorization": f"Bearer {user_token}"}, + ) + response.raise_for_status() + collection = Collection.model_validate(response.json()) + _log.debug(f"Extracted collection '{collection.id}' with assets: {list(collection.assets.keys())}") + assets.update(collection.to_dict().get("assets", {})) + break + return assets + + +def _extract_qualified_value_payload(qualified_value) -> object | None: + value = qualified_value.value + + # Primary path for non-union values. + payload = to_jsonable(value) + if payload is not None: + return payload + + # Fallback for union models where actual_instance is None but oneof validators are populated. + for attr in ( + "oneof_schema_1_validator", + "oneof_schema_2_validator", + "oneof_schema_3_validator", + "oneof_schema_4_validator", + "oneof_schema_5_validator", + "oneof_schema_6_validator", + "oneof_schema_7_validator", + ): + if hasattr(value, attr): + candidate = getattr(value, attr) + if candidate is not None: + return candidate + + return None + + def collect_ogc_results(*, api_client: ApiClientWrapper, job_id: str, user_token: str) -> BenchmarkResults: result_api = ResultApi(api_client.api_client) results = result_api.get_result(job_id=job_id) assets = {} + _log.info(f"Collecting OGC API results for job {job_id} with {len(results)} outputs...") for result_name, result_value in results.items(): if not result_value.actual_instance: _log.debug(f"Ignoring result '{result_name}' with None value") @@ -196,32 +262,32 @@ def collect_ogc_results(*, api_client: ApiClientWrapper, job_id: str, user_token if STAC_COLLECTION_SCHEMA == schema_reference: _log.info(f"STAC Collection found in results: '{result_name}'") - collection = Collection.model_validate(qualified_value.value.actual_instance) + collection_payload = _extract_qualified_value_payload(qualified_value) + if not isinstance(collection_payload, dict): + _log.warning( + f"Processing result: '{result_name}' can not be processed, " + f"invalid STAC collection payload type {type(collection_payload)}" + ) + continue + collection = Collection.model_validate(collection_payload) _log.debug(f"Extracted collection '{collection.id}' with assets: {list(collection.assets.keys())}") assets.update(collection.assets.to_dict()) elif GEOJSON_FEATURECOLLECTION_SCHEMA == schema_reference: _log.info(f"GeoJSON FeatureCollection found in results: '{result_name}'") - feature_collection = qualified_value.value.oneof_schema_2_validator or {} - for feature in feature_collection.get("features", []): - for link in feature.get("links", []): - if "collection" == link.get("rel") and link.get("href"): - collection_link: str = link.get("href") - _log.debug( - f"GeoJSON FeatureCollection results: '{result_name}' " - "points to a valid collection URL: {collection_link}" - ) - - response: HTTPXResponse = http_get( - collection_link, - follow_redirects=True, - headers={"Authorization": f"Bearer {user_token}"}, - ) - response.raise_for_status() - collection = Collection.model_validate(response.json()) - _log.debug( - f"Extracted collection '{collection.id}' with assets: {list(collection.assets.keys())}" - ) - assets.update(collection.to_dict().get("assets", {})) + feature_collection = _extract_qualified_value_payload(qualified_value) + if not isinstance(feature_collection, dict): + _log.warning( + f"Processing result: '{result_name}' can not be processed, " + f"invalid GeoJSON payload type {type(feature_collection)}" + ) + continue + assets.update( + _extract_assets_from_feature_collection( + feature_collection, + result_name=result_name, + user_token=user_token, + ) + ) else: _log.warning( f"Processing result: '{result_name}' can not be processed, " @@ -241,12 +307,14 @@ def download_ogc_results( actual_dir.mkdir(parents=True, exist_ok=True) paths: list[Path] = [] + _log.info(f"Downloading {len(results_metadata.assets)} OGC API results to {actual_dir=}") results_path = actual_dir / "job-results.json" results_path.write_text(json.dumps(dataclasses.asdict(results_metadata), indent=2), encoding="utf8") paths.append(results_path) for output_name, output_data in sorted(results_metadata.assets.items()): output_data = to_jsonable(output_data) + _log.debug(f"Downloading OGC API result '{output_name}' to {actual_dir=}") if isinstance(output_data, dict) and output_data.get("href"): ref = output_data["href"] file_name = _resolve_output_filename(output_name=output_name, href=ref) diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py index 0e8835c1e..bbd8e829e 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging from pathlib import Path from apex_algorithm_qa_tools.benchmarks.ogc import ( @@ -35,6 +36,7 @@ def __init__(self, *, scenario: OGCAPIBenchmarkScenario, request): namespace=self.namespace, ) self._job = None + self._job_id = None self._results = None def create_job(self): @@ -54,6 +56,7 @@ def run_job(self, *, max_minutes: int | None): ) def collect_artifacts(self) -> BenchmarkRunnerArtifacts: + # TODO: remove after testing self._job_id = "opensartoolkit-v2-2-1-k6wp5" if self._job_id is None: raise RuntimeError("Cannot collect OGC API metadata before run_job().") diff --git a/qa/tools/apex_algorithm_qa_tools/scenarios/common.py b/qa/tools/apex_algorithm_qa_tools/scenarios/common.py index d769029d0..ed12a54a1 100644 --- a/qa/tools/apex_algorithm_qa_tools/scenarios/common.py +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/common.py @@ -57,10 +57,10 @@ def lint_benchmark_scenario(scenario: BenchmarkScenario): def lint_ogc_fields(scenario: OGCAPIBenchmarkScenario): - assert scenario.backend.startswith(("http://", "https://")), ( - f"Unsupported OGC endpoint: {scenario.backend!r}" + assert scenario.endpoint.startswith(("http://", "https://")), ( + f"Unsupported OGC endpoint: {scenario.endpoint!r}" ) - assert scenario.process_id + assert scenario.application or scenario.process_id assert isinstance(scenario.parameters, dict) @@ -76,10 +76,17 @@ def download_reference_data(scenario: BenchmarkScenario, reference_dir: Path) -> path.parent.mkdir(parents=True, exist_ok=True) with TimingLogger(title=f"Downloading {source=} to {path=}", logger=_log.info): - # TODO: support other sources than HTTP? - resp = requests.get(source, stream=True) - with path.open("wb") as f: - for chunk in resp.iter_content(chunk_size=128): - f.write(chunk) + # Handle file:// URLs (local files) + if source.startswith("file://"): + file_path = source[7:] # Remove "file://" prefix + with open(file_path, "rb") as src_file: + with path.open("wb") as f: + f.write(src_file.read()) + else: + # Handle HTTP(S) URLs + resp = requests.get(source, stream=True) + with path.open("wb") as f: + for chunk in resp.iter_content(chunk_size=128): + f.write(chunk) return reference_dir From b08120b6dd7755058658a62829fb84290338b32b Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Thu, 25 Jun 2026 12:09:08 +0200 Subject: [PATCH 6/6] ogc-benchmarks: ready for full e2e test --- .../benchmark_scenarios/gep_api_ost.json | 6 +- qa/benchmarks/tests/test_benchmarks.py | 141 +++++++++--------- .../apex_algorithm_qa_tools/benchmarks/ogc.py | 71 ++++++--- .../benchmarks/runners/ogc.py | 3 - 4 files changed, 121 insertions(+), 100 deletions(-) diff --git a/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json b/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json index 2dd382cee..3ddf925c8 100644 --- a/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json +++ b/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json @@ -51,11 +51,9 @@ ] } }, - "results": { - "s3_endpoint": "s3.waw3-1.cloudferro.com" - }, "reference_data": { - "job-results.json": "file:///Users/bramjanssen/projects/apex/apex_algorithms/qa/benchmarks/data/job-results.json" + "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 diff --git a/qa/benchmarks/tests/test_benchmarks.py b/qa/benchmarks/tests/test_benchmarks.py index 68082bea8..548472430 100644 --- a/qa/benchmarks/tests/test_benchmarks.py +++ b/qa/benchmarks/tests/test_benchmarks.py @@ -1,6 +1,5 @@ import json import logging -import re from pathlib import Path import pytest @@ -93,12 +92,12 @@ def _on_phase_exception(phase: str, exc: Exception): artifacts = None - # with track_phase(phase="create-job"): - # runner.create_job() + with track_phase(phase="create-job"): + runner.create_job() - # with track_phase(phase="run-job"): - # max_minutes = request.config.getoption("--maximum-job-time-in-minutes") - # runner.run_job(max_minutes=max_minutes) + with track_phase(phase="run-job"): + max_minutes = request.config.getoption("--maximum-job-time-in-minutes") + runner.run_job(max_minutes=max_minutes) with track_phase(phase="collect-metadata"): artifacts = runner.collect_artifacts() @@ -124,68 +123,68 @@ def _on_phase_exception(phase: str, exc: Exception): # Upload assets on failure upload_assets_on_fail(*paths) - # # Pre-compute S3 URLs for actual files (used in error messages and benchmark reports) - # actual_s3_urls = { - # str(p.relative_to(actual_dir)): upload_assets_on_fail.get_url(p) - # for p in sorted(actual_dir.rglob("*")) if p.is_file() - # } - # actual_s3_urls = {k: v for k, v in actual_s3_urls.items() if v is not None} - - # with track_phase(phase="download-reference"): - # reference_dir = download_reference_data( - # scenario=scenario, reference_dir=tmp_path / "reference" - # ) - - # if report_path is not None: - # report = json.loads(report_path.read_text()) - # report["actual_files"] = { - # str(p.relative_to(actual_dir)): f"{p.stat().st_size / 1024:.1f} kb" - # for p in sorted(actual_dir.rglob("*")) if p.is_file() - # } - # ref_files = {} - # for p in sorted(reference_dir.rglob("*")): - # if not p.is_file(): - # continue - # rel = p.relative_to(reference_dir) - # size_str = f"{p.stat().st_size / 1024:.1f} kb" - # actual_counterpart = actual_dir / rel - # if not actual_counterpart.exists(): - # size_str += " (missing in actual)" - # elif actual_counterpart.stat().st_size != p.stat().st_size: - # size_str += f" (actual: {actual_counterpart.stat().st_size / 1024:.1f} kb)" - # ref_files[str(rel)] = size_str - # report["reference_files"] = ref_files - # if actual_s3_urls: - # report["actual_data"] = actual_s3_urls - # report_path.write_text(json.dumps(report, indent=2)) - # # Also write to CWD so the report is accessible on Jenkins workspace - # cwd_report_dir = Path("benchmark_reports") - # cwd_report_dir.mkdir(exist_ok=True) - # (cwd_report_dir / f"{scenario.id}_benchmark_report.json").write_text( - # json.dumps(report, indent=2) - # ) - - # with track_phase( - # phase="compare", describe_exception=analyse_results_comparison_exception - # ): - # # Compare actual results with reference data - # try: - # assert_job_results_allclose( - # actual=actual_dir, - # expected=reference_dir, - # tmp_path=tmp_path, - # rtol=scenario.reference_options.get("rtol", 1e-3), - # atol=scenario.reference_options.get("atol", 1), - # pixel_tolerance=scenario.reference_options.get("pixel_tolerance", 1), - # ) - # except AssertionError as e: - # msg = str(e) - # if scenario.reference_data: - # msg += "\n\nReference data URLs:" - # for name, url in scenario.reference_data.items(): - # msg += f"\n {name}: {url}" - # if actual_s3_urls: - # msg += "\n\nActual data S3 URLs (uploaded on failure):" - # for name, url in actual_s3_urls.items(): - # msg += f"\n {name}: {url}" - # raise AssertionError(msg) from None + # Pre-compute S3 URLs for actual files (used in error messages and benchmark reports) + actual_s3_urls = { + str(p.relative_to(actual_dir)): upload_assets_on_fail.get_url(p) + for p in sorted(actual_dir.rglob("*")) if p.is_file() + } + actual_s3_urls = {k: v for k, v in actual_s3_urls.items() if v is not None} + + with track_phase(phase="download-reference"): + reference_dir = download_reference_data( + scenario=scenario, reference_dir=tmp_path / "reference" + ) + + if report_path is not None: + report = json.loads(report_path.read_text()) + report["actual_files"] = { + str(p.relative_to(actual_dir)): f"{p.stat().st_size / 1024:.1f} kb" + for p in sorted(actual_dir.rglob("*")) if p.is_file() + } + ref_files = {} + for p in sorted(reference_dir.rglob("*")): + if not p.is_file(): + continue + rel = p.relative_to(reference_dir) + size_str = f"{p.stat().st_size / 1024:.1f} kb" + actual_counterpart = actual_dir / rel + if not actual_counterpart.exists(): + size_str += " (missing in actual)" + elif actual_counterpart.stat().st_size != p.stat().st_size: + size_str += f" (actual: {actual_counterpart.stat().st_size / 1024:.1f} kb)" + ref_files[str(rel)] = size_str + report["reference_files"] = ref_files + if actual_s3_urls: + report["actual_data"] = actual_s3_urls + report_path.write_text(json.dumps(report, indent=2)) + # Also write to CWD so the report is accessible on Jenkins workspace + cwd_report_dir = Path("benchmark_reports") + cwd_report_dir.mkdir(exist_ok=True) + (cwd_report_dir / f"{scenario.id}_benchmark_report.json").write_text( + json.dumps(report, indent=2) + ) + + with track_phase( + phase="compare", describe_exception=analyse_results_comparison_exception + ): + # Compare actual results with reference data + try: + assert_job_results_allclose( + actual=actual_dir, + expected=reference_dir, + tmp_path=tmp_path, + rtol=scenario.reference_options.get("rtol", 1e-3), + atol=scenario.reference_options.get("atol", 1), + pixel_tolerance=scenario.reference_options.get("pixel_tolerance", 1), + ) + except AssertionError as e: + msg = str(e) + if scenario.reference_data: + msg += "\n\nReference data URLs:" + for name, url in scenario.reference_data.items(): + msg += f"\n {name}: {url}" + if actual_s3_urls: + msg += "\n\nActual data S3 URLs (uploaded on failure):" + for name, url in actual_s3_urls.items(): + msg += f"\n {name}: {url}" + raise AssertionError(msg) from None diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py index a63be038f..1096b207f 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py @@ -3,6 +3,7 @@ import dataclasses import json import logging +import mimetypes import os import re import time @@ -42,6 +43,10 @@ "https://schemas.opengis.net/ogcapi/features/part1/1.0/openapi/schemas/featureCollectionGeoJSON.yaml" ) +MIMETYPE_EXTENSION_MAP = { + "image/tiff; application=geotiff; profile=cloud-optimized": ".tif" +} + def get_client_credentials_env_var(url: str) -> str: """ @@ -250,7 +255,7 @@ def collect_ogc_results(*, api_client: ApiClientWrapper, job_id: str, user_token f"Processing result\n* Name: '{result_name}'\n" f"* media type: {qualified_value.media_type}\n" f"* Python type: {type(qualified_value.value)}\n" - "* schema {qualified_value.var_schema}..." + f"* schema {qualified_value.var_schema}..." ) if not isinstance(schema_reference, str): @@ -297,6 +302,24 @@ def collect_ogc_results(*, api_client: ApiClientWrapper, job_id: str, user_token return BenchmarkResults(assets=assets) +def _extract_download_link_from_asset(asset: dict) -> str | None: + """ + Extracts the download link from an asset dictionary. Checks if the `href` field is present and contains an HTTPS URL. + If this is not the case, look for an alternative link in `alternate` field. If no valid link is found, return None. + + Args: + asset (dict): The asset dictionary. + + Returns: + str | None: The download link if available, otherwise None. + """ + refs = [asset.get("href"), asset.get("alternate", {}).get("https", {}).get("href")] + for href in refs: + if href and href.startswith("https://"): + return href + return None + + def download_ogc_results( *, results_metadata: BenchmarkResults, @@ -315,13 +338,18 @@ def download_ogc_results( for output_name, output_data in sorted(results_metadata.assets.items()): output_data = to_jsonable(output_data) _log.debug(f"Downloading OGC API result '{output_name}' to {actual_dir=}") - if isinstance(output_data, dict) and output_data.get("href"): - ref = output_data["href"] - file_name = _resolve_output_filename(output_name=output_name, href=ref) + if isinstance(output_data, dict): + ref = _extract_download_link_from_asset(output_data) + mimetype = output_data.get("type") + if not ref: + _log.warning(f"Result '{output_name}' does not contain a valid download link, skipping download.") + continue + + file_name = _resolve_output_filename(output_name=output_name, href=ref, mimetype=mimetype) target = ensure_safe_relative_target(actual_dir, file_name) target.parent.mkdir(parents=True, exist_ok=True) download_file( - _convert_s3_href_to_https(ref, s3_endpoint=details.s3_endpoint if details else None), + ref, target, user_token=user_token, ) @@ -335,24 +363,23 @@ def download_ogc_results( return paths -def _convert_s3_href_to_https(href: str, s3_endpoint: str | None) -> str: - parsed = urlparse(href) - if not s3_endpoint: - raise ValueError(f"Cannot convert s3 href to https without s3_endpoint: {href=}, {s3_endpoint=}") - if parsed.scheme != "s3": - return href - bucket = parsed.netloc - key = parsed.path.lstrip("/") - if not bucket or not key: - raise ValueError(f"Invalid s3 href: {href!r}") - url = f"https://{bucket}.{s3_endpoint}/{key}" - _log.debug(f"Converted s3 href to https: {href} -> {url}") - return url - - -def _resolve_output_filename(*, output_name: str, href: str) -> str: +def _resolve_output_filename(*, output_name: str, href: str, mimetype: str | None = None) -> str: parsed = urlparse(href) name = Path(parsed.path).name if name: - return name + has_extension = bool(Path(name).suffix) + _log.debug(f"Result '{output_name}' download link '{href}' has filename '{name}' with extension: {has_extension}") + if has_extension: + return name + else: + _log.warning( + f"Result '{output_name}' download link '{href}' does not contain a file extension, " + f"falling back to using output name with mimetype-based extension using mimetype '{mimetype}'." + ) + extension = MIMETYPE_EXTENSION_MAP.get(mimetype) + if not extension and mimetype: + extension = mimetypes.guess_extension(mimetype) + _log.debug(f"Result '{output_name}' mimetype '{mimetype}' maps to extension: {extension}") + if extension: + return f"{output_name}{extension}" return f"{output_name}.bin" diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py index bbd8e829e..9af2d6f12 100644 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py @@ -1,6 +1,5 @@ from __future__ import annotations -import logging from pathlib import Path from apex_algorithm_qa_tools.benchmarks.ogc import ( @@ -56,8 +55,6 @@ def run_job(self, *, max_minutes: int | None): ) def collect_artifacts(self) -> BenchmarkRunnerArtifacts: - # TODO: remove after testing - self._job_id = "opensartoolkit-v2-2-1-k6wp5" if self._job_id is None: raise RuntimeError("Cannot collect OGC API metadata before run_job().")