Optional directory listing (#311)

* Do not expand 'listing' field on Directory objects unless loadListing is provided.
loadListing can be "shallow" (expand one level) or "deep" (expand recursively).

* Implement mechanism for registering schema extensions in cwltool.

* Make sure empty directory literals get created.

* Tests for Directory listing behavior extensions.

Author: tetron

cwltool/builder.py

@@ -8,7 +8,7 @@
 
 from . import expression
 from .errors import WorkflowException
-from .pathmapper import PathMapper, adjustFileObjs, normalizeFilesDirs
+from .pathmapper import PathMapper, adjustFileObjs, normalizeFilesDirs, get_listing
 from .stdfsaccess import StdFsAccess
 from .utils import aslist
 
@@ -42,6 +42,10 @@ def __init__(self):  # type: () -> None
         self.build_job_script = None  # type: Callable[[List[str]], Text]
         self.debug = False  # type: bool
 
+        # One of None, "shallow", "deep"
+        # Will be default None for CWL v1.1
+        self.loadListing = "deep"  # type: Union[None, str]
+
     def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
         # type: (Dict[Text, Any], Any, Union[int, List[int]], List[int]) -> List[Dict[Text, Any]]
         if tail_pos is None:
@@ -112,9 +116,10 @@ def bind_input(self, schema, datum, lead_pos=None, tail_pos=None):
 
             if schema["type"] == "File":
                 self.files.append(datum)
-                if binding and binding.get("loadContents"):
-                    with self.fs_access.open(datum["location"], "rb") as f:
-                        datum["contents"] = f.read(CONTENT_LIMIT)
+                if binding:
+                    if binding.get("loadContents"):
+                        with self.fs_access.open(datum["location"], "rb") as f:
+                            datum["contents"] = f.read(CONTENT_LIMIT)
 
                 if "secondaryFiles" in schema:
                     if "secondaryFiles" not in datum:
@@ -142,6 +147,9 @@ def _capture_files(f):
                 adjustFileObjs(datum.get("secondaryFiles", []), _capture_files)
 
             if schema["type"] == "Directory":
+                ll = self.loadListing or (binding and binding.get("loadListing"))
+                if ll:
+                    get_listing(self.fs_access, datum, (ll == "deep"))
                 self.files.append(datum)
 
         # Position to front of the sort key

cwltool/draft2tool.py

@@ -20,8 +20,8 @@
 from .pathmapper import adjustDirObjs
 from .errors import WorkflowException
 from .job import CommandLineJob
-from .pathmapper import PathMapper
-from .process import Process, shortname, uniquename, getListing, normalizeFilesDirs, compute_checksums
+from .pathmapper import PathMapper, get_listing, trim_listing
+from .process import Process, shortname, uniquename, normalizeFilesDirs, compute_checksums
 from .stdfsaccess import StdFsAccess
 from .utils import aslist
 
@@ -419,6 +419,7 @@ def collect_output_ports(self, ports, builder, outdir, compute_checksum=True):
                                 % (shortname(port["id"]), indent(u(str(e)))))
 
             if ret:
