Merge branch 'master' into deprecate_drafts

Author: Anton Khodak

.travis.yml

@@ -1,5 +1,6 @@
 sudo: false
 language: python
+cache: pip
 python:
   - 2.7
   - 3.3
@@ -10,9 +11,13 @@ os:
   - linux
 install:
   - pip install tox-travis
+jobs:
+  include:
+    - stage: release-test
+      script: RELEASE_SKIP=head ./release-test.sh
 script: tox
 branches:
   only:
     - master
 notifications:
-  email: false
+  email: false

Makefile

@@ -56,7 +56,7 @@ install: FORCE
 dist: dist/${MODULE}-$(VERSION).tar.gz
 
 dist/${MODULE}-$(VERSION).tar.gz: $(SOURCES)
-	./setup.py sdist
+	./setup.py sdist bdist_wheel
 
 ## clean       : clean up all temporary / machine-generated files
 clean: FORCE

appveyor.yml

@@ -31,6 +31,7 @@ install:
   - "python -c \"import struct; print(struct.calcsize('P') * 8)\""
 
 build_script:
+  - "%CMD_IN_ENV% pip install -U setuptools pip"
   - "%CMD_IN_ENV% pip install ."
 
 

cwltool/argparser.py

@@ -78,6 +78,7 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
             lead_pos = []
         bindings = []  # type: List[Dict[Text,Text]]
         binding = None  # type: Dict[Text,Any]
+        value_from_expression = False
         if "inputBinding" in schema and isinstance(schema["inputBinding"], dict):
             binding = copy.copy(schema["inputBinding"])
 
@@ -87,29 +88,33 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
                 binding["position"] = aslist(lead_pos) + [0] + aslist(tail_pos)
 
             binding["datum"] = datum
+            if "valueFrom" in binding:
+                value_from_expression = True
 
         # Handle union types
         if isinstance(schema["type"], list):
-            for t in schema["type"]:
-                if isinstance(t, (str, Text)) and self.names.has_name(t, ""):
-                    avsc = self.names.get_name(t, "")
-                elif isinstance(t, dict) and "name" in t and self.names.has_name(t["name"], ""):
-                    avsc = self.names.get_name(t["name"], "")
-                else:
-                    avsc = AvroSchemaFromJSONData(t, self.names)
-                if validate.validate(avsc, datum):
-                    schema = copy.deepcopy(schema)
-                    schema["type"] = t
-                    return self.bind_input(schema, datum, lead_pos=lead_pos, tail_pos=tail_pos)
-            raise validate.ValidationException(u"'%s' is not a valid union %s" % (datum, schema["type"]))
+            if not value_from_expression:
+                for t in schema["type"]:
+                    if isinstance(t, (str, Text)) and self.names.has_name(t, ""):
+                        avsc = self.names.get_name(t, "")
+                    elif isinstance(t, dict) and "name" in t and self.names.has_name(t["name"], ""):
+                        avsc = self.names.get_name(t["name"], "")
+                    else:
+                        avsc = AvroSchemaFromJSONData(t, self.names)
+                    if validate.validate(avsc, datum):
+                        schema = copy.deepcopy(schema)
+                        schema["type"] = t
+                        return self.bind_input(schema, datum, lead_pos=lead_pos, tail_pos=tail_pos)
+                raise validate.ValidationException(u"'%s' is not a valid union %s" % (datum, schema["type"]))
         elif isinstance(schema["type"], dict):
-            st = copy.deepcopy(schema["type"])
-            if binding and "inputBinding" not in st and st["type"] == "array" and "itemSeparator" not in binding:
-                st["inputBinding"] = {}
-            for k in ("secondaryFiles", "format", "streamable"):
-                if k in schema:
-                    st[k] = schema[k]
-            bindings.extend(self.bind_input(st, datum, lead_pos=lead_pos, tail_pos=tail_pos))
+            if not value_from_expression:
+                st = copy.deepcopy(schema["type"])
+                if binding and "inputBinding" not in st and st["type"] == "array" and "itemSeparator" not in binding:
+                    st["inputBinding"] = {}
+                for k in ("secondaryFiles", "format", "streamable"):
+                    if k in schema:
+                        st[k] = schema[k]
+                bindings.extend(self.bind_input(st, datum, lead_pos=lead_pos, tail_pos=tail_pos))
         else:
             if schema["type"] in self.schemaDefs:
                 schema = self.schemaDefs[schema["type"]]
@@ -212,15 +217,18 @@ def generate_arg(self, binding):  # type: (Dict[Text,Any]) -> List[Text]
 
         prefix = binding.get("prefix")
         sep = binding.get("separate", True)
+        if prefix is None and not sep:
+            with SourceLine(binding, "separate", WorkflowException, _logger.isEnabledFor(logging.DEBUG)):
+                raise WorkflowException("'separate' option can not be specified without prefix")
 
         l = []  # type: List[Dict[Text,Text]]
         if isinstance(value, list):
-            if binding.get("itemSeparator"):
+            if binding.get("itemSeparator") and value:
                 l = [binding["itemSeparator"].join([self.tostr(v) for v in value])]
             elif binding.get("valueFrom"):
                 value = [self.tostr(v) for v in value]
                 return ([prefix] if prefix else []) + value
