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..3ddf925c8 --- /dev/null +++ b/algorithm_catalog/terradue/gep_api_ost/benchmark_scenarios/gep_api_ost.json @@ -0,0 +1,62 @@ +[ + { + "id": "gep_api_ost", + "type": "ogc_api_process", + "description": "Benchmark for the GEP API OST process", + "endpoint": "https://processing.geohazards-tep.eu", + "namespace": "44454d3235313539", + "application": "opensartoolkit-v2-2-1", + "auth": { + "url": "https://iam.terradue.com/", + "realm": "master" + }, + "parameters": { + "resolution": 100, + "ard-type": "OST_GTC", + "with-speckle-filter": "NO-FILTER", + "resampling-method": "BILINEAR_INTERPOLATION", + "target_datetime": "2024-11-12T00:00:00Z", + "bbox": { + "class": "https://raw.githubusercontent.com/eoap/schemas/main/geojson.yaml#Polygon", + "type": "Polygon", + "bbox": [ + 12.146, + 41.701, + 12.629, + 41.967 + ], + "coordinates": [ + [ + [ + 12.146, + 41.701 + ], + [ + 12.629, + 41.701 + ], + [ + 12.629, + 41.967 + ], + [ + 12.146, + 41.967 + ], + [ + 12.146, + 41.701 + ] + ] + ] + } + }, + "reference_data": { + "job-results.json": "file:///Users/bramjanssen/projects/apex/apex_algorithms/qa/benchmarks/data/job-results.json", + "ost-ard-cog.tif": "file:///Users/bramjanssen/projects/apex/apex_algorithms/qa/benchmarks/data/ost-ard-cog.tif" + }, + "reference_options": { + "atol": 1 + } + } +] \ No newline at end of file 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/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/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 916369353..34a397537 100644 --- a/qa/benchmarks/tests/conftest.py +++ b/qa/benchmarks/tests/conftest.py @@ -92,92 +92,4 @@ def pytest_collection_modifyitems(session, config, items): selected_ids = set(item.nodeid for item in selected) deselected = [item for item in items if item.nodeid not in selected_ids] config.hook.pytest_deselected(items=deselected) - items[:] = selected - - -def _get_client_credentials_env_var(url: str) -> str: - """ - Get client credentials env var name for a given backend URL. - """ - if not re.match(r"https?://", url): - url = f"https://{url}" - parsed = urllib.parse.urlparse(url) - hostname = parsed.hostname - if hostname in { - "openeo.dataspace.copernicus.eu", - "openeofed.dataspace.copernicus.eu", - "openeo.terrascope.be", - }: - return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSE" - elif hostname in { - "openeo-staging.dataspace.copernicus.eu", - "openeo-staging.terrascope.be", - "openeo-dev.terrascope.be", - }: - return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG" - elif re.fullmatch(r"openeo\.dev\.([a-z0-9-]+)\.openeo-int\.v1\.dataspace\.copernicus\.eu", hostname): - return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG" - elif hostname in {"openeo.cloud", "openeo.eodc.eu"}: - return "OPENEO_AUTH_CLIENT_CREDENTIALS_EGI" - elif hostname in { - "openeo-dev.vito.be", - "openeo.vito.be", - }: - return "OPENEO_AUTH_CLIENT_CREDENTIALS_TERRASCOPE" - else: - raise ValueError(f"Unsupported backend: {url=} ({hostname=})") - - -@pytest.fixture -def connection_factory(request, capfd) -> Callable[[], openeo.Connection]: - """ - Fixture for a function that sets up an authenticated connection to an openEO backend. - - This is implemented as a fixture to have access to other fixtures that allow - deeper integration with the pytest framework. - For example, the `request` fixture allows to identify the currently running test/benchmark. - """ - - # Identifier for the current test/benchmark, to be injected automatically - # into requests to the backend for tracking/cross-referencing purposes - origin = f"apex-algorithms/benchmarks/{request.session.name}/{request.node.name}" - - def get_connection(url: str) -> openeo.Connection: - session = requests.Session() - session.params["_origin"] = origin - - _log.info(f"Connecting to {url!r}") - connection = openeo.connect(url, auto_validate=False, session=session) - connection.default_headers["X-OpenEO-Client-Context"] = "APEx Algorithm Benchmarks" - - # Authentication: - # In production CI context, we want to extract client credentials - # from environment variables (based on backend url). - # In absence of such environment variables, to allow local development, - # we fall back on a traditional `authenticate_oidc()` - # which automatically supports various authentication flows (device code, refresh token, client credentials, etc.) - auth_env_var = _get_client_credentials_env_var(url) - _log.info(f"Checking for {auth_env_var=} to drive auth against {url=}.") - if auth_env_var in os.environ: - client_credentials = os.environ[auth_env_var] - provider_id, client_id, client_secret = client_credentials.split("/", 2) - _log.info(f"Extracted {provider_id=} {client_id=} from {auth_env_var=}") - connection.authenticate_oidc_client_credentials( - provider_id=provider_id, - client_id=client_id, - client_secret=client_secret, - ) - else: - # Temporarily disable output capturing, - # to make sure that the OIDC device code instructions are shown - # to the user running interactively. - with capfd.disabled(): - # Use a shorter max poll time by default - # to alleviate the default impression that the test seem to hang - # because of the OIDC device code poll loop. - max_poll_time = int(os.environ.get("OPENEO_OIDC_DEVICE_CODE_MAX_POLL_TIME") or 30) - connection.authenticate_oidc(max_poll_time=max_poll_time) - - return connection - - return get_connection + items[:] = selected \ No newline at end of file diff --git a/qa/benchmarks/tests/test_benchmarks.py b/qa/benchmarks/tests/test_benchmarks.py index e480bc91a..548472430 100644 --- a/qa/benchmarks/tests/test_benchmarks.py +++ b/qa/benchmarks/tests/test_benchmarks.py @@ -1,16 +1,8 @@ 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 +10,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 +33,6 @@ ) def test_run_benchmark( scenario: BenchmarkScenario, - connection_factory, tmp_path: Path, track_metric, track_phase, @@ -39,34 +40,36 @@ 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({ + report_data = { "scenario_id": scenario.id, + "scenario_type": scenario.type, "scenario_description": scenario.description, - "scenario_backend": scenario.backend, "scenario_source": str(scenario.source) if scenario.source else None, "reference_data": scenario.reference_data, "reference_options": scenario.reference_options, - }, indent=2)) + } + # Add scenario type-specific fields if they exist + if hasattr(scenario, 'backend'): + report_data["scenario_backend"] = scenario.backend + if hasattr(scenario, 'process_id'): + report_data["process_id"] = scenario.process_id + elif hasattr(scenario, 'application'): + report_data["application"] = scenario.application + if hasattr(scenario, 'endpoint'): + report_data["endpoint"] = scenario.endpoint + + report_path = tmp_path / "benchmark_report.json" + report_path.write_text(json.dumps(report_data, indent=2)) upload_assets_on_fail(report_path) def _on_phase_exception(phase: str, exc: Exception): @@ -87,50 +90,36 @@ def _on_phase_exception(phase: str, exc: Exception): track_phase.on_exception = _on_phase_exception - with track_phase(phase="create-job"): - # TODO #14 scenario option to use synchronous instead of batch job mode? - job = connection.create_job( - process_graph=scenario.process_graph, - title=f"APEx benchmark {scenario.id}", - additional=scenario.job_options, - ) - track_metric("job_id", job.job_id) + artifacts = None - if report_path is not None: - report = json.loads(report_path.read_text()) - report["job_id"] = job.job_id - report_path.write_text(json.dumps(report, indent=2)) + with track_phase(phase="create-job"): + runner.create_job() with track_phase(phase="run-job"): - # TODO: monitor timing and progress - # TODO: separate "job started" and run phases? max_minutes = request.config.getoption("--maximum-job-time-in-minutes") - if max_minutes: - def _timeout_handler(signum, frame): - raise TimeoutError( - f"Batch job {job.job_id} exceeded maximum allowed time of {max_minutes} minutes" - ) - - old_handler = signal.signal(signal.SIGALRM, _timeout_handler) - signal.alarm(max_minutes * 60) - try: - job.start_and_wait() - finally: - if max_minutes: - signal.alarm(0) - signal.signal(signal.SIGALRM, old_handler) + runner.run_job(max_minutes=max_minutes) with track_phase(phase="collect-metadata"): - collect_metrics_from_job_metadata(job, track_metric=track_metric) - - results = job.get_results() - collect_metrics_from_results_metadata(results, track_metric=track_metric) + artifacts = runner.collect_artifacts() + if artifacts.job_id: + track_metric("job_id", artifacts.job_id) + if report_path is not None: + report = json.loads(report_path.read_text()) + report["job_id"] = artifacts.job_id + report_path.write_text(json.dumps(report, indent=2)) + + collect_metrics_from_job_metadata( + artifacts.job_metadata, + track_metric=track_metric, + ) + collect_metrics_from_results_metadata( + artifacts.results_metadata, + track_metric=track_metric, + ) with track_phase(phase="download-actual"): - # Download actual results actual_dir = tmp_path / "actual" - paths = results.download_files(target=actual_dir, include_stac_metadata=True) - + paths = runner.download_actual(actual_dir=actual_dir) # Upload assets on failure upload_assets_on_fail(*paths) diff --git a/qa/tools/apex_algorithm_qa_tools/benchmarks.py b/qa/tools/apex_algorithm_qa_tools/benchmarks.py deleted file mode 100644 index aada509f0..000000000 --- a/qa/tools/apex_algorithm_qa_tools/benchmarks.py +++ /dev/null @@ -1,58 +0,0 @@ -""" -Reusable utilities to use in benchmarking -""" - -from typing import Union - -from apex_algorithm_qa_tools.pytest.pytest_track_metrics import MetricsTracker -from openeo.rest.job import BatchJob, JobResults - - -def collect_metrics_from_job_metadata( - job_metadata: Union[BatchJob, 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: - track_metric( - f"usage:{usage_metric}:{usage_data['unit']}", usage_data["value"] - ) - - -def collect_metrics_from_results_metadata( - results_metadata: Union[BatchJob, JobResults, 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(): - if "proj:shape" in asset_data: - y, x = asset_data["proj:shape"][:2] - proj_shape_area_mpx.append(y * x / 1e6) - - if "proj:epsg" in asset_data and "proj:bbox" in asset_data: - epsg_code = int(asset_data["proj:epsg"]) - if (32601 <= epsg_code < 32660) or (32701 <= epsg_code < 32760): - # UTM zone: bbox coordinates are in meter -> calculate area in km2 - w, s, e, n = asset_data["proj:bbox"][:4] - proj_shape_area_km2.append((n - s) * (e - w) / 1e6) - - # When there are multiple results, just report maximum for now - if proj_shape_area_mpx: - track_metric("results:proj:shape:area:megapixel", max(proj_shape_area_mpx)) - if proj_shape_area_km2: - track_metric("results:proj:bbox:area:utm:km2", max(proj_shape_area_km2)) - - -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" 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/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 new file mode 100644 index 000000000..9fcee450a --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/common.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import dataclasses +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 + + +@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: BenchmarkJobMetadata, + track_metric: MetricsTracker, +): + 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: BenchmarkResults, track_metric: MetricsTracker): + proj_shape_area_mpx = [] + proj_shape_area_km2 = [] + 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) + + if "proj:epsg" in asset_data and "proj:bbox" in asset_data: + epsg_code = int(asset_data["proj:epsg"]) + if (32601 <= epsg_code < 32660) or (32701 <= epsg_code < 32760): + # UTM zone: bbox coordinates are in meter -> calculate area in km2 + w, s, e, n = asset_data["proj:bbox"][:4] + proj_shape_area_km2.append((n - s) * (e - w) / 1e6) + + # When there are multiple results, just report maximum for now + if proj_shape_area_mpx: + track_metric("results:proj:shape:area:megapixel", max(proj_shape_area_mpx)) + if proj_shape_area_km2: + track_metric("results:proj:bbox:area:utm:km2", max(proj_shape_area_km2)) + + +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" + + +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 new file mode 100644 index 000000000..1096b207f --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/ogc.py @@ -0,0 +1,385 @@ +from __future__ import annotations + +import dataclasses +import json +import logging +import mimetypes +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" +) + +MIMETYPE_EXTENSION_MAP = { + "image/tiff; application=geotiff; profile=cloud-optimized": ".tif" +} + + +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: + # 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, _, client_secret = client_credentials.partition("/") + 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/'." + "Or set OGC_AUTH_TOKEN with a valid JWT token." + ) + + +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}...") + + # 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 + + +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 _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") + 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" + f"* 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_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 = _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, " + 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 _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, + 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] = [] + + _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): + 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( + ref, + 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 _resolve_output_filename(*, output_name: str, href: str, mimetype: str | None = None) -> str: + parsed = urlparse(href) + name = Path(parsed.path).name + if 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/openeo.py b/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py new file mode 100644 index 000000000..bb7aae39a --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/openeo.py @@ -0,0 +1,126 @@ +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", + "openeo.terrascope.be", + }: + return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSE" + elif hostname in { + "openeo-staging.dataspace.copernicus.eu", + "openeo-staging.terrascope.be", + "openeo-dev.terrascope.be", + }: + return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG" + elif re.fullmatch(r"openeo\.dev\.([a-z0-9-]+)\.openeo-int\.v1\.dataspace\.copernicus\.eu", hostname): + return "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSESTAG" + elif hostname in {"openeo.cloud", "openeo.eodc.eu"}: + return "OPENEO_AUTH_CLIENT_CREDENTIALS_EGI" + elif hostname in { + "openeo-dev.vito.be", + "openeo.vito.be", + }: + return "OPENEO_AUTH_CLIENT_CREDENTIALS_TERRASCOPE" + else: + raise ValueError(f"Unsupported backend: {url=} ({hostname=})") + + +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(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) + + +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) 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..e0ce4e697 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/base.py @@ -0,0 +1,56 @@ +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 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: 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/{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..1a3647010 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/factory.py @@ -0,0 +1,17 @@ +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): + if scenario.type == "openeo": + return OpenEOBenchmarkRunner( + scenario=scenario, + request=request, + ) + 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 new file mode 100644 index 000000000..9af2d6f12 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/ogc.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +from pathlib import Path + +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._job_id = None + self._results = None + + def create_job(self): + self._job = create_ogc_job( + scenario=self.scenario, + ) + + 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 collect_artifacts(self) -> BenchmarkRunnerArtifacts: + if self._job_id is None: + raise RuntimeError("Cannot collect OGC API metadata before run_job().") + + 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 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 new file mode 100644 index 000000000..f1e9502e7 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/benchmarks/runners/openeo.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +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.openeo import openEOBenchmarkScenario + + +class OpenEOBenchmarkRunner(BenchmarkRunner): + def __init__(self, *, scenario: openEOBenchmarkScenario, 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/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__) 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..3b7f6555d --- /dev/null +++ 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 new file mode 100644 index 000000000..ed12a54a1 --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/common.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import glob +import logging +import re +from pathlib import Path +from typing import List + +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.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]: + # 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) + 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.endpoint.startswith(("http://", "https://")), ( + f"Unsupported OGC endpoint: {scenario.endpoint!r}" + ) + assert scenario.application or 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=}", + 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): + # 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 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..dafe1ae7c --- /dev/null +++ b/qa/tools/apex_algorithm_qa_tools/scenarios/factory.py @@ -0,0 +1,30 @@ +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 +from apex_algorithm_qa_tools.scenarios.ogc import OGCAPIBenchmarkScenario + + +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) + 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/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..080f1228e 100644 --- a/qa/tools/pyproject.toml +++ b/qa/tools/pyproject.toml @@ -21,7 +21,9 @@ 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", + "stac-pydantic" ] 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_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: 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..94cbe745a 100644 --- a/schemas/benchmark_scenario.json +++ b/schemas/benchmark_scenario.json @@ -1,48 +1,159 @@ { "$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." + "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", + "ogc_api_process" + ] + }, + "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" + ] }, - "reference_data": { + "openeo": { "type": "object", - "description": "Reference data of the benchmark, to compare actual results with." + "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": [ + "backend", + "process_graph" + ] }, - "reference_options": { + "ogcapi_process": { "type": "object", - "description": "Options to fine-tune how actual and reference results should be compared." + "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" + ] + } + ] } }, - "required": [ - "id", - "type", - "backend", - "process_graph" - ] -} + "oneOf": [ + { + "allOf": [ + { + "$ref": "#/$defs/common" + }, + { + "$ref": "#/$defs/openeo" + } + ] + }, + { + "allOf": [ + { + "$ref": "#/$defs/common" + }, + { + "$ref": "#/$defs/ogcapi_process" + } + ] + } + ], + "unevaluatedProperties": false +} \ No newline at end of file