+                adjustDirObjs(ret, trim_listing)
                 adjustFileObjs(ret,
                                cast(Callable[[Any], Any],  # known bug in mypy
                                     # https://github.com/python/mypy/issues/797
@@ -468,8 +469,10 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                             raise
 
                 for files in r:
-                    if files["class"] == "Directory" and "listing" not in files:
-                        getListing(fs_access, files)
+                    if files["class"] == "Directory":
+                        ll = builder.loadListing or (binding and binding.get("loadListing"))
+                        if ll:
+                            get_listing(fs_access, files, (ll == "deep"))
                     else:
                         with fs_access.open(files["location"], "rb") as f:
                             contents = ""

cwltool/extensions.yml

@@ -0,0 +1,22 @@
+$base: http://commonwl.org/cwltool#
+$namespaces:
+  cwl: "https://w3id.org/cwl/cwl#"
+$graph:
+- $import: https://w3id.org/cwl/CommonWorkflowLanguage.yml
+
+- name: LoadListingRequirement
+  type: record
+  extends: cwl:ProcessRequirement
+  fields:
+    class:
+      type: string
+      doc: "Always 'LoadListingRequirement'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    loadListing:
+      type:
+        - "null"
+        - type: enum
+          name: LoadListingEnum
+          symbols: [shallow, deep]

cwltool/job.py

@@ -154,9 +154,10 @@ def run(self, dry_run=False, pull_image=True, rm_container=True,
 
         if img_id:
             runtime = ["docker", "run", "-i"]
-            for src in self.pathmapper.files():
-                vol = self.pathmapper.mapper(src)
-                if vol.type == "File":
+            for src, vol in self.pathmapper.items():
+                if not vol.staged:
+                    continue
+                if vol.type in ("File", "Directory"):
                     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))
@@ -209,7 +210,7 @@ def run(self, dry_run=False, pull_image=True, rm_container=True,
             env["HOME"] = self.outdir
             env["TMPDIR"] = self.tmpdir
 
-            stageFiles(self.pathmapper, os.symlink)
+            stageFiles(self.pathmapper, os.symlink, ignoreWritable=True)
 
         scr, _ = get_feature(self, "ShellCommandRequirement")
 

cwltool/main.py

@@ -20,13 +20,13 @@
 
 from . import draft2tool
 from . import workflow
-from .builder import adjustFileObjs
-from .pathmapper import adjustDirObjs
+from .pathmapper import adjustDirObjs, get_listing, adjustFileObjs, trim_listing
 from .cwlrdf import printrdf, printdot
 from .errors import WorkflowException, UnsupportedRequirement
 from .load_tool import fetch_document, validate_document, make_tool
 from .pack import pack
-from .process import shortname, Process, getListing, relocateOutputs, cleanIntermediate, scandeps, normalizeFilesDirs
+from .process import (shortname, Process, relocateOutputs, cleanIntermediate,
+                      scandeps, normalizeFilesDirs, use_custom_schema, use_standard_schema)
 from .resolver import tool_resolver
 from .stdfsaccess import StdFsAccess
 
@@ -148,9 +148,13 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                                                 "relative to primary file or current working directory.")
 
     parser.add_argument("--enable-dev", action="store_true",
-                        help="Allow loading and running development versions "
+                        help="Enable loading and running development versions "
                              "of CWL spec.", default=False)
 
+    parser.add_argument("--enable-ext", action="store_true",
+                        help="Enable loading and running cwltool extensions "
+                             "to CWL spec.", default=False)
+
     parser.add_argument("--default-container",
                         help="Specify a default docker container that will be used if the workflow fails to specify one.")
     parser.add_argument("--no-match-user", action="store_true",
@@ -175,8 +179,9 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                          dest="compute_checksum")
 
     parser.add_argument("--relax-path-checks", action="store_true",
-                        default=False, help="Relax requirements on path names. Currently "
-                                            "allows spaces.", dest="relax_path_checks")
+                        default=False, help="Relax requirements on path names to permit "
+                        "spaces and hash characters.", dest="relax_path_checks")
+
     parser.add_argument("workflow", type=Text, nargs="?", default=None)
     parser.add_argument("job_order", nargs=argparse.REMAINDER)
 
@@ -234,7 +239,8 @@ def output_callback(out, processStatus):
 
     if final_output and final_output[0] and finaloutdir:
         final_output[0] = relocateOutputs(final_output[0], finaloutdir,
-                                          output_dirs, kwargs.get("move_outputs"))
+                                          output_dirs, kwargs.get("move_outputs"),
+                                          kwargs["make_fs_access"](""))
 
     if kwargs.get("rm_tmpdir"):
         cleanIntermediate(output_dirs)
@@ -488,9 +494,8 @@ def pathToLoc(p):
 
     adjustDirObjs(job_order_object, pathToLoc)
     adjustFileObjs(job_order_object, pathToLoc)
+    adjustDirObjs(job_order_object, trim_listing)
     normalizeFilesDirs(job_order_object)
-    adjustDirObjs(job_order_object, cast(Callable[..., Any],
-                                         functools.partial(getListing, make_fs_access(input_basedir))))
 
     if "cwl:tool" in job_order_object:
         del job_order_object["cwl:tool"]
