Merge branch 'master' into sfc

Author: tetron

.gitignore

@@ -1,3 +1,7 @@
+# Generated during tests
+pytestdebug.log
+tmp/
+
 # Python temps
 __pycache__/
 *.py[cod]

.travis.yml

@@ -34,6 +34,11 @@ jobs:
       env:
        - version=v1.1
       script: ${TRAVIS_BUILD_DIR}/travis.bash
+    - python: "3.8"
+      name: "CWL v1.2 conformance tests"
+      env:
+       - version=v1.2.0-dev1
+      script: ${TRAVIS_BUILD_DIR}/travis.bash
     - python: "3.7"
       script: RELEASE_SKIP=head ${TRAVIS_BUILD_DIR}/release-test.sh
 script: tox

MANIFEST.in

@@ -24,6 +24,10 @@ include cwltool/schemas/v1.1.0-dev1/*.yml
 include cwltool/schemas/v1.1.0-dev1/*.md
 include cwltool/schemas/v1.1.0-dev1/salad/schema_salad/metaschema/*.yml
 include cwltool/schemas/v1.1.0-dev1/salad/schema_salad/metaschema/*.md
+include cwltool/schemas/v1.2.0-dev1/*.yml
+include cwltool/schemas/v1.2.0-dev1/*.md
+include cwltool/schemas/v1.2.0-dev1/salad/schema_salad/metaschema/*.yml
+include cwltool/schemas/v1.2.0-dev1/salad/schema_salad/metaschema/*.md
 include cwltool/cwlNodeEngine.js
 include cwltool/cwlNodeEngineJSConsole.js
 include cwltool/cwlNodeEngineWithContext.js

cwltool/argparser.py

@@ -799,6 +799,7 @@ def generate_parser(
     records: List[str],
     input_required: bool = True,
 ) -> argparse.ArgumentParser:
+    toolparser.description = tool.tool.get("doc", None)
     toolparser.add_argument("job_order", nargs="?", help="Job input json file")
     namemap["job_order"] = "job_order"
 

cwltool/checker.py

@@ -166,8 +166,10 @@ def static_checker(
     for parm in src_parms:
         src_dict[parm["id"]] = parm
 
-    step_inputs_val = check_all_types(src_dict, step_inputs, "source")
-    workflow_outputs_val = check_all_types(src_dict, workflow_outputs, "outputSource")
+    step_inputs_val = check_all_types(src_dict, step_inputs, "source", param_to_step)
+    workflow_outputs_val = check_all_types(
+        src_dict, workflow_outputs, "outputSource", param_to_step
+    )
 
     warnings = step_inputs_val["warning"] + workflow_outputs_val["warning"]
     exceptions = step_inputs_val["exception"] + workflow_outputs_val["exception"]
@@ -212,18 +214,21 @@ def static_checker(
                 "%s\n%s" % (msg1, bullets([msg3, msg4, msg5], "  "))
             )
         elif sink.get("not_connected"):
-            msg = SourceLine(sink, "type").makeError(
-                "'%s' is not an input parameter of %s, expected %s"
-                % (
-                    shortname(sink["id"]),
-                    param_to_step[sink["id"]]["run"],
-                    ", ".join(
-                        shortname(s["id"])
-                        for s in param_to_step[sink["id"]]["inputs"]
-                        if not s.get("not_connected")
-                    ),
+            if not sink.get("used_by_step"):
+                msg = SourceLine(sink, "type").makeError(
+                    "'%s' is not an input parameter of %s, expected %s"
+                    % (
+                        shortname(sink["id"]),
+                        param_to_step[sink["id"]]["run"],
+                        ", ".join(
+                            shortname(s["id"])
+                            for s in param_to_step[sink["id"]]["inputs"]
+                            if not s.get("not_connected")
+                        ),
+                    )
                 )
-            )
+            else:
+                msg = ""
         else:
             msg = (
                 SourceLine(src, "type").makeError(
@@ -241,11 +246,17 @@ def static_checker(
                     "  source has linkMerge method %s" % linkMerge
                 )
 
-        warning_msgs.append(msg)
+        if warning.message is not None:
+            msg += "\n" + SourceLine(sink).makeError("  " + warning.message)
+
+        if msg:
+            warning_msgs.append(msg)
+
     for exception in exceptions:
         src = exception.src
         sink = exception.sink
         linkMerge = exception.linkMerge
+        extra_message = exception.message
         msg = (
             SourceLine(src, "type").makeError(
                 "Source '%s' of type %s is incompatible"
@@ -257,6 +268,9 @@ def static_checker(
                 % (shortname(sink["id"]), json_dumps(sink["type"]))
             )
         )
+        if extra_message is not None:
+            msg += "\n" + SourceLine(sink).makeError("  " + extra_message)
+
         if linkMerge is not None:
             msg += "\n" + SourceLine(sink).makeError(
                 "  source has linkMerge method %s" % linkMerge
@@ -278,19 +292,19 @@ def static_checker(
             exception_msgs.append(msg)
 
     all_warning_msg = strip_dup_lineno("\n".join(warning_msgs))
-    all_exception_msg = strip_dup_lineno("\n".join(exception_msgs))
+    all_exception_msg = strip_dup_lineno("\n" + "\n".join(exception_msgs))
 
-    if warnings:
+    if all_warning_msg:
         _logger.warning("Workflow checker warning:\n%s", all_warning_msg)
     if exceptions:
         raise validate.ValidationException(all_exception_msg)
 
 
-SrcSink = namedtuple("SrcSink", ["src", "sink", "linkMerge"])
+SrcSink = namedtuple("SrcSink", ["src", "sink", "linkMerge", "message"])
 
 
-def check_all_types(src_dict, sinks, sourceField):
-    # type: (Dict[str, Any], List[Dict[str, Any]], str) -> Dict[str, List[SrcSink]]
+def check_all_types(src_dict, sinks, sourceField, param_to_step):
+    # type: (Dict[str, Any], List[Dict[str, Any]], str, Dict[str, Dict[str, Any]]) -> Dict[str, List[SrcSink]]
     """
     Given a list of sinks, check if their types match with the types of their sources.
 
@@ -299,21 +313,93 @@ def check_all_types(src_dict, sinks, sourceField):
     validation = {"warning": [], "exception": []}  # type: Dict[str, List[SrcSink]]
     for sink in sinks:
         if sourceField in sink:
+
             valueFrom = sink.get("valueFrom")
+            pickValue = sink.get("pickValue")
+
+            extra_message = None
+            if pickValue is not None:
+                extra_message = "pickValue is: %s" % pickValue
+
             if isinstance(sink[sourceField], MutableSequence):
-                srcs_of_sink = [src_dict[parm_id] for parm_id in sink[sourceField]]
                 linkMerge = sink.get(
                     "linkMerge",
                     ("merge_nested" if len(sink[sourceField]) > 1 else None),
                 )
+
+                if pickValue in ["first_non_null", "only_non_null"]:
+                    linkMerge = None
+
+                srcs_of_sink = []  # type: List[Any]
+                for parm_id in sink[sourceField]:
+                    srcs_of_sink += [src_dict[parm_id]]
+                    if (
+                        is_conditional_step(param_to_step, parm_id)
+                        and pickValue is None
+                    ):
+                        validation["warning"].append(
+                            SrcSink(
+                                src_dict[parm_id],
+                                sink,
+                                linkMerge,
+                                message="Source is from conditional step, but pickValue is not used",
+                            )
+                        )
             else:
                 parm_id = sink[sourceField]
                 srcs_of_sink = [src_dict[parm_id]]
                 linkMerge = None
+
+                if pickValue is not None:
+                    validation["warning"].append(
+                        SrcSink(
+                            src_dict[parm_id],
+                            sink,
+                            linkMerge,
+                            message="pickValue is used but only a single input source is declared",
+                        )
+                    )
+
+                if is_conditional_step(param_to_step, parm_id):
+                    src_typ = srcs_of_sink[0]["type"]
+                    snk_typ = sink["type"]
+
+                    if not isinstance(src_typ, list):
+                        src_typ = [src_typ]
+                    if "null" not in src_typ:
+                        src_typ = ["null"] + src_typ
+
+                    if (
+                        "null" not in snk_typ
+                    ):  # Given our type names this works even if not a list
+                        validation["warning"].append(
+                            SrcSink(
+                                src_dict[parm_id],
+                                sink,
+                                linkMerge,
+                                message="Source is from conditional step and may produce `null`",
+                            )
+                        )
+
+                    srcs_of_sink[0]["type"] = src_typ
+
             for src in srcs_of_sink:
                 check_result = check_types(src, sink, linkMerge, valueFrom)
                 if check_result == "warning":
-                    validation["warning"].append(SrcSink(src, sink, linkMerge))
+                    validation["warning"].append(
+                        SrcSink(src, sink, linkMerge, message=extra_message)
+                    )
                 elif check_result == "exception":
-                    validation["exception"].append(SrcSink(src, sink, linkMerge))
+                    validation["exception"].append(
+                        SrcSink(src, sink, linkMerge, message=extra_message)
+                    )
+
     return validation
+
+
+def is_conditional_step(param_to_step: Dict[str, Dict[str, Any]], parm_id: str) -> bool:
+    source_step = param_to_step.get(parm_id)
+    if source_step is not None:
+        if source_step.get("when") is not None:
+            return True
+    return False

cwltool/command_line_tool.py

@@ -162,6 +162,17 @@ def job(
         yield job
 
 
+class AbstractOperation(Process):
+    def job(
+        self,
+        job_order,  # type: Mapping[str, str]
+        output_callbacks,  # type: Callable[[Any, Any], Any]
+        runtimeContext,  # type: RuntimeContext
+    ):
+        # type: (...) -> Generator[ExpressionTool.ExpressionJob, None, None]
+        raise WorkflowException("Abstract operation cannot be executed.")
+
+
 def remove_path(f):  # type: (Dict[str, Any]) -> None
     if "path" in f:
         del f["path"]

cwltool/context.py

@@ -132,7 +132,7 @@ def __init__(self, kwargs: Optional[Dict[str, Any]] = None) -> None:
         self.eval_timeout = 20  # type: float
         self.postScatterEval = (
             None
-        )  # type: Optional[Callable[[MutableMapping[str, Any]], Dict[str, Any]]]
+        )  # type: Optional[Callable[[MutableMapping[str, Any]], Optional[MutableMapping[str, Any]]]]
         self.on_error = "stop"  # type: str
         self.strict_memory_limit = False  # type: bool
 

cwltool/docker.py

@@ -1,14 +1,15 @@
 """Enables Docker software containers via the {dx-,u,}docker runtimes."""
 
 import datetime
+import csv
 import os
 import re
 import shutil
 import sys
 import tempfile
 import threading
 from distutils import spawn
-from io import open  # pylint: disable=redefined-builtin
+from io import open, StringIO  # pylint: disable=redefined-builtin
 from typing import Dict, List, MutableMapping, Optional, Set, Tuple
 
 import requests
@@ -222,11 +223,20 @@ def get_from_requirements(
     def append_volume(runtime, source, target, writable=False):
         # type: (List[str], str, str, bool) -> None
         """Add binding arguments to the runtime list."""
-        runtime.append(
-            "--volume={}:{}:{}".format(
-                docker_windows_path_adjust(source), target, "rw" if writable else "ro"
-            )
-        )
+        options = [
+            "type=bind",
+            "source=" + source,
+            "target=" + target,
+        ]
+        if not writable:
+            options.append("readonly")
+        output = StringIO()
+        csv.writer(output).writerow(options)
+        mount_arg = output.getvalue().strip()
+        runtime.append("--mount={}".format(mount_arg))
+        # Unlike "--volume", "--mount" will fail if the volume doesn't already exist.
+        if not os.path.exists(source):
+            os.mkdir(source)
 
     def add_file_or_directory_volume(
         self, runtime: List[str], volume: MapperEnt, host_outdir_tgt: Optional[str]

cwltool/executors.py

@@ -7,11 +7,22 @@
 import threading
 from abc import ABCMeta, abstractmethod
 from threading import Lock
-from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union
+from typing import (
+    Any,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Set,
+    Tuple,
+    Union,
+    MutableMapping,
+)
 
 import psutil
 
 from schema_salad.validate import ValidationException
+from schema_salad.sourceline import SourceLine
 
 from .command_line_tool import CallbackJob
 from .context import RuntimeContext, getdefault
@@ -68,6 +79,14 @@ def execute(
         if not runtime_context.basedir:
             raise WorkflowException("Must provide 'basedir' in runtimeContext")
 
+        def check_for_abstract_op(tool: MutableMapping[str, Any]) -> None:
+            if tool["class"] == "Operation":
+                raise SourceLine(tool, "class", WorkflowException).makeError(
+                    "Workflow has unrunnable abstract Operation"
+                )
+
+        process.visit(check_for_abstract_op)
+
         finaloutdir = None  # Type: Optional[str]
         original_outdir = runtime_context.outdir
         if isinstance(original_outdir, str):

cwltool/job.py

@@ -433,7 +433,9 @@ def shouldquote(x: Any) -> bool:
             _logger.info("[job %s] completed %s", self.name, processStatus)
 
         if _logger.isEnabledFor(logging.DEBUG):
-            _logger.debug("[job %s] %s", self.name, json_dumps(outputs, indent=4))
+            _logger.debug(
+                "[job %s] outputs %s", self.name, json_dumps(outputs, indent=4)
+            )
 
         if self.generatemapper is not None and runtimeContext.secret_store is not None:
             # Delete any runtime-generated files containing secrets.