functional tests

CHANGES.rst

@@ -12,10 +12,17 @@ Changes
 
 Changes:
 --------
-- No change.
+- Support `CWL` definition for ``ScatterFeatureRequirement`` for `Workflow` parallel step distribution of an
+  input array (resolves `#105 <https://github.com/crim-ca/weaver/issues/105>`_).
+- Add formatter and better logging details when executing ``builtin`` `Process` ``jsonarray2netcdf``.
+- Add `OGC` Media-Type ontology for ``File`` format references within `CWL` definition.
+- Replace `EDAM` NetCDF format reference by `OGC` NetCDF Media-Type with expected ontology definitions by processes.
+  For backward compatibility, corresponding `EDAM` references will be converted to `OGC` Media-Type whenever possible.
+- Adjust ``builtin`` process ``jsonarray2netcdf`` (version ``2.0``) to employ `OGC` Media-Type for NetCDF.
 
 Fixes:
 ------
+- Fix implementation of various functional test cases for `Workflow` execution.
 - Fix ``owslib`` version with enforced ``pyproj`` dependency failing in Python 3.10
   (resolves `#459 <https://github.com/crim-ca/weaver/issues/459>`_).
 

setup.cfg

@@ -131,5 +131,11 @@ exclude_lines =
 	LOGGER.error
 	LOGGER.exception
 	LOGGER.log
+	self.logger.debug
+	self.logger.info
+	self.logger.warning
+	self.logger.error
+	self.logger.exception
+	self.logger.log
 	@overload
 	if not result.success:

tests/functional/application-packages/DockerNetCDF2Text/package.cwl

@@ -22,8 +22,11 @@ inputs:
     type: File
     inputBinding:
       position: 1
+    format: ogc:netcdf
 outputs:
   output_txt:
     type: File
     outputBinding:
       glob: "*.txt"
+$namespaces:
+  ogc: "http://www.opengis.net/def/media-type/ogc/1.0/"

tests/functional/application-packages/WPS1JsonArray2NetCDF/deploy.yml

@@ -28,9 +28,14 @@ executionUnit:
       inputs:
         input:
           type: File
+          format: "iana:application/json"
       outputs:
         output:
           type: File[]
+          format: "ogc:netcdf"
           outputBinding:
             glob: "*.nc"
+      $namespaces:
+        iana: "https://www.iana.org/assignments/media-types/"
+        ogc: "http://www.opengis.net/def/media-type/ogc/1.0/"
 deploymentProfileName: http://www.opengis.net/profiles/eoc/wpsApplication

tests/functional/application-packages/WorkflowRESTScatterCopyNetCDF/package.cwl

@@ -1,9 +1,12 @@
+#!/usr/bin/env cwl-runner
 cwlVersion: v1.0
 class: Workflow
 requirements:
   ScatterFeatureRequirement: {}
 inputs:
-  input_json: File
+  input_json:
+    type: File
+    format: "iana:application/json"
 outputs:
   output:
     type:
@@ -25,3 +28,5 @@ steps:
       input_nc: parse/output
     out:
       - output_txt
+$namespaces:
+  iana: "https://www.iana.org/assignments/media-types/"

tests/functional/application-packages/WorkflowWPS1ScatterCopyNetCDF/package.cwl

@@ -1,9 +1,12 @@
+#!/usr/bin/env cwl-runner
 cwlVersion: v1.0
 class: Workflow
 requirements:
   ScatterFeatureRequirement: {}
 inputs:
-  input_json: File
+  input_json:
+    type: File
+    format: "iana:application/json"
 outputs:
   output:
     type:
@@ -24,3 +27,5 @@ steps:
       input_nc: parse/output
     out:
       - output_txt
+$namespaces:
+  iana: "https://www.iana.org/assignments/media-types/"

tests/functional/application-packages/WorkflowWPS1SelectCopyNetCDF/package.cwl

@@ -1,3 +1,4 @@
+#!/usr/bin/env cwl-runner
 cwlVersion: v1.0
 class: Workflow
 inputs:

tests/functional/test_workflow.py

@@ -940,14 +940,16 @@ def test_workflow_mixed_rest_builtin_wps1_docker_select_requirements(self):
             nc_refs = []
             for i in range(3):
                 nc_name = f"test-file-{i}.nc"