@@ -601,6 +606,7 @@ def main(argsl=None,  # type: List[str]
                      'debug': False,
                      'version': False,
                      'enable_dev': False,
+                     'enable_ext': False,
                      'strict': True,
                      'rdf_serializer': None,
                      'basedir': None,
@@ -636,6 +642,13 @@ def main(argsl=None,  # type: List[str]
         if args.relax_path_checks:
             draft2tool.ACCEPTLIST_RE = draft2tool.ACCEPTLIST_EN_RELAXED_RE
 
+        if args.enable_ext:
+            res = pkg_resources.resource_stream(__name__, 'extensions.yml')
+            use_custom_schema("v1.0", "http://commonwl.org/cwltool", res.read())
+            res.close()
+        else:
+            use_standard_schema("v1.0")
+
         try:
             document_loader, workflowobj, uri = fetch_document(args.workflow, resolver=resolver,
                                                                fetcher_constructor=fetcher_constructor)
@@ -708,13 +721,16 @@ def main(argsl=None,  # type: List[str]
                 setattr(args, 'move_outputs', "copy")
             setattr(args, "tmp_outdir_prefix", args.cachedir)
 
-        if job_order_object is None:
-            job_order_object = load_job_order(args, tool, stdin,
-                                              print_input_deps=args.print_input_deps,
-                                              relative_deps=args.relative_deps,
-                                              stdout=stdout,
-                                              make_fs_access=make_fs_access,
-                                              fetcher_constructor=fetcher_constructor)
+        try:
+            if job_order_object is None:
+                    job_order_object = load_job_order(args, tool, stdin,
+                                                      print_input_deps=args.print_input_deps,
+                                                      relative_deps=args.relative_deps,
+                                                      stdout=stdout,
+                                                      make_fs_access=make_fs_access,
+                                                      fetcher_constructor=fetcher_constructor)
+        except SystemExit as e:
+            return e.code
 
         if isinstance(job_order_object, int):
             return job_order_object
@@ -731,6 +747,7 @@ def main(argsl=None,  # type: List[str]
 
             # This is the workflow output, it needs to be written
             if out is not None:
+
                 def locToPath(p):
                     if p["location"].startswith("file://"):
                         p["path"] = uri_file_path(p["location"])

cwltool/pathmapper.py

@@ -11,9 +11,11 @@
 from typing import Any, Callable, Set, Text, Tuple, Union
 from six.moves import urllib
 
+from .stdfsaccess import abspath, StdFsAccess
+
 _logger = logging.getLogger("cwltool")
 
-MapperEnt = collections.namedtuple("MapperEnt", ["resolved", "target", "type"])
+MapperEnt = collections.namedtuple("MapperEnt", ["resolved", "target", "type", "staged"])
 
 
 def adjustFiles(rec, op):  # type: (Any, Union[Callable[..., Any], partial[Any]]) -> None
@@ -77,12 +79,6 @@ def addLocation(d):
     adjustDirObjs(job, addLocation)
 
 
-def abspath(src, basedir):  # type: (Text, Text) -> Text
-    if src.startswith(u"file://"):
-        ab = unicode(uri_file_path(str(src)))
-    else:
-        ab = src if os.path.isabs(src) else os.path.join(basedir, src)
-    return ab
 
 
 def dedup(listing):  # type: (List[Any]) -> List[Any]
@@ -106,6 +102,38 @@ def mark(d):
 
     return dd
 
+def get_listing(fs_access, rec, recursive=True):
+    # type: (StdFsAccess, Dict[Text, Any], bool) -> None
+    if "listing" in rec:
+        return
+    listing = []
+    loc = rec["location"]
+    for ld in fs_access.listdir(loc):
+        parse = urllib.parse.urlparse(ld)
+        bn = os.path.basename(urllib.request.url2pathname(parse.path))
+        if fs_access.isdir(ld):
+            ent = {u"class": u"Directory",
+                   u"location": ld,
+                   u"basename": bn}
+            if recursive:
+                get_listing(fs_access, ent, recursive)
+            listing.append(ent)
+        else:
+            listing.append({"class": "File", "location": ld, "basename": bn})
+    rec["listing"] = listing
+
+def trim_listing(obj):
+    """Remove 'listing' field from Directory objects that are file references.
+
+    It redundant and potentially expensive to pass fully enumerated Directory
+    objects around if not explicitly needed, so delete the 'listing' field when
+    it is safe to do so.
+
+    """
+
+    if obj.get("location", "").startswith("file://") and "listing" in obj:
+        del obj["listing"]
+
 
 class PathMapper(object):
     """Mapping of files from relative path provided in the file to a tuple of
@@ -148,44 +176,42 @@ def __init__(self, referenced_files, basedir, stagedir, separateDirs=True):
         self.separateDirs = separateDirs
         self.setup(dedup(referenced_files), basedir)
 
-    def visitlisting(self, listing, stagedir, basedir, copy=False):
-        # type: (List[Dict[Text, Any]], Text, Text, bool) -> None
+    def visitlisting(self, listing, stagedir, basedir, copy=False, staged=False):
+        # type: (List[Dict[Text, Any]], Text, Text, bool, bool) -> None
         for ld in listing:
-            tgt = os.path.join(stagedir, ld["basename"])
-            if ld["class"] == "Directory":
-                self.visit(ld, stagedir, basedir, copy=ld.get("writable", copy))
-            else:
-                self.visit(ld, stagedir, basedir, copy=ld.get("writable", copy))
+            self.visit(ld, stagedir, basedir, copy=ld.get("writable", copy), staged=staged)
 
-    def visit(self, obj, stagedir, basedir, copy=False):
-        # type: (Dict[Text, Any], Text, Text, bool) -> None
+    def visit(self, obj, stagedir, basedir, copy=False, staged=False):
+        # type: (Dict[Text, Any], Text, Text, bool, bool) -> None
         tgt = os.path.join(stagedir, obj["basename"])
+        if obj["location"] in self._pathmap:
+            return
         if obj["class"] == "Directory":
-            self._pathmap[obj["location"]] = MapperEnt(obj["location"], tgt, "Directory")
-            self.visitlisting(obj.get("listing", []), tgt, basedir, copy=copy)
+            if obj["location"].startswith("file://"):
+                resolved = uri_file_path(obj["location"])
+            else:
+                resolved = obj["location"]
+            self._pathmap[obj["location"]] = MapperEnt(resolved, tgt, "WritableDirectory" if copy else "Directory", staged)
+            if obj["location"].startswith("file://"):
+                staged = False
+            self.visitlisting(obj.get("listing", []), tgt, basedir, copy=copy, staged=staged)
         elif obj["class"] == "File":
             path = obj["location"]
-            if path in self._pathmap:
-                return
             ab = abspath(path, basedir)
             if "contents" in obj and obj["location"].startswith("_:"):
-                self._pathmap[obj["location"]] = MapperEnt(obj["contents"], tgt, "CreateFile")
+                self._pathmap[obj["location"]] = MapperEnt(obj["contents"], tgt, "CreateFile", staged)
             else:
-                if copy:
-                    self._pathmap[path] = MapperEnt(ab, tgt, "WritableFile")
-                else:
-                    with SourceLine(obj, "location", validate.ValidationException):
-                        # Dereference symbolic links
-                        deref = ab
+                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)
                         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, "File")
-                self.visitlisting(obj.get("secondaryFiles", []), stagedir, basedir, copy=copy)
+                    self._pathmap[path] = MapperEnt(deref, tgt, "WritableFile" if copy else "File", staged)
+                    self.visitlisting(obj.get("secondaryFiles", []), stagedir, basedir, copy=copy, staged=staged)
 
     def setup(self, referenced_files, basedir):
         # type: (List[Any], Text) -> None
@@ -196,13 +222,13 @@ def setup(self, referenced_files, basedir):
         for fob in referenced_files:
             if self.separateDirs:
                 stagedir = os.path.join(self.stagedir, "stg%s" % uuid.uuid4())
-            self.visit(fob, stagedir, basedir)
+            self.visit(fob, stagedir, basedir, staged=True)
 
     def mapper(self, src):  # type: (Text) -> MapperEnt
         if u"#" in src:
             i = src.index(u"#")
             p = self._pathmap[src[:i]]
-            return MapperEnt(p.resolved, p.target + src[i:], None)
+            return MapperEnt(p.resolved, p.target + src[i:], p.type, p.staged)
         else:
             return self._pathmap[src]