Eliminate dirent (#112)

* Recover eliminate dirent work.

* Squashed 'cwltool/schemas/' changes from afe31c1..b40896c

b40896c Fix basename.
34d946e Make location optional to support File contents literals.
f69576a Remove mapSubject entryname.
7d002ad Fix field name conflict.
d873510 Use file content literals for InitialWorkDir.
3fb9340 Get rid of Dirent.  Update spec for basename.  Add contents field to File spec.
d5b2627 Fix ttl and context generation
01cf6db Site generation works again
7302ecc Make "location" field optional when "listing" is provided.
63dcbdc Merge branch 'master' of github.com:common-workflow-language/common-workflow-language
09bd8ef Updating site generation for draft-4.
def6e52 Merge commit '6266928eac323672c0c44d16dca91d808fbc5e68'
6266928 Squashed 'draft-4/salad/' changes from 9c8c36f..c509a07
6f9cb1c fix the draft-4 tests
cb78824 Merge pull request #236 from common-workflow-language/directory
fcc5374 Merge remote-tracking branch 'origin/master' into directory
a90b71f replace errant draft-4.dev3 with dev2
11b96f1 Merge pull request #242 from common-workflow-language/scatter-inputs
93b60d9 Merge remote-tracking branch 'origin/master' into scatter-inputs
9caa2cf Merge pull request #240 from common-workflow-language/dependency_hint
db4113b Update text to say that 'inputs' on scatter is the post-scatter input object.
9ba0d38 Add test using value of inputs post-scatter.
cae0407 Add test staging files into output directory for update.
972e3b7 Merge pull request #232 from common-workflow-language/move-file-related-fields
6c6ab97 fix failing tests
0b12a24 soften description, allow for list of specs, add identifier
43b5fb9 Describe LooseDependencyHints.
7eeb70a reparent file specific fields from SchemaBase to Parameter

git-subtree-dir: cwltool/schemas
git-subtree-split: b40896c

* Get cwltest from master.

* Putting Dirent back in for InitialWorkDir

* Squashed 'cwltool/schemas/' changes from b40896c..d5f2322

d5f2322 Restore Dirent for InitialWorkDir only.

git-subtree-dir: cwltool/schemas
git-subtree-split: d5f2322

* Squashed 'cwltool/schemas/' changes from d5f2322..95b7454

95b7454 Using Dirent.

git-subtree-dir: cwltool/schemas
git-subtree-split: 95b7454

* Allow entryname to be optional.

* Squashed 'cwltool/schemas/' changes from 95b7454..c1cd55f

c1cd55f Remove conflicting mapSubject/mapPredicate

git-subtree-dir: cwltool/schemas
git-subtree-split: c1cd55f

* Fix support for file literals.

* Add --on-error

* update types to be consistent

Author: tetron

cwltool/builder.py

@@ -6,7 +6,7 @@
 from typing import Any, Union, AnyStr, Callable
 from .errors import WorkflowException
 from .stdfsaccess import StdFsAccess
-from .pathmapper import PathMapper, adjustFileObjs, adjustDirObjs
+from .pathmapper import PathMapper, adjustFileObjs, adjustDirObjs, normalizeFilesDirs
 
 CONTENT_LIMIT = 64 * 1024
 
@@ -31,7 +31,8 @@ def __init__(self):  # type: () -> None
         self.resources = None  # type: Dict[str, Union[int, str]]
         self.bindings = []  # type: List[Dict[str, Any]]
         self.timeout = None  # type: int
-        self.pathmapper = None # type: PathMapper
+        self.pathmapper = None  # type: PathMapper
+        self.stagedir = None  # type: unicode
 
     def bind_input(self, schema, datum, lead_pos=[], tail_pos=[]):
         # type: (Dict[unicode, Any], Any, List[int], List[int]) -> List[Dict[str, Any]]
@@ -112,7 +113,8 @@ def bind_input(self, schema, datum, lead_pos=[], tail_pos=[]):
                         if isinstance(sf, dict) or "$(" in sf or "${" in sf:
                             secondary_eval = self.do_eval(sf, context=datum)
                             if isinstance(secondary_eval, basestring):
-                                sfpath = {"location": secondary_eval, "class": "File"}
+                                sfpath = {"location": secondary_eval,
+                                          "class": "File"}
                             else:
                                 sfpath = secondary_eval
                         else:
@@ -121,6 +123,7 @@ def bind_input(self, schema, datum, lead_pos=[], tail_pos=[]):
                             datum["secondaryFiles"].extend(sfpath)
                         else:
                             datum["secondaryFiles"].append(sfpath)
+                    normalizeFilesDirs(datum["secondaryFiles"])
 
                 def _capture_files(f):
                     self.files.append(f)
@@ -187,8 +190,14 @@ def generate_arg(self, binding):  # type: (Dict[str,Any]) -> List[str]
 
         return [a for a in args if a is not None]
 
-    def do_eval(self, ex, context=None, pull_image=True):
-        # type: (Dict[str,str], Any, bool) -> Any
+    def do_eval(self, ex, context=None, pull_image=True, recursive=False):
+        # type: (Union[Dict[str, str], unicode], Any, bool, bool) -> Any
+        if recursive:
+            if isinstance(ex, dict):
+                return {k: self.do_eval(v, context, pull_image, recursive) for k,v in ex.iteritems()}
+            if isinstance(ex, list):
+                return [self.do_eval(v, context, pull_image, recursive) for v in ex]
+
         return expression.do_eval(ex, self.job, self.requirements,
                                   self.outdir, self.tmpdir,
                                   self.resources,

cwltool/cwltest.py

@@ -58,9 +58,9 @@ def compare(a, b):  # type: (Any, Any) -> bool
                         raise CompareFail(u"%s not in %s" % (json.dumps(i, indent=4, sort_keys=True), json.dumps(b, indent=4, sort_keys=True)))
 
             a = {k: v for k, v in a.iteritems()
-                 if k not in ("path", "location", "listing")}
+                 if k not in ("path", "location", "listing", "basename")}
             b = {k: v for k, v in b.iteritems()
-                 if k not in ("path", "location", "listing")}
+                 if k not in ("path", "location", "listing", "basename")}
 
             if len(a) != len(b):
                 raise CompareFail(u"expected %s\ngot %s" % (json.dumps(a, indent=4, sort_keys=True), json.dumps(b, indent=4, sort_keys=True)))

