Issue #404/#481 test_connection streamlining

- Use `dummy_backend` fixture for more compact setup
- Separate TestExecuteFromJsonResources and TestExecuteWithValidation

Author: soxofaan

openeo/rest/_testing.py

@@ -1,3 +1,4 @@
+import json
 import re
 from typing import Optional, Union
 
@@ -11,16 +12,33 @@ class DummyBackend:
     and allows inspection of posted process graphs
     """
 
+    __slots__ = (
+        "connection",
+        "sync_requests",
+        "batch_jobs",
+        "validation_requests",
+        "next_result",
+        "next_validation_errors",
+    )
+
     # Default result (can serve both as JSON or binary data)
     DEFAULT_RESULT = b'{"what?": "Result data"}'
 
     def __init__(self, requests_mock, connection: Connection):
         self.connection = connection
         self.sync_requests = []
         self.batch_jobs = {}
+        self.validation_requests = []
         self.next_result = self.DEFAULT_RESULT
-        requests_mock.post(connection.build_url("/result"), content=self._handle_post_result)
-        requests_mock.post(connection.build_url("/jobs"), content=self._handle_post_jobs)
+        self.next_validation_errors = []
+        requests_mock.post(
+            connection.build_url("/result"),
+            content=self._handle_post_result,
+        )
+        requests_mock.post(
+            connection.build_url("/jobs"),
+            content=self._handle_post_jobs,
+        )
         requests_mock.post(
             re.compile(connection.build_url(r"/jobs/(job-\d+)/results$")), content=self._handle_post_job_results
         )
@@ -32,12 +50,19 @@ def __init__(self, requests_mock, connection: Connection):
             re.compile(connection.build_url("/jobs/(.*?)/results/result.data$")),
             content=self._handle_get_job_result_asset,
         )
+        requests_mock.post(connection.build_url("/validation"), json=self._handle_post_validation)
 
     def _handle_post_result(self, request, context):
         """handler of `POST /result` (synchronous execute)"""
         pg = request.json()["process"]["process_graph"]
         self.sync_requests.append(pg)
-        return self.next_result
+        result = self.next_result
+        if isinstance(result, (dict, list)):
+            result = json.dumps(result).encode("utf-8")
+        elif isinstance(result, str):
+            result = result.encode("utf-8")
+        assert isinstance(result, bytes)
+        return result
 
     def _handle_post_jobs(self, request, context):
         """handler of `POST /jobs` (create batch job)"""
@@ -83,6 +108,12 @@ def _handle_get_job_result_asset(self, request, context):
         assert self.batch_jobs[job_id]["status"] == "finished"
         return self.next_result
 
+    def _handle_post_validation(self, request, context):
+        """Handler of `POST /validation` (validate process graph)."""
+        pg = request.json()["process_graph"]
+        self.validation_requests.append(pg)
+        return {"errors": self.next_validation_errors}
+
     def get_sync_pg(self) -> dict:
         """Get one and only synchronous process graph"""
         assert len(self.sync_requests) == 1

tests/rest/conftest.py

@@ -1,13 +1,11 @@
 import contextlib
 import re
 import typing
-from typing import List, Optional
 from unittest import mock
 
 import pytest
 import time_machine
 
-import openeo
 from openeo.rest._testing import DummyBackend, build_capabilities
 from openeo.rest.connection import Connection
 
@@ -73,26 +71,32 @@ def assert_oidc_device_code_flow(url: str = "https://oidc.test/dc", elapsed: flo
 
 
 @pytest.fixture
-def con100(requests_mock):
-    requests_mock.get(API_URL, json={"api_version": "1.0.0"})
+def api_capabilities() -> dict:
+    """Fixture to be overridden for customizing the capabilities doc used by connection fixtures."""
+    return {}
+
+
+@pytest.fixture
+def connection(api_version, requests_mock, api_capabilities) -> Connection:
+    requests_mock.get(API_URL, json=build_capabilities(api_version=api_version, **api_capabilities))
     con = Connection(API_URL)
     return con
 
 
 @pytest.fixture
-def con120(requests_mock):
-    requests_mock.get(API_URL, json={"api_version": "1.2.0"})
+def con100(requests_mock, api_capabilities):
+    requests_mock.get(API_URL, json=build_capabilities(api_version="1.0.0", **api_capabilities))
     con = Connection(API_URL)
     return con
 
 
 @pytest.fixture
-def dummy_backend(requests_mock, con100) -> DummyBackend:
-    yield DummyBackend(requests_mock=requests_mock, connection=con100)
+def con120(requests_mock, api_capabilities):
+    requests_mock.get(API_URL, json=build_capabilities(api_version="1.2.0", **api_capabilities))
+    con = Connection(API_URL)
+    return con
 
 
 @pytest.fixture
-def connection_with_pgvalidation(api_version, requests_mock) -> Connection:
-    """Connection fixture to a backend that supports validation of the process graph."""
-    requests_mock.get(API_URL, json=build_capabilities(api_version=api_version, validation=True))
-    return openeo.connect(API_URL)
+def dummy_backend(requests_mock, con100) -> DummyBackend:
+    yield DummyBackend(requests_mock=requests_mock, connection=con100)