Merge branch 'master' into defaultcontainerWarning

Author: mr-c

cwltool/builder.py

@@ -18,9 +18,6 @@
 from .stdfsaccess import StdFsAccess
 from .utils import aslist, get_feature, docker_windows_path_adjust, onWindows
 
-# if six.PY3:
-# AvroSchemaFromJSONData = avro.schema.SchemaFromJSONData
-# else:
 AvroSchemaFromJSONData = avro.schema.make_avsc_object
 
 CONTENT_LIMIT = 64 * 1024

cwltool/docker_uid.py renamed to cwltool/docker_id.py

@@ -2,25 +2,25 @@
 from __future__ import absolute_import
 
 import subprocess
-from typing import List, Text
+from typing import List, Text, Tuple
 
 
-def docker_vm_uid():  # type: () -> int
+def docker_vm_id():  # type: () -> Tuple[int, int]
     """
-    Returns the UID of the default docker user inside the VM
+    Returns the User ID and Group ID of the default docker user inside the VM
 
     When a host is using boot2docker or docker-machine to run docker with
     boot2docker.iso (As on Mac OS X), the UID that mounts the shared filesystem
     inside the VirtualBox VM is likely different than the user's UID on the host.
-    :return: The numeric UID (as a string) of the docker account inside
+    :return: A tuple containing numeric User ID and Group ID of the docker account inside
     the boot2docker VM
     """
     if boot2docker_running():
-        return boot2docker_uid()
+        return boot2docker_id()
     elif docker_machine_running():
-        return docker_machine_uid()
+        return docker_machine_id()
     else:
-        return None
+        return (None, None)
 
 
 def check_output_and_strip(cmd):  # type: (List[Text]) -> Text
@@ -95,23 +95,26 @@ def cmd_output_to_int(cmd):  # type: (List[Text]) -> int
     return None
 
 
-def boot2docker_uid():  # type: () -> int
+def boot2docker_id():  # type: () -> Tuple[int, int]
     """
-    Gets the UID of the docker user inside a running boot2docker vm
-    :return: the UID, or None if error (e.g. boot2docker not present or stopped)
+    Gets the UID and GID of the docker user inside a running boot2docker vm
+    :return: Tuple (UID, GID), or (None, None) if error (e.g. boot2docker not present or stopped)
     """
-    return cmd_output_to_int(['boot2docker', 'ssh', 'id', '-u'])
-
+    uid = cmd_output_to_int(['boot2docker', 'ssh', 'id', '-u'])
+    gid = cmd_output_to_int(['boot2docker', 'ssh', 'id', '-g'])
+    return (uid, gid)
 
-def docker_machine_uid():  # type: () -> int
+def docker_machine_id():  # type: () -> Tuple[int, int]
     """
     Asks docker-machine for active machine and gets the UID of the docker user
     inside the vm
-    :return: the UID, or None if error (e.g. docker-machine not present or stopped)
+    :return: tuple (UID, GID), or (None, None) if error (e.g. docker-machine not present or stopped)
     """
     machine_name = docker_machine_name()
-    return cmd_output_to_int(['docker-machine', 'ssh', machine_name, "id -u"])
+    uid = cmd_output_to_int(['docker-machine', 'ssh', machine_name, "id -u"])
+    gid = cmd_output_to_int(['docker-machine', 'ssh', machine_name, "id -g"])
+    return (uid, gid)
 
 
 if __name__ == '__main__':
-    print(docker_vm_uid())
+    print(docker_vm_id())

cwltool/draft2tool.py

@@ -120,8 +120,9 @@ def revmap_file(builder, outdir, f):
         if f["location"].startswith("file://"):
             path = convert_pathsep_to_unix(uri_file_path(f["location"]))
             revmap_f = builder.pathmapper.reversemap(path)
-            if revmap_f:
-                f["location"] = revmap_f[1]
+            if revmap_f and not builder.pathmapper.mapper(revmap_f[0]).type.startswith("Writable"):
+                f["basename"] = os.path.basename(path)
+                f["location"] = revmap_f[0]
             elif path == builder.outdir:
                 f["location"] = outdir
             elif path.startswith(builder.outdir):
