Merge branch 'master' into py3_window

Author: mr-c

cwltool/builder.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 import copy
 import os
+import logging
 from typing import Any, Callable, Dict, List, Text, Type, Union
 
 import six
@@ -18,6 +19,8 @@
 from .stdfsaccess import StdFsAccess
 from .utils import aslist, get_feature, docker_windows_path_adjust, onWindows
 
+_logger = logging.getLogger("cwltool")
+
 AvroSchemaFromJSONData = avro.schema.make_avsc_object
 
 CONTENT_LIMIT = 64 * 1024
@@ -147,18 +150,25 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
                         datum["secondaryFiles"] = []
                     for sf in aslist(schema["secondaryFiles"]):
                         if isinstance(sf, dict) or "$(" in sf or "${" in sf:
-                            secondary_eval = self.do_eval(sf, context=datum)
-                            if isinstance(secondary_eval, string_types):
-                                sfpath = {"location": secondary_eval,
-                                          "class": "File"}
-                            else:
-                                sfpath = secondary_eval
-                        else:
-                            sfpath = {"location": substitute(datum["location"], sf), "class": "File"}
-                        if isinstance(sfpath, list):
-                            datum["secondaryFiles"].extend(sfpath)
+                            sfpath = self.do_eval(sf, context=datum)
                         else:
-                            datum["secondaryFiles"].append(sfpath)
+                            sfpath = substitute(datum["basename"], sf)
+                        for sfname in aslist(sfpath):
+                            found = False
+                            for d in datum["secondaryFiles"]:
+                                if not d.get("basename"):
+                                    d["basename"] = d["location"][d["location"].rindex("/")+1:]
+                                if d["basename"] == sfname:
+                                    found = True
+                            if not found:
+                                if isinstance(sfname, dict):
+                                    datum["secondaryFiles"].append(sfname)
+                                else:
+                                    datum["secondaryFiles"].append({
+                                        "location": datum["location"][0:datum["location"].rindex("/")+1]+sfname,
+                                        "basename": sfname,
+                                        "class": "File"})
+
                     normalizeFilesDirs(datum["secondaryFiles"])
 
                 def _capture_files(f):

cwltool/draft2tool.py