-                nc_refs.append(os.path.join("file://" + tmp_dir, nc_name))
+                nc_path = os.path.join(tmp_dir, nc_name)
+                nc_refs.append(f"file://{nc_path}")
                 with open(os.path.join(tmp_dir, nc_name), mode="w", encoding="utf-8") as tmp_file:
                     tmp_file.write(f"DUMMY NETCDF DATA #{i}")
             with open(os.path.join(tmp_dir, "netcdf-array.json"), mode="w", encoding="utf-8") as tmp_file:
                 json.dump(nc_refs, tmp_file)  # must match execution body
 
             def mock_tmp_input(requests_mock):
                 mocked_file_server(tmp_dir, tmp_host, self.settings, requests_mock=requests_mock)
+                mocked_wps_output(self.settings, requests_mock=requests_mock)
 
             results = self.workflow_runner(WorkflowProcesses.WORKFLOW_REST_SELECT_COPY_NETCDF,
                                            [WorkflowProcesses.APP_DOCKER_NETCDF_2_TEXT,  # indirectly needed by WPS-1
@@ -969,10 +971,6 @@ def mock_tmp_input(requests_mock):
                              message="Workflow output data should have made it through the "
                                      "workflow of different process types.")
 
-    # FIXME: implement + re-enable 'CWL_REQUIREMENT_SCATTER'
-    @pytest.mark.xfail(
-        reason="ScatterFeatureRequirement not yet supported (https://github.com/crim-ca/weaver/issues/105)"
-    )
     def test_workflow_mixed_rest_builtin_wps1_docker_scatter_requirements(self):
         """
         Test the use of multiple applications of different :term:`Process` type in a :term:`Workflow`.
@@ -981,25 +979,76 @@ def test_workflow_mixed_rest_builtin_wps1_docker_scatter_requirements(self):
             1. Convert JSON array of NetCDF references to corresponding NetCDF files
                (process registered with ``WPS1Requirement`` using WPS-1 interface of builtin ``jsonarray2netcdf``).
             2. Convert NetCDF file to raw text data dumps (using scattered applications per-file).
+
+        .. note::
+            Because ``jsonarray2netcdf`` is running in subprocess instantiated by :mod:`cwltool`, file-server
+            location cannot be mocked by the test suite. Employ local test paths as if they where already fetched.
+
+        .. seealso::
+            Inverse :term:`WPS-1` / :term:`OGC API - Processes` process references from
+            :meth:`test_workflow_mixed_wps1_builtin_rest_docker_scatter_requirements`.
+        """
+
+        with contextlib.ExitStack() as stack:
+            tmp_host = "https://mocked-file-server.com"    # must match in 'WorkflowWPS1ScatterCopyNetCDF/execute.yml'
+            tmp_dir = stack.enter_context(tempfile.TemporaryDirectory())
+            nc_refs = []
+            for i in range(3):
+                nc_name = f"test-file-{i}.nc"
+                nc_path = os.path.join(tmp_dir, nc_name)
+                nc_refs.append(f"file://{nc_path}")
+                with open(os.path.join(tmp_dir, nc_name), mode="w", encoding="utf-8") as tmp_file:
+                    tmp_file.write(f"DUMMY NETCDF DATA #{i}")
+            with open(os.path.join(tmp_dir, "netcdf-array.json"), mode="w", encoding="utf-8") as tmp_file:
+                json.dump(nc_refs, tmp_file)  # must match execution body
+
+            def mock_tmp_input(requests_mock):
+                mocked_file_server(tmp_dir, tmp_host, self.settings, requests_mock=requests_mock)
+                mocked_wps_output(self.settings, requests_mock=requests_mock)
+
+            self.workflow_runner(WorkflowProcesses.WORKFLOW_WPS1_SCATTER_COPY_NETCDF,
+                                 [WorkflowProcesses.APP_DOCKER_NETCDF_2_TEXT,  # required for reference by WPS below
+                                  WorkflowProcesses.APP_WPS1_DOCKER_NETCDF_2_TEXT],
+                                 log_full_trace=True, requests_mock_callback=mock_tmp_input)
+
+    def test_workflow_mixed_wps1_builtin_rest_docker_scatter_requirements(self):
+        """
+        Test the use of multiple applications of different :term:`Process` type in a :term:`Workflow`.
+
+        Steps:
+            1. Convert JSON array of NetCDF references to corresponding NetCDF files
+               (process registered with ``WPS1Requirement`` using WPS-1 interface of builtin ``jsonarray2netcdf``).
+            2. Convert NetCDF file to raw text data dumps (using scattered applications per-file).
+
+        .. note::
+            Because ``jsonarray2netcdf`` is running in subprocess instantiated by :mod:`cwltool`, file-server
+            location cannot be mocked by the test suite. Employ local test paths as if they where already fetched.
+
+        .. seealso::
+            Inverse :term:`WPS-1` / :term:`OGC API - Processes` process references from
+            :meth:`test_workflow_mixed_rest_builtin_wps1_docker_scatter_requirements`.
         """
 
         with contextlib.ExitStack() as stack:
-            tmp_host = "https://mocked-file-server.com"  # must match in 'Execute_WorkflowScatterCopyNestedOutDir.json'
+            tmp_host = "https://mocked-file-server.com"  # must match in 'WorkflowRESTScatterCopyNetCDF/execute.yml'
             tmp_dir = stack.enter_context(tempfile.TemporaryDirectory())
             nc_refs = []
             for i in range(3):
                 nc_name = f"test-file-{i}.nc"
-                nc_refs.append(os.path.join(tmp_host, nc_name))
+                nc_path = os.path.join(tmp_dir, nc_name)
+                nc_refs.append(f"file://{nc_path}")
                 with open(os.path.join(tmp_dir, nc_name), mode="w", encoding="utf-8") as tmp_file:
                     tmp_file.write(f"DUMMY NETCDF DATA #{i}")
             with open(os.path.join(tmp_dir, "netcdf-array.json"), mode="w", encoding="utf-8") as tmp_file:
                 json.dump(nc_refs, tmp_file)  # must match execution body
 
             def mock_tmp_input(requests_mock):
                 mocked_file_server(tmp_dir, tmp_host, self.settings, requests_mock=requests_mock)
+                mocked_wps_output(self.settings, requests_mock=requests_mock)
 
             self.workflow_runner(WorkflowProcesses.WORKFLOW_REST_SCATTER_COPY_NETCDF,
-                                 [WorkflowProcesses.APP_WPS1_DOCKER_NETCDF_2_TEXT],
+                                 [WorkflowProcesses.APP_WPS1_JSON_ARRAY_2_NETCDF,  # no need to register its builtin ref
+                                  WorkflowProcesses.APP_DOCKER_NETCDF_2_TEXT],
                                  log_full_trace=True, requests_mock_callback=mock_tmp_input)
 
     def test_workflow_docker_applications(self):

tests/functional/test_wps_package.py

@@ -41,6 +41,8 @@
     EDAM_MAPPING,
     EDAM_NAMESPACE,
     IANA_NAMESPACE,
+    OGC_MAPPING,
+    OGC_NAMESPACE,
     AcceptLanguage,
     ContentType,
     get_cwl_file_format
@@ -57,7 +59,7 @@
     from weaver.typedefs import JSON
 
 EDAM_PLAIN = EDAM_NAMESPACE + ":" + EDAM_MAPPING[ContentType.TEXT_PLAIN]
-EDAM_NETCDF = EDAM_NAMESPACE + ":" + EDAM_MAPPING[ContentType.APP_NETCDF]
+OGC_NETCDF = OGC_NAMESPACE + ":" + OGC_MAPPING[ContentType.APP_NETCDF]
 # note: x-tar cannot be mapped during CWL format resolution (not official schema),
 #       it remains explicit tar definition in WPS context
 IANA_TAR = IANA_NAMESPACE + ":" + ContentType.APP_TAR  # noqa # pylint: disable=unused-variable
@@ -2497,7 +2499,7 @@ def test_deploy_literal_and_complex_io_from_wps_xml_reference(self):
         assert isinstance(pkg["inputs"], list)
         assert pkg["inputs"][0]["id"] == "tasmax"
         assert "default" not in pkg["inputs"][0]
-        assert pkg["inputs"][0]["format"] == EDAM_NETCDF
+        assert pkg["inputs"][0]["format"] == OGC_NETCDF
         assert isinstance(pkg["inputs"][0]["type"], list), "since minOccurs=1, single value non-array must be allowed"
         assert len(pkg["inputs"][0]["type"]) == 2, "single type and array type of same base"
         assert pkg["inputs"][0]["type"][0] == "File", "since minOccurs=1, should be type directly"
@@ -2515,7 +2517,7 @@ def test_deploy_literal_and_complex_io_from_wps_xml_reference(self):
         assert isinstance(pkg["outputs"], list)
         assert pkg["outputs"][0]["id"] == "output_netcdf"
         assert "default" not in pkg["outputs"][0]
-        assert pkg["outputs"][0]["format"] == EDAM_NETCDF
+        assert pkg["outputs"][0]["format"] == OGC_NETCDF
         assert pkg["outputs"][0]["type"] == "File"
         assert pkg["outputs"][0]["outputBinding"]["glob"] == "output_netcdf/*.nc"
         assert pkg["outputs"][1]["id"] == "output_log"

tests/processes/test_convert.py

@@ -15,7 +15,7 @@
 from pywps.validator.mode import MODE
 
 from weaver.exceptions import PackageTypeError
-from weaver.formats import EDAM_MAPPING, EDAM_NAMESPACE_DEFINITION, ContentType
+from weaver.formats import OGC_MAPPING, OGC_NAMESPACE_DEFINITION, ContentType
 from weaver.processes.constants import WPS_INPUT, WPS_LITERAL, WPS_OUTPUT, ProcessSchema
 from weaver.processes.convert import _are_different_and_set  # noqa: W0212
 from weaver.processes.convert import (
@@ -102,9 +102,9 @@ def test_any2cwl_io_from_wps():
     assert cwl_io == {
         "id": "test",
         "type": "File",
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}"
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}"
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
     # retry by manually injecting the type to validate that
     # pre-resolved type can also be converted directly from object
@@ -115,10 +115,10 @@ def test_any2cwl_io_from_wps():
     assert cwl_io == {
         "id": "test",
         "type": "File",
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}",
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}",
         "default": None,
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
     wps_io.min_occurs = 10
     wps_io.max_occurs = 20