@@ -202,7 +203,7 @@ def makeJobRunner(self, use_container=True):  # type: (Optional[bool]) -> JobBas
                     dockerReq = self.requirements[0]
                     if default_container == windows_default_container_id and use_container and onWindows():
                         _logger.warning(DEFAULT_CONTAINER_MSG%(windows_default_container_id, windows_default_container_id))
-    
+
         if dockerReq and use_container:
             return DockerCommandLineJob()
         else:
@@ -217,6 +218,17 @@ def makePathMapper(self, reffiles, stagedir, **kwargs):
         # type: (List[Any], Text, **Any) -> PathMapper
         return PathMapper(reffiles, kwargs["basedir"], stagedir)
 
+    def updatePathmap(self, outdir, pathmap, fn):
+        # type: (Text, PathMapper, Dict) -> None
+        if "location" in fn:
+            pathmap.update(fn["location"], pathmap.mapper(fn["location"]).resolved,
+                           os.path.join(outdir, fn["basename"]),
+                           ("Writable" if fn.get("writable") else "") + fn["class"], False)
+        for sf in fn.get("secondaryFiles", []):
+            self.updatePathmap(outdir, pathmap, sf)
+        for ls in fn.get("listing", []):
+            self.updatePathmap(os.path.join(outdir, fn["basename"]), pathmap, ls)
+
     def job(self,
             job_order,  # type: Dict[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
@@ -341,46 +353,10 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
         builder.pathmapper = self.makePathMapper(reffiles, builder.stagedir, **make_path_mapper_kwargs)
         builder.requirements = j.requirements
 
-        if _logger.isEnabledFor(logging.DEBUG):
-            _logger.debug(u"[job %s] path mappings is %s", j.name,
-                          json.dumps({p: builder.pathmapper.mapper(p) for p in builder.pathmapper.files()}, indent=4))
-
         _check_adjust = partial(check_adjust, builder)
 
         visit_class([builder.files, builder.bindings], ("File", "Directory"), _check_adjust)
 
-        if self.tool.get("stdin"):
-            with SourceLine(self.tool, "stdin", validate.ValidationException):
-                j.stdin = builder.do_eval(self.tool["stdin"])
-                reffiles.append({"class": "File", "path": j.stdin})
-
-        if self.tool.get("stderr"):
-            with SourceLine(self.tool, "stderr", validate.ValidationException):
-                j.stderr = builder.do_eval(self.tool["stderr"])
-                if os.path.isabs(j.stderr) or ".." in j.stderr:
-                    raise validate.ValidationException("stderr must be a relative path, got '%s'" % j.stderr)
-
-        if self.tool.get("stdout"):
-            with SourceLine(self.tool, "stdout", validate.ValidationException):
-                j.stdout = builder.do_eval(self.tool["stdout"])
-                if os.path.isabs(j.stdout) or ".." in j.stdout or not j.stdout:
-                    raise validate.ValidationException("stdout must be a relative path, got '%s'" % j.stdout)
-
-        if _logger.isEnabledFor(logging.DEBUG):
-            _logger.debug(u"[job %s] command line bindings is %s", j.name, json.dumps(builder.bindings, indent=4))
-
-        dockerReq = self.get_requirement("DockerRequirement")[0]
-        if dockerReq and kwargs.get("use_container"):
-            out_prefix = kwargs.get("tmp_outdir_prefix")
-            j.outdir = kwargs.get("outdir") or tempfile.mkdtemp(prefix=out_prefix)
-            tmpdir_prefix = kwargs.get('tmpdir_prefix')
-            j.tmpdir = kwargs.get("tmpdir") or tempfile.mkdtemp(prefix=tmpdir_prefix)
-            j.stagedir = tempfile.mkdtemp(prefix=tmpdir_prefix)
-        else:
-            j.outdir = builder.outdir
-            j.tmpdir = builder.tmpdir
-            j.stagedir = builder.stagedir
-
         initialWorkdir = self.get_requirement("InitialWorkDirRequirement")[0]
         j.generatefiles = {"class": "Directory", "listing": [], "basename": ""}
         if initialWorkdir:
@@ -416,6 +392,45 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
                             t["entry"]["writable"] = t.get("writable")
                         ls[i] = t["entry"]
             j.generatefiles[u"listing"] = ls
+            for l in ls:
+                self.updatePathmap(builder.outdir, builder.pathmapper, l)
+            visit_class([builder.files, builder.bindings], ("File", "Directory"), _check_adjust)
+
+        if _logger.isEnabledFor(logging.DEBUG):
+            _logger.debug(u"[job %s] path mappings is %s", j.name,
+                          json.dumps({p: builder.pathmapper.mapper(p) for p in builder.pathmapper.files()}, indent=4))
+
+        if self.tool.get("stdin"):
+            with SourceLine(self.tool, "stdin", validate.ValidationException):
+                j.stdin = builder.do_eval(self.tool["stdin"])
+                reffiles.append({"class": "File", "path": j.stdin})
+
+        if self.tool.get("stderr"):
+            with SourceLine(self.tool, "stderr", validate.ValidationException):
+                j.stderr = builder.do_eval(self.tool["stderr"])
+                if os.path.isabs(j.stderr) or ".." in j.stderr:
+                    raise validate.ValidationException("stderr must be a relative path, got '%s'" % j.stderr)
+
+        if self.tool.get("stdout"):
+            with SourceLine(self.tool, "stdout", validate.ValidationException):
+                j.stdout = builder.do_eval(self.tool["stdout"])
+                if os.path.isabs(j.stdout) or ".." in j.stdout or not j.stdout:
+                    raise validate.ValidationException("stdout must be a relative path, got '%s'" % j.stdout)
+
+        if _logger.isEnabledFor(logging.DEBUG):
+            _logger.debug(u"[job %s] command line bindings is %s", j.name, json.dumps(builder.bindings, indent=4))
+
+        dockerReq = self.get_requirement("DockerRequirement")[0]
+        if dockerReq and kwargs.get("use_container"):
+            out_prefix = kwargs.get("tmp_outdir_prefix")
+            j.outdir = kwargs.get("outdir") or tempfile.mkdtemp(prefix=out_prefix)
+            tmpdir_prefix = kwargs.get('tmpdir_prefix')
+            j.tmpdir = kwargs.get("tmpdir") or tempfile.mkdtemp(prefix=tmpdir_prefix)
+            j.stagedir = tempfile.mkdtemp(prefix=tmpdir_prefix)
+        else:
+            j.outdir = builder.outdir
+            j.tmpdir = builder.tmpdir
+            j.stagedir = builder.stagedir
 
         inplaceUpdateReq = self.get_requirement("http://commonwl.org/cwltool#InplaceUpdateRequirement")[0]
 

cwltool/job.py

@@ -19,7 +19,7 @@
 from .utils import copytree_with_merge, docker_windows_path_adjust, onWindows
 from . import docker
 from .builder import Builder
-from .docker_uid import docker_vm_uid
+from .docker_id import docker_vm_id
 from .errors import WorkflowException
 from .pathmapper import PathMapper
 from .process import (UnsupportedRequirement, empty_subtree, get_feature,
@@ -89,6 +89,7 @@ def deref_links(outputs):  # type: (Any) -> None
         if outputs.get("class") == "File":
             st = os.lstat(outputs["path"])
             if stat.S_ISLNK(st.st_mode):
+                outputs["basename"] = os.path.basename(outputs["path"])
                 outputs["path"] = os.readlink(outputs["path"])
         else:
             for v in outputs.values():
@@ -147,7 +148,7 @@ def _setup(self):  # type: () -> None
 
         for knownfile in self.pathmapper.files():
             p = self.pathmapper.mapper(knownfile)
-            if p.type == "File" and not os.path.isfile(p[0]):
+            if p.type == "File" and not os.path.isfile(p[0]) and p.staged:
                 raise WorkflowException(
                     u"Input file %s (at %s) not found or is not a regular "
                     "file." % (knownfile, self.pathmapper.mapper(knownfile)[0]))
@@ -390,13 +391,12 @@ def run(self, pull_image=True, rm_container=True,
         if self.stdout:
             runtime.append("--log-driver=none")
 
-        if onWindows():  # windows os dont have getuid or geteuid functions
-            euid = docker_vm_uid()
-        else:
-            euid = docker_vm_uid() or os.geteuid()
+        euid, egid = docker_vm_id()
+        if not onWindows():  # MS Windows does not have getuid() or geteuid() functions
+            euid, egid = euid or os.geteuid(), egid or os.getgid()
 
-        if kwargs.get("no_match_user", None) is False and euid is not None:
-            runtime.append(u"--user=%s" % (euid))
+        if kwargs.get("no_match_user", None) is False and (euid, egid) != (None, None):
+            runtime.append(u"--user=%d:%d" % (euid, egid))
 
         if rm_container:
             runtime.append(u"--rm")

cwltool/load_tool.py

@@ -22,6 +22,7 @@
 from . import process, update
 from .errors import WorkflowException
 from .process import Process, shortname
+from .update import ALLUPDATES
 
 _logger = logging.getLogger("cwltool")
 
@@ -161,12 +162,17 @@ def validate_document(document_loader,  # type: Loader
     if "cwlVersion" in workflowobj:
         if not isinstance(workflowobj["cwlVersion"], (str, Text)):
             raise Exception("'cwlVersion' must be a string, got %s" % type(workflowobj["cwlVersion"]))
+        if workflowobj["cwlVersion"] not in list(ALLUPDATES):
+            # print out all the Supported Versions of cwlVersion
+            versions = list(ALLUPDATES) # ALLUPDATES is a dict
+            versions.sort()
+            raise ValidationException("'cwlVersion' not valid. Supported CWL versions are: \n{}".format("\n".join(versions)))
         workflowobj["cwlVersion"] = re.sub(
             r"^(?:cwl:|https://w3id.org/cwl/cwl#)", "",
             workflowobj["cwlVersion"])
     else:
-        _logger.warning("No cwlVersion found, treating this file as draft-2.")
-        workflowobj["cwlVersion"] = "draft-2"
+        raise ValidationException("No cwlVersion found."
+            "Use the following syntax in your CWL workflow to declare version: cwlVersion: <version>")
 
     if workflowobj["cwlVersion"] == "draft-2":
         workflowobj = cast(CommentedMap, cmap(update._draft2toDraft3dev1(

cwltool/pathmapper.py

@@ -249,3 +249,6 @@ def reversemap(self, target):  # type: (Text) -> Tuple[Text, Text]
             if v[1] == target:
                 return (k, v[0])
         return None
+
+    def update(self, key, resolved, target, type, stage):  # type: (Text, Text, Text, Text, bool) -> None
+        self._pathmap[key] = MapperEnt(resolved, target, type, stage)

cwltool/process.py

@@ -34,7 +34,7 @@
 from .builder import Builder
 from .errors import UnsupportedRequirement, WorkflowException
 from .pathmapper import (PathMapper, adjustDirObjs, get_listing,
-                         normalizeFilesDirs, visit_class)
+                         normalizeFilesDirs, visit_class, trim_listing)
 from .stdfsaccess import StdFsAccess
 from .utils import aslist, get_feature, copytree_with_merge, onWindows
 
@@ -283,7 +283,7 @@ def moveIt(src, dst):
     outfiles = []  # type: List[Dict[Text, Any]]
     collectFilesAndDirs(outputObj, outfiles)
     pm = PathMapper(outfiles, "", outdir, separateDirs=False)
-    stageFiles(pm, stageFunc=moveIt,symLink=False)
+    stageFiles(pm, stageFunc=moveIt, symLink=False)
 
     def _check_adjust(f):
         f["location"] = file_uri(pm.mapper(f["location"])[1])
@@ -386,12 +386,13 @@ def fillInDefaults(inputs, job):
     # type: (List[Dict[Text, Text]], Dict[Text, Union[Dict[Text, Any], List, Text]]) -> None
     for e, inp in enumerate(inputs):
         with SourceLine(inputs, e, WorkflowException):
-            if shortname(inp[u"id"]) in job:
-                pass
-            elif shortname(inp[u"id"]) not in job and u"default" in inp:
-                job[shortname(inp[u"id"])] = copy.copy(inp[u"default"])
-            elif shortname(inp[u"id"]) not in job and aslist(inp[u"type"])[0] == u"null":
+            fieldname = shortname(inp[u"id"])
+            if job.get(fieldname) is not None:
                 pass
+            elif job.get(fieldname) is None and u"default" in inp:
+                job[fieldname] = copy.copy(inp[u"default"])
+            elif job.get(fieldname) is None and u"null" in aslist(inp[u"type"]):
+                job[fieldname] = None
             else:
                 raise WorkflowException("Missing required input parameter `%s`" % shortname(inp["id"]))
 

cwltool/sandboxjs.py

@@ -42,7 +42,7 @@ def check_js_threshold_version(working_alias):
     """
     # parse nodejs version into int Tuple: 'v4.2.6\n' -> [4, 2, 6]
     current_version_str = subprocess.check_output(
-        [working_alias, "-v"]).decode('ascii')
+        [working_alias, "-v"]).decode('utf-8')
 
     current_version = [int(v) for v in current_version_str.strip().strip('v').split('.')]
     minimum_node_version = [int(v) for v in minimum_node_version_str.split('.')]
@@ -64,7 +64,7 @@ def new_js_proc():
     trynodes = ("nodejs", "node")
     for n in trynodes:
         try:
-            if subprocess.check_output([n, "--eval", "process.stdout.write('t')"]) != "t":
+            if subprocess.check_output([n, "--eval", "process.stdout.write('t')"]).decode('utf-8') != "t":
                 continue
             else:
                 nodejs = subprocess.Popen([n, "--eval", nodecode],
@@ -87,9 +87,11 @@ def new_js_proc():
             nodeimg = "node:slim"
             global have_node_slim
             if not have_node_slim:
-                dockerimgs = subprocess.check_output(["docker", "images", nodeimg]).decode('utf-8')
+                dockerimgs = subprocess.check_output(["docker", "images", "-q", nodeimg]).decode('utf-8')
+                # if output is an empty string
                 if len(dockerimgs.split("\n")) <= 1:
-                    nodejsimg = subprocess.check_output(["docker", "pull", nodeimg])
+                    # pull node:slim docker container
+                    nodejsimg = subprocess.check_output(["docker", "pull", nodeimg]).decode('utf-8')
                     _logger.info("Pulled Docker image %s %s", nodeimg, nodejsimg)
                 have_node_slim = True
             nodejs = subprocess.Popen(["docker", "run",
@@ -134,7 +136,8 @@ def execjs(js, jslib, timeout=None, debug=False):  # type: (Union[Mapping, Text]
 
     killed = []
 
-    def term():
+    """ Kill the node process if it exceeds timeout limit"""
+    def terminate():
         try:
             killed.append(True)
             nodejs.kill()
@@ -144,7 +147,7 @@ def term():
     if timeout is None:
         timeout = 20
 
-    tm = threading.Timer(timeout, term)
+    tm = threading.Timer(timeout, terminate)
     tm.start()
 
     stdin_buf = BytesIO((json.dumps(fn) + "\n").encode('utf-8'))
@@ -154,7 +157,8 @@ def term():
     rselect = [nodejs.stdout, nodejs.stderr]  # type: List[BytesIO]
     wselect = [nodejs.stdin]  # type: List[BytesIO]
 
-    # On windows system standard input/output are not handled properly by select  module(modules like  pywin32, msvcrt, gevent don't work either)
+    # On windows system standard input/output are not handled properly by select module
+    # (modules like  pywin32, msvcrt, gevent don't work either)
     if sys.platform=='win32':
         READ_BYTES_SIZE = 512
 

cwltool/schemas/.gitattributes

@@ -0,0 +1,4 @@
+*.txt text eol=lf
+*.fastq text eol=lf
+*.fai text eol=lf
+*.fasta text eol=lf