@@ -35,7 +35,7 @@
 ACCEPTLIST_EN_RELAXED_RE = re.compile(r".*")  # Accept anything
 ACCEPTLIST_RE = ACCEPTLIST_EN_STRICT_RE
 DEFAULT_CONTAINER_MSG="""We are on Microsoft Windows and not all components of this CWL description have a
-container specified. This means that these steps will be executed in the default container, 
+container specified. This means that these steps will be executed in the default container,
 which is %s.
 
 Note, this could affect portability if this CWL description relies on non-POSIX features
@@ -116,17 +116,26 @@ def revmap_file(builder, outdir, f):
     if not split.scheme:
         outdir = file_uri(str(outdir))
 
+    # builder.outdir is the inner (container/compute node) output directory
+    # outdir is the outer (host/storage system) output directory
+
     if "location" in 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 and not builder.pathmapper.mapper(revmap_f[0]).type.startswith("Writable"):
                 f["basename"] = os.path.basename(path)
-                f["location"] = revmap_f[0]
+                f["location"] = revmap_f[1]
             elif path == builder.outdir:
                 f["location"] = outdir
             elif path.startswith(builder.outdir):
                 f["location"] = builder.fs_access.join(outdir, path[len(builder.outdir) + 1:])
+            elif f["location"].startswith(outdir):
+                revmap_f = builder.pathmapper.reversemap(builder.fs_access.join(builder.outdir, f["location"][len(outdir) + 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[1]
         return f
 
     if "path" in f:
@@ -190,7 +199,7 @@ def __init__(self, toolpath_object, **kwargs):
         super(CommandLineTool, self).__init__(toolpath_object, **kwargs)
         self.find_default_container = kwargs.get("find_default_container", None)
 
-    def makeJobRunner(self, use_container=True):  # type: (Optional[bool]) -> JobBase
+    def makeJobRunner(self, use_container=True, **kwargs):  # type: (Optional[bool], **Any) -> JobBase
         dockerReq, _ = self.get_requirement("DockerRequirement")
         if not dockerReq and use_container:
             if self.find_default_container:
@@ -216,7 +225,8 @@ def makeJobRunner(self, use_container=True):  # type: (Optional[bool]) -> JobBas
 
     def makePathMapper(self, reffiles, stagedir, **kwargs):
         # type: (List[Any], Text, **Any) -> PathMapper
-        return PathMapper(reffiles, kwargs["basedir"], stagedir)
+        return PathMapper(reffiles, kwargs["basedir"], stagedir,
+                          separateDirs=kwargs.get("separateDirs", True))
 
     def updatePathmap(self, outdir, pathmap, fn):
         # type: (Text, PathMapper, Dict) -> None
@@ -325,9 +335,10 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
 
         reffiles = copy.deepcopy(builder.files)
 
-        j = self.makeJobRunner(kwargs.get("use_container"))
+        j = self.makeJobRunner(**kwargs)
         j.builder = builder
         j.joborder = builder.job
+        j.make_pathmapper = self.makePathMapper
         j.stdin = None
         j.stderr = None
         j.stdout = None
@@ -350,6 +361,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
         if "stagedir" in make_path_mapper_kwargs:
             make_path_mapper_kwargs = make_path_mapper_kwargs.copy()
             del make_path_mapper_kwargs["stagedir"]
+
         builder.pathmapper = self.makePathMapper(reffiles, builder.stagedir, **make_path_mapper_kwargs)
         builder.requirements = j.requirements
 
@@ -566,7 +578,12 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                         elif gb.startswith("/"):
                             raise WorkflowException("glob patterns must not start with '/'")
                         try:
+                            prefix = fs_access.glob(outdir)
                             r.extend([{"location": g,
+                                       "path": fs_access.join(builder.outdir, g[len(prefix[0])+1:]),
+                                       "basename": os.path.basename(g),
+                                       "nameroot": os.path.splitext(os.path.basename(g))[0],
+                                       "nameext": os.path.splitext(os.path.basename(g))[1],
                                        "class": "File" if fs_access.isfile(g) else "Directory"}
                                       for g in fs_access.glob(fs_access.join(outdir, gb))])
                         except (OSError, IOError) as e:
@@ -576,12 +593,14 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                             raise
 
                 for files in r:
+                    rfile = files.copy()
+                    revmap(rfile)
                     if files["class"] == "Directory":
                         ll = builder.loadListing or (binding and binding.get("loadListing"))
                         if ll and ll != "no_listing":
                             get_listing(fs_access, files, (ll == "deep_listing"))
                     else:
-                        with fs_access.open(files["location"], "rb") as f:
+                        with fs_access.open(rfile["location"], "rb") as f:
                             contents = b""
                             if binding.get("loadContents") or compute_checksum:
                                 contents = f.read(CONTENT_LIMIT)
@@ -625,28 +644,39 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                     else:
                         r = r[0]
 
-            # Ensure files point to local references outside of the run environment
-            adjustFileObjs(r, cast(  # known bug in mypy
-                # https://github.com/python/mypy/issues/797
-                Callable[[Any], Any], revmap))
-
             if "secondaryFiles" in schema:
                 with SourceLine(schema, "secondaryFiles", WorkflowException):
                     for primary in aslist(r):
                         if isinstance(primary, dict):
-                            primary["secondaryFiles"] = []
+                            primary.setdefault("secondaryFiles", [])
+                            pathprefix = primary["path"][0:primary["path"].rindex("/")+1]
                             for sf in aslist(schema["secondaryFiles"]):
                                 if isinstance(sf, dict) or "$(" in sf or "${" in sf:
                                     sfpath = builder.do_eval(sf, context=primary)
-                                    if isinstance(sfpath, string_types):
-                                        sfpath = revmap({"location": sfpath, "class": "File"})
+                                    subst = False
                                 else:
-                                    sfpath = {"location": substitute(primary["location"], sf), "class": "File"}
-
+                                    sfpath = sf
+                                    subst = True
                                 for sfitem in aslist(sfpath):
-                                    if fs_access.exists(sfitem["location"]):
+                                    if isinstance(sfitem, string_types):
+                                        if subst:
+                                            sfitem = {"path": substitute(primary["path"], sfitem)}
+                                        else:
+                                            sfitem = {"path": pathprefix+sfitem}
+                                    if "path" in sfitem and "location" not in sfitem:
+                                        revmap(sfitem)
+                                    if fs_access.isfile(sfitem["location"]):
+                                        sfitem["class"] = "File"
+                                        primary["secondaryFiles"].append(sfitem)
+                                    elif fs_access.isdir(sfitem["location"]):
+                                        sfitem["class"] = "Directory"
                                         primary["secondaryFiles"].append(sfitem)
 
+            # Ensure files point to local references outside of the run environment
+            adjustFileObjs(r, cast(  # known bug in mypy
+                # https://github.com/python/mypy/issues/797
+                Callable[[Any], Any], revmap))
+
             if not r and optional:
                 r = None
 

cwltool/job.py

@@ -132,6 +132,7 @@ def __init__(self):  # type: () -> None
         self.name = None  # type: Text
         self.command_line = None  # type: List[Text]
         self.pathmapper = None  # type: PathMapper
+        self.make_pathmapper = None  # type: Callable[..., PathMapper]
         self.generatemapper = None  # type: PathMapper
         self.collect_outputs = None  # type: Union[Callable[[Any], Any], functools.partial[Any]]
         self.output_callback = None  # type: Callable[[Any, Any], Any]
@@ -142,7 +143,7 @@ def __init__(self):  # type: () -> None
         self.stagedir = None  # type: Text
         self.inplace_update = None  # type: bool
 
-    def _setup(self):  # type: () -> None
+    def _setup(self, kwargs):  # type: (Dict) -> None
         if not os.path.exists(self.outdir):
             os.makedirs(self.outdir)
 
@@ -154,8 +155,12 @@ def _setup(self):  # type: () -> None
                     "file." % (knownfile, self.pathmapper.mapper(knownfile)[0]))
 
         if self.generatefiles["listing"]:
-            self.generatemapper = PathMapper(cast(List[Any], self.generatefiles["listing"]),
-                                             self.outdir, self.outdir, separateDirs=False)
+            make_path_mapper_kwargs = kwargs
+            if "basedir" in make_path_mapper_kwargs:
+                make_path_mapper_kwargs = make_path_mapper_kwargs.copy()
+                del make_path_mapper_kwargs["basedir"]
+            self.generatemapper = self.make_pathmapper(cast(List[Any], self.generatefiles["listing"]),
+                                                       self.outdir, basedir=self.outdir, separateDirs=False, **make_path_mapper_kwargs)
             _logger.debug(u"[job %s] initial work dir %s", self.name,
                           json.dumps({p: self.generatemapper.mapper(p) for p in self.generatemapper.files()}, indent=4))
 
@@ -275,7 +280,7 @@ def run(self, pull_image=True, rm_container=True,
             rm_tmpdir=True, move_outputs="move", **kwargs):
         # type: (bool, bool, bool, Text, **Any) -> None
 
-        self._setup()
+        self._setup(kwargs)
 
         env = self.environment
         if not os.path.exists(self.tmpdir):
@@ -371,7 +376,7 @@ def run(self, pull_image=True, rm_container=True,
                     "Docker is not available for this tool, try --no-container"
                     " to disable Docker: %s" % e)
 
-        self._setup()
+        self._setup(kwargs)
 
         runtime = [u"docker", u"run", u"-i"]
 

cwltool/pathmapper.py

@@ -5,6 +5,11 @@
 import stat
 import uuid
 from functools import partial
+from tempfile import NamedTemporaryFile
+
+import requests
+from cachecontrol import CacheControl
+from cachecontrol.caches import FileCache
 from typing import Any, Callable, Dict, Iterable, List, Set, Text, Tuple, Union
 
 import schema_salad.validate as validate
@@ -139,6 +144,29 @@ def trim_listing(obj):
     if obj.get("location", "").startswith("file://") and "listing" in obj:
         del obj["listing"]
 
+# Download http Files
+def downloadHttpFile(httpurl):
+    # type: (Text) -> Text
+    cache_session = None
+    if "XDG_CACHE_HOME" in os.environ:
+        directory = os.environ["XDG_CACHE_HOME"]
+    elif "HOME" in os.environ:
+        directory = os.environ["HOME"]
+    else:
+        directory = os.path.expanduser('~')
+
+    cache_session = CacheControl(
+        requests.Session(),
+        cache=FileCache(
+            os.path.join(directory, ".cache", "cwltool")))
+
+    r = cache_session.get(httpurl, stream=True)
+    with NamedTemporaryFile(mode='wb', delete=False) as f:
+        for chunk in r.iter_content(chunk_size=16384):
+            if chunk:  # filter out keep-alive new chunks
+                f.write(chunk)
+    r.close()
+    return f.name
 
 class PathMapper(object):
     """Mapping of files from relative path provided in the file to a tuple of