-            elif prefix:
+            elif prefix and value:
                 return [prefix]
             else:
                 return []
@@ -244,8 +252,8 @@ def generate_arg(self, binding):  # type: (Dict[Text,Any]) -> List[Text]
 
         return [a for a in args if a is not None]
 
-    def do_eval(self, ex, context=None, pull_image=True, recursive=False):
-        # type: (Union[Dict[Text, Text], Text], Any, bool, bool) -> Any
+    def do_eval(self, ex, context=None, pull_image=True, recursive=False, strip_whitespace=True):
+        # type: (Union[Dict[Text, Text], Text], Any, bool, bool, bool) -> Any
         if recursive:
             if isinstance(ex, dict):
                 return {k: self.do_eval(v, context, pull_image, recursive) for k, v in iteritems(ex)}
@@ -260,4 +268,5 @@ def do_eval(self, ex, context=None, pull_image=True, recursive=False):
                                   timeout=self.timeout,
                                   debug=self.debug,
                                   js_console=self.js_console,
-                                  force_docker_pull=self.force_docker_pull)
+                                  force_docker_pull=self.force_docker_pull,
+                                  strip_whitespace=strip_whitespace)

cwltool/builder.py

@@ -150,8 +150,8 @@ def revmap_file(builder, outdir, f):
                                     u"file pass through." % (path, builder.outdir))
         return f
 
-
-    raise WorkflowException(u"Output File object is missing both `location` and `path` fields: %s" % f)
+    raise WorkflowException(u"Output File object is missing both 'location' "
+                            "and 'path' fields: %s" % f)
 
 
 class CallbackJob(object):
@@ -382,7 +382,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
             else:
                 for t in initialWorkdir["listing"]:
                     if "entry" in t:
-                        et = {u"entry": builder.do_eval(t["entry"])}
+                        et = {u"entry": builder.do_eval(t["entry"], strip_whitespace=False)}
                         if "entryname" in t:
                             et["entryname"] = builder.do_eval(t["entryname"])
                         else:

cwltool/command_line_tool.py

@@ -0,0 +1,171 @@
+import logging
+import tempfile
+import threading
+
+import os
+from abc import ABCMeta, abstractmethod
+
+from typing import Dict, Text, Any, Tuple, Set, List
+
+from .builder import Builder
+from .errors import WorkflowException
+from .mutation import MutationManager
+from .job import JobBase
+from .process import relocateOutputs, cleanIntermediate, Process
+from . import loghandler
+
+_logger = logging.getLogger("cwltool")
+
+class JobExecutor(object):
+    __metaclass__ = ABCMeta
+
+    def __init__(self):
+        # type: (...) -> None
+        self.final_output = []  # type: List
+        self.final_status = []  # type: List
+        self.output_dirs = set()  # type: Set
+
+    def __call__(self, *args, **kwargs):
+        return self.execute(*args, **kwargs)
+
+    def output_callback(self, out, processStatus):
+        self.final_status.append(processStatus)
+        self.final_output.append(out)
+
+    @abstractmethod
+    def run_jobs(self,
+                 t,  # type: Process
+                 job_order_object,  # type: Dict[Text, Any]
+                 logger,
+                 **kwargs  # type: Any
+                 ):
+        pass
+
+    def execute(self, t,  # type: Process
+                job_order_object,  # type: Dict[Text, Any]
+                logger=_logger,
+                **kwargs  # type: Any
+                ):
+        # type: (...) -> Tuple[Dict[Text, Any], Text]
+
+        if "basedir" not in kwargs:
+            raise WorkflowException("Must provide 'basedir' in kwargs")
+
+        finaloutdir = os.path.abspath(kwargs.get("outdir")) if kwargs.get("outdir") else None
+        kwargs["outdir"] = tempfile.mkdtemp(prefix=kwargs["tmp_outdir_prefix"]) if kwargs.get(
+            "tmp_outdir_prefix") else tempfile.mkdtemp()
+        self.output_dirs.add(kwargs["outdir"])
+        kwargs["mutation_manager"] = MutationManager()
+
+        jobReqs = None
+        if "cwl:requirements" in job_order_object:
+            jobReqs = job_order_object["cwl:requirements"]
+        elif ("cwl:defaults" in t.metadata and "cwl:requirements" in t.metadata["cwl:defaults"]):
+            jobReqs = t.metadata["cwl:defaults"]["cwl:requirements"]
+        if jobReqs:
+            for req in jobReqs:
+                t.requirements.append(req)
+
+        self.run_jobs(t, job_order_object, logger, **kwargs)
+
+        if self.final_output and self.final_output[0] and finaloutdir:
+            self.final_output[0] = relocateOutputs(self.final_output[0], finaloutdir,
+                                                   self.output_dirs, kwargs.get("move_outputs"),
+                                                   kwargs["make_fs_access"](""),
+                                                   kwargs["compute_checksum"])
+
+        if kwargs.get("rm_tmpdir"):
+            cleanIntermediate(self.output_dirs)
+
+        if self.final_output and self.final_status:
+            return (self.final_output[0], self.final_status[0])
+        else:
+            return (None, "permanentFail")
+
+
+class SingleJobExecutor(JobExecutor):
+    def run_jobs(self,
+                 t,  # type: Process
+                 job_order_object,  # type: Dict[Text, Any]
+                 logger,
+                 **kwargs  # type: Any
+                 ):
+        jobiter = t.job(job_order_object,
+                        self.output_callback,
+                        **kwargs)
+
+        try:
+            for r in jobiter:
+                if r:
+                    builder = kwargs.get("builder", None)  # type: Builder
+                    if builder is not None:
+                        r.builder = builder
+                    if r.outdir:
+                        self.output_dirs.add(r.outdir)
+                    r.run(**kwargs)
+                else:
+                    logger.error("Workflow cannot make any more progress.")
+                    break
+        except WorkflowException:
+            raise
+        except Exception as e:
+            logger.exception("Got workflow error")
+            raise WorkflowException(Text(e))
+
+
+class MultithreadedJobExecutor(JobExecutor):
+    def __init__(self):
+        super(MultithreadedJobExecutor, self).__init__()
+        self.threads = set()
+        self.exceptions = []
+
+    def run_job(self,
+                job, # type:  JobBase
+                **kwargs  # type:  Any
+                ):
+        # type: (...) -> None
+        def runner():
+            try:
+                job.run(**kwargs)
+            except WorkflowException as e:
+                self.exceptions.append(e)
+            except Exception as e:
+                self.exceptions.append(WorkflowException(Text(e)))
+
+            self.threads.remove(thread)
+
+        thread = threading.Thread(target=runner)
+        thread.daemon = True
+        self.threads.add(thread)
+        thread.start()
+
+    def wait_for_next_completion(self):  # type: () -> None
+        if self.exceptions:
+            raise self.exceptions[0]
+
+    def run_jobs(self,
+                 t,  # type: Process
+                 job_order_object,  # type: Dict[Text, Any]
+                 logger,
+                 **kwargs  # type: Any
+                 ):
+
+        jobiter = t.job(job_order_object, self.output_callback, **kwargs)
+
+        for r in jobiter:
+            if r:
+                builder = kwargs.get("builder", None)  # type: Builder
+                if builder is not None:
+                    r.builder = builder
+                if r.outdir:
+                    self.output_dirs.add(r.outdir)
+                self.run_job(r, **kwargs)
+            else:
+                if len(self.threads):
+                    self.wait_for_next_completion()
+                else:
+                    logger.error("Workflow cannot make any more progress.")
+                    break
+
+        while len(self.threads) > 0:
+            self.wait_for_next_completion()

