Add ephemeral_flask_server testing utility

e.g. to simplify testing of pystac + pystac_client based logic (which uses a mix of urllib and requests)

related to work under Open-EO/openeo-geopyspark-driver#1307

Author: soxofaan

CHANGELOG.md

@@ -19,7 +19,12 @@ and start a new "In Progress" section above it.
 
 <!-- start-of-changelog -->
 
-## In progress: 0.136.0
+## In progress: 0.137.0
+
+- Add `ephemeral_flask_server` testing utility (`openeo_driver.testing`) for request mocking based on a Flask app. Allows to do request/response mocking independently from actual request library (`requests`, `urllib`, `urllib3`, etc.) through a well-documented API (Flask).
+
+
+## 0.136.0
 
 - Start supporting custom `UdfRuntimes` implementation in `OpenEoBackendImplementation` ([#415](https://github.com/Open-EO/openeo-python-driver/issues/415))
 - Process graph parsing (dry-run) for very large graphs got faster. ([#426](https://github.com/Open-EO/openeo-python-driver/issues/426))

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.136.0a4"
+__version__ = "0.137.0a1"

openeo_driver/testing.py

@@ -8,18 +8,23 @@
 import logging
 import math
 import multiprocessing
+import queue
 import re
+import threading
+import typing
 import urllib.request
 from pathlib import Path
 from typing import Any, Callable, Collection, Dict, Optional, Pattern, Tuple, Union
 from unittest import mock
 
 import attrs
+import flask
 import openeo
 import openeo.processes
 import pytest
 import shapely.geometry.base
 import shapely.wkt
+import werkzeug.serving
 from flask import Response
 from flask.testing import FlaskClient
 from openeo.utils.version import ComparableVersion
@@ -648,7 +653,7 @@ def test_my_function(caplog, monkeypatch)
 
 
 @contextlib.contextmanager
-def ephemeral_fileserver(path: Union[Path, str], host: str = "localhost", port: int = 0) -> str:
+def ephemeral_fileserver(path: Union[Path, str], host: str = "localhost", port: int = 0) -> typing.Iterator[str]:
     """
     Context manager to run a short-lived (static) file HTTP server, serving files from a given local test data folder.
 
@@ -696,6 +701,63 @@ def run(queue: multiprocessing.Queue):
         server_process.close()
 
 
+@contextlib.contextmanager
+def ephemeral_flask_server(app: flask.Flask) -> typing.Iterator[str]:
+    """
+    Context manager to run a Flask app in a separate thread for testing purposes.
+
+    Usage example:
+
+        def test_hello():
+
+            app = flask.Flask(__name__)
+            @app.route("/hello")
+            def hello():
+                return "Hello, World!"
+
+            with ephemeral_flask_server(app) as root_url:
+                resp = requests.get(f"{root_url}/hello")
+                assert resp.status_code == 200
+                assert resp.text == "Hello, World!"
+    """
+
+    class FlaskServerThread(threading.Thread):
+        def __init__(self, app: flask.Flask, root_url_queue: queue.Queue, *, host: str = "127.0.0.1", port: int = 0):
+            super().__init__()
+            self.daemon = True
+
+            self._server = werkzeug.serving.make_server(host, port, app)
+            self._ctx = app.app_context()
+            self._ctx.push()
+            self._root_url_queue = root_url_queue
+
+        def run(self):
+            root_url = f"http://{self._server.server_address[0]}:{self._server.server_port}"
+            self._root_url_queue.put(root_url)
+            _log.debug(f"FlaskServerThread: start serving at {root_url=}")
+            self._server.serve_forever()
+
+        def shutdown(self):
+            self._ctx.pop()
+            self._server.shutdown()
+            _log.debug("FlaskServerThread: is shut down")
+
+    # Use queue to communicate the root URL back to the main thread
+    root_url_queue = queue.Queue()
+    _log.debug(f"ephemeral_flask_server: starting server thread with {app=}")
+    server_thread = FlaskServerThread(app=app, root_url_queue=root_url_queue)
+
+    server_thread.start()
+    root_url = root_url_queue.get(timeout=2)
+    _log.debug(f"ephemeral_flask_server: {server_thread=} is serving at {root_url=}")
+
+    try:
+        yield root_url
+    finally:
+        _log.debug(f"ephemeral_flask_server: shutting down {server_thread=}")
+        server_thread.shutdown()
+
+
 def config_overrides(config_getter: ConfigGetter = _backend_config_getter, **kwargs):
     """
     *Only to be used in unit tests*

tests/test_testing.py

@@ -10,6 +10,7 @@
 import numpy
 import pytest
 import requests
+import requests.exceptions
 
 from openeo_driver.config import get_backend_config
 from openeo_driver.testing import (
@@ -25,6 +26,7 @@
     caplog_with_custom_formatter,
     config_overrides,
     ephemeral_fileserver,
+    ephemeral_flask_server,
     preprocess_check_and_replace,
 )
 
@@ -258,6 +260,51 @@ def test_ephemeral_fileserver_failure(tmp_path, caplog):
     assert "terminated with exitcode" in caplog.text
 
 
+def test_ephemeral_flask_server_basic():
+    app = flask.Flask(__name__)
+
+    @app.route("/hello")
+    def hello():
+        return "Hello, World!"
+
+    with ephemeral_flask_server(app) as root_url:
+        resp = requests.get(f"{root_url}/hello")
+        assert resp.status_code == 200
+        assert resp.text == "Hello, World!"
+
+
+def test_ephemeral_flask_server_shutdown():
+    app = flask.Flask(__name__)
+
+    @app.route("/hello")
+    def hello():
+        return "Hello, World!"
+
+    with ephemeral_flask_server(app) as root_url:
+        resp = requests.get(f"{root_url}/hello")
+        assert resp.text == "Hello, World!"
+
+    with pytest.raises(requests.exceptions.ConnectionError, match="Connection refused"):
+        requests.get(f"{root_url}/hello")
+
+
+def test_ephemeral_flask_server_dynamic():
+    app = flask.Flask(__name__)
+
+    @app.route("/hello/<name>")
+    def hello(name):
+        return flask.jsonify({"hello": name})
+
+    with ephemeral_flask_server(app) as root_url:
+        resp = requests.get(f"{root_url}/hello/alice")
+        assert resp.status_code == 200
+        assert resp.json() == {"hello": "alice"}
+
+        resp = requests.get(f"{root_url}/hello/bob")
+        assert resp.status_code == 200
+        assert resp.json() == {"hello": "bob"}
+
+
 def test_approxify_basic():
     assert {"a": 1.2345} == approxify({"a": 1.2345})
     assert {"a": 1.2345} != approxify({"a": 1.23466666})