cwltool/draft2tool.py

@@ -6,7 +6,6 @@
 import glob
 import logging
 import hashlib
-import random
 import re
 import urlparse
 import tempfile
@@ -17,7 +16,7 @@
 import shellescape
 from typing import Callable, Any, Union, Generator, cast
 
-from .process import Process, shortname, uniquename, getListing
+from .process import Process, shortname, uniquename, getListing, normalizeFilesDirs
 from .errors import WorkflowException
 from .utils import aslist
 from . import expression
@@ -50,7 +49,9 @@ def __init__(self):  # type: () -> None
 
         def run(self, **kwargs):  # type: (**Any) -> None
             try:
-                self.output_callback(self.builder.do_eval(self.script), "success")
+                ev = self.builder.do_eval(self.script)
+                normalizeFilesDirs(ev)
+                self.output_callback(ev, "success")
             except Exception as e:
                 _logger.warn(u"Failed to evaluate expression:\n%s", e, exc_info=(e if kwargs.get('debug') else False))
                 self.output_callback({}, "permanentFail")
@@ -110,6 +111,17 @@ def run(self, **kwargs):
                                                            self.cachebuilder, self.outdir),
                                             "success")
 
+# map files to assigned path inside a container. We need to also explicitly
+# walk over input as implicit reassignment doesn't reach everything in builder.bindings
+def check_adjust(builder, f):
+    # type: (Builder, Dict[str, Any]) -> Dict[str,Any]
+    f["path"] = builder.pathmapper.mapper(f["location"])[1]
+    f["dirname"], f["basename"] = os.path.split(f["path"])
+    if f["class"] == "File":
+        f["nameroot"], f["nameext"] = os.path.splitext(f["basename"])
+    if not ACCEPTLIST_RE.match(f["basename"]):
+        raise WorkflowException("Invalid filename: '%s' contains illegal characters" % (f["basename"]))
+    return f
 
 class CommandLineTool(Process):
     def __init__(self, toolpath_object, **kwargs):
@@ -120,7 +132,7 @@ def makeJobRunner(self):  # type: () -> CommandLineJob
         return CommandLineJob()
 
     def makePathMapper(self, reffiles, stagedir, **kwargs):
-        # type: (Set[Any], unicode, **Any) -> PathMapper
+        # type: (List[Any], unicode, **Any) -> PathMapper
         dockerReq, _ = self.get_requirement("DockerRequirement")
         try:
             return PathMapper(reffiles, kwargs["basedir"], stagedir)