cwltool/executors.py

@@ -173,9 +173,10 @@ def evaluator(ex, jslib, obj, fullJS=False, timeout=None, force_docker_pull=Fals
 
 def interpolate(scan, rootvars,
                 timeout=None, fullJS=None, jslib="", force_docker_pull=False,
-                debug=False, js_console=False):
-    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text], bool, bool, bool) -> JSON
-    scan = scan.strip()
+                debug=False, js_console=False, strip_whitespace=True):
+    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text], bool, bool, bool, bool) -> JSON
+    if strip_whitespace:
+        scan = scan.strip()
     parts = []
     w = scanner(scan)
     while w:
@@ -185,7 +186,7 @@ def interpolate(scan, rootvars,
             e = evaluator(scan[w[0] + 1:w[1]], jslib, rootvars, fullJS=fullJS,
                           timeout=timeout, force_docker_pull=force_docker_pull,
                           debug=debug, js_console=js_console)
-            if w[0] == 0 and w[1] == len(scan):
+            if w[0] == 0 and w[1] == len(scan) and len(parts) <= 1:
                 return e
             leaf = json.dumps(e, sort_keys=True)
             if leaf[0] == '"':
@@ -202,8 +203,9 @@ def interpolate(scan, rootvars,
 
 
 def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
-            context=None, pull_image=True, timeout=None, force_docker_pull=False, debug=False, js_console=False):
-    # type: (Union[dict, AnyStr], Dict[Text, Union[Dict, List, Text]], List[Dict[Text, Any]], Text, Text, Dict[Text, Union[int, Text]], Any, bool, int, bool, bool, bool) -> Any
+            context=None, pull_image=True, timeout=None, force_docker_pull=False,
+            debug=False, js_console=False, strip_whitespace=True):
+    # type: (Union[dict, AnyStr], Dict[Text, Union[Dict, List, Text]], List[Dict[Text, Any]], Text, Text, Dict[Text, Union[int, Text]], Any, bool, int, bool, bool, bool, bool) -> Any
 
     runtime = copy.copy(resources)
     runtime["tmpdir"] = docker_windows_path_adjust(tmpdir)
@@ -231,7 +233,8 @@ def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
                                jslib=jslib,
                                force_docker_pull=force_docker_pull,
                                debug=debug,
-                               js_console=js_console)
+                               js_console=js_console,
+                               strip_whitespace=strip_whitespace)
 
         except Exception as e:
             raise WorkflowException("Expression evaluation error:\n%s" % e)