Merge branch 'issue112-config-migration'

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 
 The format is roughly based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## [0.18.0]
+
+- Start porting `AggregatorConfig` fields to `AggregatorBackendConfig` ([#112](https://github.com/Open-EO/openeo-aggregator/issues/112))
+
+
 ## [0.17.0]
 
 - Support `aggregator_config` as `AggregatorConfig` variable name to simplify config system migration

setup.py

@@ -28,7 +28,7 @@
         "requests",
         "attrs",
         "openeo>=0.27.0",
-        "openeo_driver>=0.81.0.dev",
+        "openeo_driver>=0.86.0.dev",
         "flask~=2.0",
         "gunicorn~=20.0",
         "python-json-logger>=2.0.0",

src/openeo_aggregator/about.py

@@ -2,7 +2,7 @@
 import sys
 from typing import Optional
 
-__version__ = "0.17.0a1"
+__version__ = "0.18.0a1"
 
 
 def log_version_info(logger: Optional[logging.Logger] = None):

src/openeo_aggregator/app.py

@@ -27,7 +27,7 @@
 _log = logging.getLogger(__name__)
 
 
-def create_app(config: Any = None, auto_logging_setup: bool = True) -> flask.Flask:
+def create_app(config: Any = None, auto_logging_setup: bool = True, flask_error_handling: bool = True) -> flask.Flask:
     """
     Flask application factory function.
     """
@@ -39,6 +39,7 @@ def create_app(config: Any = None, auto_logging_setup: bool = True) -> flask.Fla
 
     log_version_info(logger=_log)
 
+    # TODO #112 move this default to the AggregatorBackendConfig getter (get_backend_config)
     os.environ.setdefault(ConfigGetter.OPENEO_BACKEND_CONFIG, str(get_config_dir() / "backend_config.py"))
 
     config: AggregatorConfig = get_config(config)
@@ -53,7 +54,7 @@ def create_app(config: Any = None, auto_logging_setup: bool = True) -> flask.Fla
     _log.info(f"Building Flask app with {backend_implementation=!r}")
     app = openeo_driver.views.build_app(
         backend_implementation=backend_implementation,
-        error_handling=config.flask_error_handling,
+        error_handling=flask_error_handling,
     )
 
     app.config.from_mapping(

src/openeo_aggregator/backend.py

@@ -71,6 +71,7 @@
     CONNECTION_TIMEOUT_JOB_START,
     CONNECTION_TIMEOUT_RESULT,
     AggregatorConfig,
+    get_backend_config,
 )
 from openeo_aggregator.connection import (
     BackendConnection,
@@ -353,7 +354,6 @@ def __init__(
         self._memoizer = memoizer_from_config(config=config, namespace="Processing")
         self.backends.on_connections_change.add(self._memoizer.invalidate)
         self._catalog = catalog
-        self._stream_chunk_size = config.streaming_chunk_size
 
     def get_process_registry(
         self, api_version: Union[str, ComparableVersion]
@@ -537,7 +537,7 @@ def evaluate(self, process_graph: dict, env: EvalEnv = None):
                 _log.error(f"Failed to process synchronously on backend {con.id}: {e!r}", exc_info=True)
                 raise OpenEOApiException(message=f"Failed to process synchronously on backend {con.id}: {e!r}")
 
-        return streaming_flask_response(backend_response, chunk_size=self._stream_chunk_size)
+        return streaming_flask_response(backend_response, chunk_size=get_backend_config().streaming_chunk_size)
 
     def preprocess_process_graph(self, process_graph: FlatPG, backend_id: str) -> dict:
         def preprocess(node: Any) -> Any:

src/openeo_aggregator/config.py

@@ -1,10 +1,11 @@
 import logging
 import os
 from pathlib import Path
-from typing import List, Union
+from typing import Callable, List, Union
 
 import attrs
 from openeo_driver.config import OpenEoBackendConfig
+from openeo_driver.config.load import ConfigGetter
 from openeo_driver.server import build_backend_deploy_metadata
 from openeo_driver.users.oidc import OidcProvider
 from openeo_driver.utils import dict_item
@@ -43,9 +44,6 @@ class AggregatorConfig(dict):
     # Dictionary mapping backend id to backend url
     aggregator_backends = dict_item()
 
-    flask_error_handling = dict_item(default=True)
-    streaming_chunk_size = dict_item(default=STREAM_CHUNK_SIZE_DEFAULT)
-
     # TODO: add validation/normalization to make sure we have a real list of OidcProvider objects?
     configured_oidc_providers: List[OidcProvider] = dict_item(default=[])
     auth_entitlement_check: Union[bool, dict] = dict_item(default=False)
@@ -63,6 +61,9 @@ class AggregatorConfig(dict):
     # List of collection ids to cover with the aggregator (when None: support union of all upstream collections)
     collection_whitelist = dict_item(default=None)
 
+    # Just a config field for test purposes (while were stripping down this config class)
+    test_dummy = dict_item(default="alice")
+
     @staticmethod
     def from_py_file(path: Union[str, Path]) -> 'AggregatorConfig':
         """Load config from Python file."""
@@ -133,3 +134,11 @@ class AggregatorBackendConfig(OpenEoBackendConfig):
     capabilities_deploy_metadata: dict = build_backend_deploy_metadata(
         packages=["openeo", "openeo_driver", "openeo_aggregator"],
     )
+
+    streaming_chunk_size: int = STREAM_CHUNK_SIZE_DEFAULT
+
+
+
+_config_getter = ConfigGetter(expected_class=AggregatorBackendConfig)
+
+get_backend_config: Callable[..., AggregatorBackendConfig] = _config_getter.get

src/openeo_aggregator/testing.py

@@ -1,19 +1,19 @@
-import collections
 import dataclasses
 import datetime
 import itertools
 import json
 import pathlib
-import time
 from typing import Any, Dict, List, Optional, Tuple, Union
 from unittest import mock
 
 import kazoo
 import kazoo.exceptions
+import openeo_driver.testing
 import pytest
 from openeo.util import rfc3339
 
 import openeo_aggregator.about
+import openeo_aggregator.config
 from openeo_aggregator.utils import Clock
 
 
@@ -290,3 +290,32 @@ def processes(self, *args) -> dict:
             processes.append(process)
 
         return {"processes": processes, "links": []}
+
+
+def config_overrides(**kwargs):
+    """
+    *Only to be used in unit tests*
+
+    `mock.patch` based mocker to override the config returned by `get_backend_config()` at run time
+
+    Can be used as context manager
+
+        >>> with config_overrides(id="foobar"):
+        ...     ...
+
+    in a fixture (as context manager):
+
+        >>> @pytest.fixture
+        ... def custom_setup()
+        ...     with config_overrides(id="foobar"):
+        ...         yield
+
+    or as test function decorator
+
+        >>> @config_overrides(id="foobar")
+        ... def test_stuff():
+    """
+    return openeo_driver.testing.config_overrides(
+        config_getter=openeo_aggregator.config._config_getter,
+        **kwargs,
+    )

tests/conftest.py

@@ -177,9 +177,13 @@ def multi_backend_connection(config) -> MultiBackendConnection:
 
 
 def get_flask_app(config: AggregatorConfig) -> flask.Flask:
-    app = create_app(config=config, auto_logging_setup=False)
-    app.config['TESTING'] = True
-    app.config['SERVER_NAME'] = 'oeoa.test'
+    app = create_app(
+        config=config,
+        auto_logging_setup=False,
+        # flask_error_handling=False,  # Failing test debug tip: set to False for deeper stack trace insights
+    )
+    app.config["TESTING"] = True
+    app.config["SERVER_NAME"] = "oeoa.test"
     return app
 
 

tests/test_config.py

@@ -21,7 +21,7 @@ def _get_config_content(config_var_name: str = "config"):
         {config_var_name} = AggregatorConfig(
             config_source=__file__,
             aggregator_backends={{"b1": "https://b1.test"}},
-            streaming_chunk_size=123
+            test_dummy="bob",
         )
         """
     )
@@ -31,8 +31,7 @@ def test_config_defaults():
     config = AggregatorConfig()
     with pytest.raises(KeyError):
         _ = config.aggregator_backends
-    assert config.flask_error_handling is True
-    assert config.streaming_chunk_size == STREAM_CHUNK_SIZE_DEFAULT
+    assert config.test_dummy == "alice"
 
 
 def test_config_aggregator_backends():
@@ -49,7 +48,7 @@ def test_config_from_py_file(tmp_path, config_var_name):
     config = AggregatorConfig.from_py_file(path)
     assert config.config_source == str(path)
     assert config.aggregator_backends == {"b1": "https://b1.test"}
-    assert config.streaming_chunk_size == 123
+    assert config.test_dummy == "bob"
 
 
 def test_config_from_py_file_wrong_config_var_name(tmp_path):
@@ -72,7 +71,7 @@ def test_get_config_py_file_path(tmp_path, convertor):
     config = get_config(convertor(config_path))
     assert config.config_source == str(config_path)
     assert config.aggregator_backends == {"b1": "https://b1.test"}
-    assert config.streaming_chunk_size == 123
+    assert config.test_dummy == "bob"
 
 
 def test_get_config_env_py_file(tmp_path):
@@ -83,4 +82,4 @@ def test_get_config_env_py_file(tmp_path):
         config = get_config()
     assert config.config_source == str(path)
     assert config.aggregator_backends == {"b1": "https://b1.test"}
-    assert config.streaming_chunk_size == 123
+    assert config.test_dummy == "bob"

tests/test_views.py

@@ -33,7 +33,7 @@
     STAC_PROPERTY_FEDERATION_BACKENDS,
     STAC_PROPERTY_PROVIDER_BACKEND,
 )
-from openeo_aggregator.testing import clock_mock
+from openeo_aggregator.testing import clock_mock, config_overrides
 
 from .conftest import assert_dict_subset, get_api100, get_flask_app
 
@@ -800,7 +800,6 @@ def test_result_basic_math_error(self, api100, requests_mock, backend1, backend2
 
     @pytest.mark.parametrize(["chunk_size"], [(16,), (128,)])
     def test_result_large_response_streaming(self, config, chunk_size, requests_mock, backend1, backend2):
-        config.streaming_chunk_size = chunk_size
         api100 = get_api100(get_flask_app(config))
 
         def post_result(request: requests.Request, context):
@@ -813,7 +812,9 @@ def post_result(request: requests.Request, context):
         api100.set_auth_bearer_token(token=TEST_USER_BEARER_TOKEN)
         pg = {"large": {"process_id": "large", "arguments": {}, "result": True}}
         request = {"process": {"process_graph": pg}}
-        res = api100.post("/result", json=request).assert_status_code(200)
+
+        with config_overrides(streaming_chunk_size=chunk_size):
+            res = api100.post("/result", json=request).assert_status_code(200)
 
         assert res.response.is_streamed
         chunks = res.response.iter_encoded()