@@ -141,8 +153,13 @@ def job(self, joborder, output_callback, **kwargs):
             cachebuilder = self._init_job(joborder, **cacheargs)
             cachebuilder.pathmapper = PathMapper(cachebuilder.files,
                                                  kwargs["basedir"],
-                                                 cachebuilder.stagedir)
-
+                                                 cachebuilder.stagedir,
+                                                 separateDirs=False)
+            _check_adjust = partial(check_adjust, cachebuilder)
+            adjustFileObjs(cachebuilder.files, _check_adjust)
+            adjustFileObjs(cachebuilder.bindings, _check_adjust)
+            adjustDirObjs(cachebuilder.files, _check_adjust)
+            adjustDirObjs(cachebuilder.bindings, _check_adjust)
             cmdline = flatten(map(cachebuilder.generate_arg, cachebuilder.bindings))
             (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
             if docker_req and kwargs.get("use_container") is not False:
@@ -151,8 +168,9 @@ def job(self, joborder, output_callback, **kwargs):
             keydict = {u"cmdline": cmdline}
 
             for _,f in cachebuilder.pathmapper.items():
-                st = os.stat(f[0])
-                keydict[f[0]] = [st.st_size, int(st.st_mtime * 1000)]
+                if f.type == "File":
+                    st = os.stat(f.resolved)
+                    keydict[f.resolved] = [st.st_size, int(st.st_mtime * 1000)]
 
             interesting = {"DockerRequirement",
                            "EnvVarRequirement",
@@ -236,19 +254,10 @@ def rm_pending_output_callback(output_callback, jobcachepending,
         builder.pathmapper = self.makePathMapper(reffiles, builder.stagedir, **kwargs)
         builder.requirements = j.requirements
 
-        # map files to assigned path inside a container. We need to also explicitly
-        # walk over input as implicit reassignment doesn't reach everything in builder.bindings
-        def _check_adjust(f):  # type: (Dict[str,Any]) -> Dict[str,Any]
-            f["path"] = builder.pathmapper.mapper(f["location"])[1]
-            f["dirname"], f["basename"] = os.path.split(f["path"])
-            if f["class"] == "File":
-                f["nameroot"], f["nameext"] = os.path.splitext(f["basename"])
-            if not ACCEPTLIST_RE.match(f["basename"]):
-                raise WorkflowException("Invalid filename: '%s' contains illegal characters" % (f["basename"]))
-            return f
-
         _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)
+
         adjustFileObjs(builder.files, _check_adjust)
         adjustFileObjs(builder.bindings, _check_adjust)
         adjustDirObjs(builder.files, _check_adjust)
@@ -260,27 +269,51 @@ def _check_adjust(f):  # type: (Dict[str,Any]) -> Dict[str,Any]
 
         _logger.debug(u"[job %s] command line bindings is %s", j.name, json.dumps(builder.bindings, indent=4))
 
-        dockerReq, _ = self.get_requirement("DockerRequirement")
+        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 = None
+            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": []}
+        j.generatefiles = {"class": "Directory", "listing": [], "basename": ""}
         if initialWorkdir:
+            ls = []  # type: List[Dict[str, Any]]
             if isinstance(initialWorkdir["listing"], (str, unicode)):
-                j.generatefiles["listing"] = builder.do_eval(initialWorkdir["listing"])
+                ls = builder.do_eval(initialWorkdir["listing"])
             else:
                 for t in initialWorkdir["listing"]:
-                    j.generatefiles["listing"].append({"entryname": builder.do_eval(t["entryname"]),
-                                                       "entry": copy.deepcopy(builder.do_eval(t["entry"]))})
+                    if "entry" in t:
+                        et = {"entry": builder.do_eval(t["entry"])}
+                        if "entryname" in t:
+                            et["entryname"] = builder.do_eval(t["entryname"])
+                        else:
+                            et["entryname"] = None
+                        ls.append(et)
+                    else:
+                        ls.append(t)
+            for i,t in enumerate(ls):
+                if "entry" in t:
+                    if isinstance(t["entry"], (str, unicode)):
+                        ls[i] = {
+                            "class": "File",
+                            "basename": t["entryname"],
+                            "contents": t["entry"]
+                        }
+                    else:
+                        if t["entryname"]:
+                            t = copy.deepcopy(t)
+                            t["entry"]["basename"] = t["entryname"]
+                        ls[i] = t["entry"]
+            j.generatefiles[u"listing"] = ls
+
+        normalizeFilesDirs(j.generatefiles)
 
         j.environment = {}
         evr = self.get_requirement("EnvVarRequirement")[0]
@@ -321,6 +354,8 @@ def collect_output_ports(self, ports, builder, outdir):
                             # https://github.com/python/mypy/issues/797
                             partial(revmap_file, builder, outdir)))
                 adjustFileObjs(ret, remove_path)
+                adjustDirObjs(ret, remove_path)
+                normalizeFilesDirs(ret)
                 validate.validate_ex(self.names.get_name("outputs_record_schema", ""), ret)
                 return ret
 
@@ -334,6 +369,7 @@ def collect_output_ports(self, ports, builder, outdir):
             if ret:
                 adjustFileObjs(ret, remove_path)
                 adjustDirObjs(ret, remove_path)
+                normalizeFilesDirs(ret)
             validate.validate_ex(self.names.get_name("outputs_record_schema", ""), ret)
             return ret if ret is not None else {}
         except validate.ValidationException as e:

cwltool/job.py

