fix backward support of output_log.* resolved to output_log.txt due t…

…o now more explicit format extraction

CHANGES.rst

@@ -14,6 +14,10 @@ Changes:
 - Add ``geotiff`` format type support via `PyWPS` (`#100 <https://github.com/crim-ca/weaver/issues/100>`_).
 - Make WPS status check more resilient to failing WPS outputs location not found in case the directory path can be
   resolved to a valid local file representing the XML status (i.e.: don't depend as much on the HTTP WPS output route).
+- Ensure backward support of generic/default ``text/plain`` I/O when extracted from a referenced WPS-1/2 XML remote
+  process which provides insufficient format details. For CWL output generated from it, replace the glob pattern to
+  match anything (``<id>.*``) instead of ``<id>.txt`` extracted from ``text/plain`` to simulate MIME-type as ``*/*``.
+  Issue log warning message for future use cases.
 
 Fixes:
 ------

tests/functional/test_builtin.py

@@ -37,7 +37,7 @@ def setUp(self):
             "weaver.wps": True,
             "weaver.wps_output": True,
             "weaver.wps_output_path": "/wpsoutputs",
-            "weaver.wps_output_dir": "/tmp",
+            "weaver.wps_output_dir": "/tmp",  # nosec: B108 # don't care hardcoded for test
             "weaver.wps_path": "/ows/wps",
             "weaver.wps_restapi_path": "/",
         }

tests/processes/test_wps_package.py

@@ -9,7 +9,7 @@
 from weaver.formats import CONTENT_TYPE_APP_JSON, CONTENT_TYPE_APP_NETCDF, CONTENT_TYPE_APP_XML, CONTENT_TYPE_TEXT_PLAIN
 from weaver.processes.wps_package import (
     WPS_LITERAL,
-    DefaultFormat,
+    DEFAULT_FORMAT,
     _are_different_and_set,
     _is_cwl_array_type,
     _is_cwl_enum_type,
@@ -356,16 +356,16 @@ def assert_formats_equal_any_order(format_result, format_expect):
 
 def test_merge_io_formats_no_wps():
     wps_fmt = []
-    cwl_fmt = [DefaultFormat]
+    cwl_fmt = [DEFAULT_FORMAT]
     res_fmt = _merge_io_formats(wps_fmt, cwl_fmt)
     assert isinstance(res_fmt, list)
     assert len(res_fmt) == 1
-    assert res_fmt[0] is DefaultFormat
+    assert res_fmt[0] is DEFAULT_FORMAT
 
 
 def test_merge_io_formats_with_wps_and_default_cwl():
     wps_fmt = [Format(CONTENT_TYPE_APP_NETCDF)]
-    cwl_fmt = [DefaultFormat]
+    cwl_fmt = [DEFAULT_FORMAT]
     res_fmt = _merge_io_formats(wps_fmt, cwl_fmt)
     assert isinstance(res_fmt, list)
     assert_formats_equal_any_order(res_fmt, [Format(CONTENT_TYPE_APP_NETCDF)])

weaver/processes/wps_package.py

@@ -122,7 +122,7 @@
 
 __all__ = [
     "PACKAGE_EXTENSIONS",
-    "DefaultFormat",
+    "DEFAULT_FORMAT",
     "WpsPackage",
     "get_process_definition",
     "get_process_location",
@@ -184,9 +184,9 @@
 WPS_FIELD_FORMAT = ["formats", "supported_formats", "supported_values", "default"]
 
 # default format if missing (minimal requirement of one)
-DefaultFormat = Format(mime_type=CONTENT_TYPE_TEXT_PLAIN)   # pylint: disable=C0103,invalid-name
+DEFAULT_FORMAT = Format(mime_type=CONTENT_TYPE_TEXT_PLAIN)
 DEFAULT_FORMAT_MISSING = "__DEFAULT_FORMAT_MISSING__"
-setattr(DefaultFormat, DEFAULT_FORMAT_MISSING, True)
+setattr(DEFAULT_FORMAT, DEFAULT_FORMAT_MISSING, True)
 
 
 def get_status_location_log_path(status_location, out_dir=None):
@@ -663,7 +663,7 @@ def _cwl2wps_io(io_info, io_select):
         else:
             # we need to minimally add 1 format, otherwise empty list is evaluated as None by pywps
             # when "supported_formats" is None, the process's json property raises because of it cannot iterate formats
-            kw["supported_formats"] = [DefaultFormat]
+            kw["supported_formats"] = [DEFAULT_FORMAT]
             kw["mode"] = MODE.NONE  # don't validate anything as default is only raw text
         if is_output:
             if io_type == "Directory":
@@ -870,7 +870,7 @@ def _json2wps_io(io_info, io_select):
         if io_type in (WPS_REFERENCE, WPS_COMPLEX):
             io_info.pop("data_type", None)
             if "supported_formats" not in io_info:
-                io_info["supported_formats"] = [DefaultFormat]
+                io_info["supported_formats"] = [DEFAULT_FORMAT]
             if ("max_occurs", "unbounded") in io_info.items():
                 io_info["max_occurs"] = PACKAGE_ARRAY_MAX_SIZE
             io_info.pop("supported_values", None)
@@ -940,7 +940,7 @@ def _wps2json_io(io_wps):
     if io_wps_json["type"] == WPS_COMPLEX:
         # FIXME: should we store 'None' in db instead of empty string when missing "encoding", "schema", etc. ?
         if "formats" not in io_wps_json or not len(io_wps_json["formats"]):
-            io_wps_json["formats"] = [DefaultFormat.json]
+            io_wps_json["formats"] = [DEFAULT_FORMAT.json]
         for io_format in io_wps_json["formats"]:
             transform_json(io_format, rename=rename, replace_values=replace_values, replace_func=replace_func)
 
@@ -1054,14 +1054,14 @@ def _merge_io_formats(wps_formats, cwl_formats):
     provided as input. In this case, *only* `WPS` formats are kept.
 
     In the event that ``DEFAULT_FORMAT_MISSING`` was written to the `CWL` formats and that no `WPS` format was
-    specified, the ``DefaultFormat`` is returned.
+    specified, the :py:data:`DEFAULT_FORMAT` is returned.
 
     :raises PackageTypeError: if inputs are invalid format lists
     """
     if not (isinstance(wps_formats, (list, tuple, set)) and isinstance(cwl_formats, (list, tuple, set))):
         raise PackageTypeError("Cannot merge formats definitions with invalid lists.")
     if not len(wps_formats):
-        wps_formats = [DefaultFormat]
+        wps_formats = [DEFAULT_FORMAT]
     if len(cwl_formats) == 1 and _get_field(cwl_formats[0], DEFAULT_FORMAT_MISSING) is True:
         return wps_formats
 
@@ -1410,7 +1410,14 @@ def _get_cwl_fmt_details(wps_fmt):
                 break
         if cwl_io_fmt:
             cwl_io["format"] = cwl_io_fmt
-
+        # for backward compatibility with deployed processes, consider text/plan as 'any' for glob pattern
+        cwl_io_txt = get_extension(CONTENT_TYPE_TEXT_PLAIN)
+        if cwl_io_ext == cwl_io_txt:
+            cwl_io_any = get_extension(CONTENT_TYPE_ANY)
+            LOGGER.warning("Replacing '%s' [%s] to generic '%s' [%s] glob pattern. "
+                           "More explicit format could be considered for %s '%s'.",
+                           CONTENT_TYPE_TEXT_PLAIN, cwl_io_txt, CONTENT_TYPE_ANY, cwl_io_any, io_select, wps_io_id)
+            cwl_io_ext = cwl_io_any
         if io_select == WPS_OUTPUT:
             # FIXME: (?) how to specify the 'name' part of the glob (using the "id" value for now)
             cwl_io["outputBinding"] = {