@@ -208,14 +236,18 @@ def visit(self, obj, stagedir, basedir, copy=False, staged=False):
                 self._pathmap[obj["location"]] = MapperEnt(obj["contents"], tgt, "CreateFile", staged)
             else:
                 with SourceLine(obj, "location", validate.ValidationException):
-                    # Dereference symbolic links
                     deref = ab
-                    st = os.lstat(deref)
-                    while stat.S_ISLNK(st.st_mode):
-                        rl = os.readlink(deref)
-                        deref = rl if os.path.isabs(rl) else os.path.join(
-                            os.path.dirname(deref), rl)
+                    if urllib.parse.urlsplit(deref).scheme in ['http','https']:
+                        deref = downloadHttpFile(path)
+                    else:
+                        # Dereference symbolic links
                         st = os.lstat(deref)
+                        while stat.S_ISLNK(st.st_mode):
+                            rl = os.readlink(deref)
+                            deref = rl if os.path.isabs(rl) else os.path.join(
+                                os.path.dirname(deref), rl)
+                            st = os.lstat(deref)
+
                     self._pathmap[path] = MapperEnt(deref, tgt, "WritableFile" if copy else "File", staged)
                     self.visitlisting(obj.get("secondaryFiles", []), stagedir, basedir, copy=copy, staged=staged)
 

