Merge branch 'toil-fixes' of github.com:common-workflow-language/cwltool into toil-fixes

Author: Peter Amstutz

cwltool/builder.py

@@ -52,6 +52,7 @@ def __init__(self):  # type: () -> None
         self.make_fs_access = None  # type: Type[StdFsAccess]
         self.debug = False  # type: bool
         self.mutation_manager = None  # type: MutationManager
+        self.force_docker_pull = False  # type: bool
 
         # One of "no_listing", "shallow_listing", "deep_listing"
         # Will be default "no_listing" for CWL v1.1
@@ -255,4 +256,5 @@ def do_eval(self, ex, context=None, pull_image=True, recursive=False):
                                   self.resources,
                                   context=context, pull_image=pull_image,
                                   timeout=self.timeout,
+                                  force_docker_pull=self.force_docker_pull,
                                   debug=self.debug)

cwltool/expression.py

@@ -153,8 +153,8 @@ def next_seg(remain, obj):  # type: (Text, Any) -> Any
         return obj
 
 
-def evaluator(ex, jslib, obj, fullJS=False, timeout=None, debug=False):
-    # type: (Text, Text, Dict[Text, Any], bool, int, bool) -> JSON
+def evaluator(ex, jslib, obj, fullJS=False, timeout=None, force_docker_pull=False, debug=False):
+    # type: (Text, Text, Dict[Text, Any], bool, int, bool, bool) -> JSON
     m = param_re.match(ex)
     if m:
         if m.end(1)+1 == len(ex) and m.group(1) == "null":
@@ -164,16 +164,17 @@ def evaluator(ex, jslib, obj, fullJS=False, timeout=None, debug=False):
         except Exception as w:
             raise WorkflowException("%s%s" % (m.group(1), w))
     elif fullJS:
-        return sandboxjs.execjs(ex, jslib, timeout=timeout, debug=debug)
+        return sandboxjs.execjs(ex, jslib, timeout=timeout, force_docker_pull=force_docker_pull, debug=debug)
     else:
         raise sandboxjs.JavascriptException(
             "Syntax error in parameter reference '%s' or used Javascript code without specifying InlineJavascriptRequirement.",
             ex)
 
 
 def interpolate(scan, rootvars,
-                timeout=None, fullJS=None, jslib="", debug=False):
-    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text], bool) -> JSON
+                timeout=None, fullJS=None, jslib="",force_docker_pull=False,
+                debug=False):
+    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text], bool, bool) -> JSON
     scan = scan.strip()
     parts = []
     w = scanner(scan)
@@ -182,7 +183,8 @@ def interpolate(scan, rootvars,
 
         if scan[w[0]] == '$':
             e = evaluator(scan[w[0] + 1:w[1]], jslib, rootvars, fullJS=fullJS,
-                          timeout=timeout, debug=debug)
+                          timeout=timeout, force_docker_pull=force_docker_pull,
+                          debug=debug)
             if w[0] == 0 and w[1] == len(scan):
                 return e
             leaf = json.dumps(e, sort_keys=True)
@@ -200,8 +202,8 @@ def interpolate(scan, rootvars,
 
 
 def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
-            context=None, pull_image=True, timeout=None, debug=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) -> Any
+            context=None, pull_image=True, timeout=None, force_docker_pull=False, debug=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) -> Any
 
     runtime = copy.copy(resources)
     runtime["tmpdir"] = docker_windows_path_adjust(tmpdir)
@@ -227,6 +229,7 @@ def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
                                timeout=timeout,
                                fullJS=fullJS,
                                jslib=jslib,
+                               force_docker_pull=force_docker_pull,
                                debug=debug)
         except Exception as e:
             raise WorkflowException("Expression evaluation error:\n%s" % e)

cwltool/main.py

@@ -226,7 +226,9 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
     exgroup.add_argument("--make-template", action="store_true",
                          help="Generate a template input object")
 
-
+    parser.add_argument("--force-docker-pull", action="store_true",
+                        default=False, help="Pull latest docker image even if"
+                                            " it is locally present", dest="force_docker_pull")
     parser.add_argument("workflow", type=Text, nargs="?", default=None)
     parser.add_argument("job_order", nargs=argparse.REMAINDER)
 

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/process.py

@@ -560,6 +560,7 @@ def _init_job(self, joborder, **kwargs):
 
         builder.make_fs_access = kwargs.get("make_fs_access") or StdFsAccess
         builder.fs_access = builder.make_fs_access(kwargs["basedir"])
+        builder.force_docker_pull = kwargs.get("force_docker_pull")
 
         loadListingReq, _ = self.get_requirement("http://commonwl.org/cwltool#LoadListingRequirement")
         if loadListingReq:

cwltool/sandboxjs.py

@@ -53,12 +53,11 @@ def check_js_threshold_version(working_alias):
         return False
 
 
-def new_js_proc():
-    # type: () -> subprocess.Popen
+def new_js_proc(force_docker_pull=False):
+    # type: (bool) -> subprocess.Popen
 
     res = resource_stream(__name__, 'cwlNodeEngine.js')
     nodecode = res.read()
-
     required_node_version, docker = (False,)*2
     nodejs = None
     trynodes = ("nodejs", "node")
@@ -86,10 +85,11 @@ def new_js_proc():
         try:
             nodeimg = "node:slim"
             global have_node_slim
+
             if not have_node_slim:
                 dockerimgs = subprocess.check_output(["docker", "images", "-q", nodeimg]).decode('utf-8')
                 # if output is an empty string
-                if len(dockerimgs.split("\n")) <= 1:
+                if (len(dockerimgs.split("\n")) <= 1) or force_docker_pull:
                     # pull node:slim docker container
                     nodejsimg = subprocess.check_output(["docker", "pull", nodeimg]).decode('utf-8')
                     _logger.info("Pulled Docker image %s %s", nodeimg, nodejsimg)
@@ -124,10 +124,10 @@ def new_js_proc():
     return nodejs
 
 
-def execjs(js, jslib, timeout=None, debug=False):  # type: (Union[Mapping, Text], Any, int, bool) -> JSON
+def execjs(js, jslib, timeout=None, force_docker_pull=False, debug=False):  # type: (Union[Mapping, Text], Any, int, bool, bool) -> JSON
 
     if not hasattr(localdata, "proc") or localdata.proc.poll() is not None or onWindows():
-        localdata.proc = new_js_proc()
+        localdata.proc = new_js_proc(force_docker_pull=force_docker_pull)
 
     nodejs = localdata.proc
 

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

requirements.txt

@@ -1,4 +1,4 @@
-requests>=1.0
+requests>=2.12.4
 ruamel.yaml>=0.12.4,<0.15
 rdflib==4.2.2
 rdflib-jsonld==0.4.0

setup.py

@@ -48,7 +48,7 @@
       include_package_data=True,
       install_requires=[
           'setuptools',
-          'requests >= 1.0',
+          'requests >= 2.12.4',
           'ruamel.yaml >= 0.12.4, < 0.15',
           'rdflib >= 4.2.2, < 4.3.0',
           'shellescape >= 3.4.1, < 3.5',

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://raw.githubusercontent.com/common-workflow-language/cwltool/master/tests/2.fasta"
+        tempdir = tempfile.mkdtemp()
+        base_file = [{
+            "class": "File",
+            "location": "https://raw.githubusercontent.com/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()