Issue #404 Add test coverage for processgraph validation in datacube

tests/rest/datacube/conftest.py

@@ -85,6 +85,12 @@ def con100(requests_mock, support_udp) -> Connection:
     return _setup_connection("1.0.0", requests_mock, build_capabilities_kwargs={"udp": support_udp})
 
 
+@pytest.fixture
+def connection_with_pgvalidation_datacube(api_version, requests_mock) -> Connection:
+    """Connection fixture to a backend that supports validation of the process graph."""
+    return _setup_connection("1.0.0", requests_mock, build_capabilities_kwargs={"udp": support_udp, "validation": True})
+
+
 @pytest.fixture
 def s2cube(connection, api_version) -> DataCube:
     return connection.load_collection("S2")

tests/rest/datacube/test_datacube.py

@@ -4,16 +4,19 @@
 - 1.0.0-style DataCube
 
 """
+import json
 import pathlib
 from datetime import date, datetime
 from unittest import mock
 
 import numpy as np
 import pytest
+import requests
 import shapely
 import shapely.geometry
 
 from openeo.rest import BandMathException
+from openeo.rest.connection import Connection
 from openeo.rest.datacube import DataCube
 
 from ... import load_json_resource
@@ -807,3 +810,104 @@ def test_save_result_format_options_vs_execute_batch(elf, s2cube, get_create_job
             },
             "result": True,
         }
+
+
+class TestProcessGraphValidation:
+    JOB_ID = "j-123"
+    PROCESS_GRAPH_DICT = {"add1": {"process_id": "add", "arguments": {"x": 3, "y": 5}, "result": True}}
+    PROCESS_GRAPH_STRING = json.dumps(PROCESS_GRAPH_DICT)
+
+    @pytest.fixture
+    def cube_add(self, requests_mock, connection_with_pgvalidation_datacube: Connection) -> DataCube:
+        requests_mock.post(API_URL + "/result", content=self._post_result_handler_json)
+        return connection_with_pgvalidation_datacube.datacube_from_json(self.PROCESS_GRAPH_STRING)
+
+    def _post_jobs_handler_json(self, response: requests.Request, context):
+        # pg = response.json()["process"]["process_graph"]
+        # assert pg == {"add": {"process_id": "add", "arguments": {"x": 3, "y": 5}, "result": True}}
+        context.headers["OpenEO-Identifier"] = self.JOB_ID
+        return b""
+
+    def _post_result_handler_json(self, response: requests.Request, context):
+        pg = response.json()["process"]["process_graph"]
+        # assert pg == {"add": {"process_id": "add", "arguments": {"x": 3, "y": 5}, "result": True}}
+        assert pg == self.PROCESS_GRAPH_DICT
+        return b'{"answer": 8}'
+
+    def _post_result_handler_tiff(self, response: requests.Request, context):
+        pg = response.json()["process"]["process_graph"]
+        assert pg == self.PROCESS_GRAPH_DICT
+        return b"TIFF data"
+
+    @pytest.mark.parametrize("validate", [True, False])
+    def test_create_job_with_pg_validation(
+        self,
+        requests_mock,
+        connection_with_pgvalidation_datacube: Connection,
+        validate,
+    ):
+        """The DataCube should pass through request for the validation to the
+        connection and the validation endpoint should only be called when
+        validation was requested.
+        """
+        m = requests_mock.post(API_URL + "/validation", json={"errors": []})
+
+        requests_mock.post(API_URL + "/jobs", status_code=201, content=self._post_jobs_handler_json)
+        cube: DataCube = connection_with_pgvalidation_datacube.load_collection("S2")
+        cube.create_job(validate=validate)
+
+        # Validation should be called if and only if it was requested
+        expected_call_count = 1 if validate else 0
+        assert m.call_count == expected_call_count
+
+    @pytest.mark.parametrize("validate", [True, False])
+    def test_execute_with_pg_validation(
+        self,
+        requests_mock,
+        cube_add: DataCube,
+        validate,
+    ):
+        """The DataCube should pass through request for the validation to the
+        connection and the validation endpoint should only be called when
+        validation was requested.
+        """
+        m = requests_mock.post(API_URL + "/validation", json={"errors": []})
+        requests_mock.post(API_URL + "/jobs", status_code=201, content=self._post_jobs_handler_json)
+        requests_mock.post(API_URL + "/result", content=self._post_result_handler_json)
+
+        cube_add.execute(validate=validate)
+
+        # Validation should be called if and only if it was requested
+        expected_call_count = 1 if validate else 0
+        assert m.call_count == expected_call_count
+
+    @pytest.mark.parametrize("validate", [True, False])
+    def test_execute_batch_with_pg_validation(
+        self,
+        requests_mock,
+        cube_add: DataCube,
+        validate,
+    ):
+        """The DataCube should pass through request for the validation to the
+        connection and the validation endpoint should only be called when
+        validation was requested.
+        """
+        m = requests_mock.post(API_URL + "/validation", json={"errors": []})
+        requests_mock.post(API_URL + "/jobs", status_code=201, content=self._post_jobs_handler_json)
+        requests_mock.post(API_URL + f"/jobs/{self.JOB_ID}/results", status_code=202)
+        job_metadata = {
+            "id": self.JOB_ID,
+            "title": f"Job {self.JOB_ID,}",
+            "description": f"Job {self.JOB_ID,}",
+            "process": self.PROCESS_GRAPH_DICT,
+            "status": "finished",
+            "created": "2017-01-01T09:32:12Z",
+            "links": [],
+        }
+        requests_mock.get(API_URL + f"/jobs/{self.JOB_ID}", status_code=200, json=job_metadata)
+
+        cube_add.execute_batch(validate=validate)
+
+        # Validation should be called if and only if it was requested
+        expected_call_count = 1 if validate else 0
+        assert m.call_count == expected_call_count