cwltool/stdfsaccess.py

@@ -13,6 +13,8 @@
 def abspath(src, basedir):  # type: (Text, Text) -> Text
     if src.startswith(u"file://"):
         ab = six.text_type(uri_file_path(str(src)))
+    elif urllib.parse.urlsplit(src).scheme in ['http','https']:
+        return src
     else:
         if basedir.startswith(u"file://"):
             ab = src if os.path.isabs(src) else basedir+ '/'+ src

tests/test_http_input.py

@@ -0,0 +1,26 @@
+from __future__ import absolute_import
+import unittest
+import os
+import tempfile
+from cwltool.pathmapper import PathMapper
+
+
+class TestHttpInput(unittest.TestCase):
+    def test_http_path_mapping(self):
+        class SubPathMapper(PathMapper):
+            def __init__(self, referenced_files, basedir, stagedir):
+                super(SubPathMapper, self).__init__(referenced_files, basedir, stagedir)
+        input_file_path = "https://github.com/raw/common-workflow-language/cwltool/master/tests/2.fasta"
+        tempdir = tempfile.mkdtemp()
+        base_file = [{
+            "class": "File",
+            "location": "https://github.com/raw/common-workflow-language/cwltool/master/tests/2.fasta",
+            "basename": "chr20.fa"
+        }]
+        path_map_obj = SubPathMapper(base_file, os.getcwd(), tempdir)
+
+        self.assertIn(input_file_path,path_map_obj._pathmap)
+        assert os.path.exists(path_map_obj._pathmap[input_file_path].resolved) == 1
+        with open(path_map_obj._pathmap[input_file_path].resolved) as f:
+            self.assertIn(">Sequence 561 BP; 135 A; 106 C; 98 G; 222 T; 0 other;",f.read())
+            f.close()