@@ -127,10 +127,10 @@ def test_any2cwl_io_from_wps():
     assert cwl_io == {
         "id": "test",
         "type": {"type": "array", "items": "File"},
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}",
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}",
         "default": None,
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
 
 class MockElementXML(dict):
@@ -161,10 +161,10 @@ def test_any2cwl_io_from_ows():
     assert cwl_io == {
         "id": "test",
         "type": "File",
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}",
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}",
         "default": None,
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
     ows_io = OWSInput(MockElementXML({}))  # skip parsing from XML, inject corresponding results directly
     ows_io.identifier = "test"
@@ -177,10 +177,10 @@ def test_any2cwl_io_from_ows():
     assert cwl_io == {
         "id": "test",
         "type": {"type": "array", "items": "File"},
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}",
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}",
         "default": None,
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
 
 def test_any2cwl_io_from_json():
@@ -196,9 +196,9 @@ def test_any2cwl_io_from_json():
     assert cwl_io == {
         "id": "test",
         "type": "File",
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}"
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}"
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
     json_io["minOccurs"] = 10
     json_io["maxOccurs"] = 20
@@ -207,9 +207,9 @@ def test_any2cwl_io_from_json():
     assert cwl_io == {
         "id": "test",
         "type": {"type": "array", "items": "File"},
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}",
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}",
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
 
 def test_any2cwl_io_from_oas():
@@ -225,9 +225,9 @@ def test_any2cwl_io_from_oas():
     assert cwl_io == {
         "id": "test",
         "type": "File",
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}"
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}"
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
     json_io["minOccurs"] = 10
     json_io["maxOccurs"] = 20
@@ -236,9 +236,9 @@ def test_any2cwl_io_from_oas():
     assert cwl_io == {
         "id": "test",
         "type": {"type": "array", "items": "File"},
-        "format": f"edam:{EDAM_MAPPING[ContentType.APP_NETCDF]}",
+        "format": f"ogc:{OGC_MAPPING[ContentType.APP_NETCDF]}",
     }
-    assert cwl_ns == EDAM_NAMESPACE_DEFINITION
+    assert cwl_ns == OGC_NAMESPACE_DEFINITION
 
 
 def test_json2wps_datatype():