@@ -58,11 +58,12 @@ def __init__(self):  # type: () -> None
         self.outdir = None  # type: str
         self.tmpdir = None  # type: str
         self.environment = None  # type: Dict[str,str]
-        self.generatefiles = None  # type: Dict[str,Union[Dict[str,str],str]]
+        self.generatefiles = None  # type: Dict[unicode, Union[List[Dict[str, str]], Dict[str,str], str]] 
+        self.stagedir = None  # type: unicode
 
     def run(self, dry_run=False, pull_image=True, rm_container=True,
             rm_tmpdir=True, move_outputs="move", **kwargs):
-        # type: (bool, bool, bool, bool, bool, **Any) -> Union[Tuple[str,Dict[None,None]],None]
+        # type: (bool, bool, bool, bool, bool, unicode, **Any) -> Union[Tuple[str,Dict[None,None]],None]
         if not os.path.exists(self.outdir):
             os.makedirs(self.outdir)
 
@@ -78,10 +79,12 @@ def run(self, dry_run=False, pull_image=True, rm_container=True,
 
         (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
 
-        for f in self.pathmapper.files():
-            p = self.pathmapper.mapper(f)
+        for knownfile in self.pathmapper.files():
+            p = self.pathmapper.mapper(knownfile)
             if p.type == "File" and not os.path.isfile(p[0]):
-                raise WorkflowException(u"Input file %s (at %s) not found or is not a regular file." % (f, self.pathmapper.mapper(f)[0]))
+                raise WorkflowException(
+                u"Input file %s (at %s) not found or is not a regular file."
+                % (knownfile, self.pathmapper.mapper(knownfile)[0]))
 
         img_id = None
         if docker_req and kwargs.get("use_container") is not False:
@@ -97,6 +100,11 @@ def run(self, dry_run=False, pull_image=True, rm_container=True,
                 vol = self.pathmapper.mapper(src)
                 if vol.type == "File":
                     runtime.append(u"--volume=%s:%s:ro" % (vol.resolved, vol.target))
+                if vol.type == "CreateFile":
+                    createtmp = os.path.join(self.stagedir, os.path.basename(vol.target))
+                    with open(createtmp, "w") as f:
+                        f.write(vol.resolved.encode("utf-8"))
+                    runtime.append(u"--volume=%s:%s:ro" % (createtmp, vol.target))
             runtime.append(u"--volume=%s:%s:rw" % (os.path.abspath(self.outdir), "/var/spool/cwl"))
             runtime.append(u"--volume=%s:%s:rw" % (os.path.abspath(self.tmpdir), "/tmp"))
             runtime.append(u"--workdir=%s" % ("/var/spool/cwl"))

cwltool/load_tool.py

@@ -8,6 +8,7 @@
 import urlparse
 from schema_salad.ref_resolver import Loader
 import schema_salad.validate as validate
+from schema_salad.validate import ValidationException
 import schema_salad.schema as schema
 from avro.schema import Names
 from . import update
@@ -37,13 +38,12 @@ def fetch_document(argsworkflow):
         workflowobj = argsworkflow
         uri = "#" + str(id(argsworkflow))
     else:
-        raise validate.ValidationException(
-            "Must be URI or object: '%s'" % argsworkflow)
+        raise ValidationException("Must be URI or object: '%s'" % argsworkflow)
 
     return document_loader, workflowobj, uri
 
 def _convert_stdstreams_to_files(workflowobj):
-    # type: (Union[Dict[unicode, Any], List[Dict[unicode, Any]]) -> None
+    # type: (Union[Dict[unicode, Any], List[Dict[unicode, Any]]]) -> None
 
     if isinstance(workflowobj, dict):
         if ('class' in workflowobj
@@ -53,7 +53,7 @@ def _convert_stdstreams_to_files(workflowobj):
                 for streamtype in ['stdout', 'stderr']:
                     if out['type'] == streamtype:
                         if 'outputBinding' in out:
-                            raise validate.ValidateException(
+                            raise ValidationException(
                                     "Not allowed to specify outputBinding when"
                                     " using %s shortcut." % streamtype)
                         if streamtype in workflowobj:
@@ -109,12 +109,11 @@ def validate_document(document_loader, workflowobj, uri,
     workflowobj["id"] = fileuri
     processobj, metadata = document_loader.resolve_all(workflowobj, fileuri)
     if not isinstance(processobj, (dict, list)):
-        raise validate.ValidationException("Workflow must be a dict or list.")
+        raise ValidationException("Workflow must be a dict or list.")
 
     if not metadata:
         if not isinstance(processobj, dict):
-            raise validate.ValidationException(
-                    "Draft-2 workflows must be a dict.")
+            raise ValidationException("Draft-2 workflows must be a dict.")
         metadata = {"$namespaces": processobj.get("$namespaces", {}),
                    "$schemas": processobj.get("$schemas", []),
                    "cwlVersion": processobj["cwlVersion"]}