Merge branch 'master' into node_fix

Author: mr-c

MANIFEST.in

@@ -1,5 +1,6 @@
 include gittaggers.py Makefile cwltool.py
 include tests/*
+include tests/tmp1/tmp2/tmp3/.gitkeep
 include tests/wf/*
 include cwltool/schemas/v1.0/*.yml
 include cwltool/schemas/draft-2/*.yml

cwltool/builder.py

@@ -8,9 +8,10 @@
 
 from . import expression
 from .errors import WorkflowException
-from .pathmapper import PathMapper, adjustFileObjs, normalizeFilesDirs, get_listing
+from .pathmapper import PathMapper, normalizeFilesDirs, get_listing, visit_class
 from .stdfsaccess import StdFsAccess
 from .utils import aslist
+from .mutation import MutationManager
 
 CONTENT_LIMIT = 64 * 1024
 
@@ -41,6 +42,7 @@ def __init__(self):  # type: () -> None
         self.make_fs_access = None  # type: Type[StdFsAccess]
         self.build_job_script = None  # type: Callable[[List[str]], Text]
         self.debug = False  # type: bool
+        self.mutation_manager = None  # type: MutationManager
 
         # One of "no_listing", "shallow_listing", "deep_listing"
         # Will be default "no_listing" for CWL v1.1
@@ -144,7 +146,7 @@ def _capture_files(f):
                     self.files.append(f)
                     return f
 
-                adjustFileObjs(datum.get("secondaryFiles", []), _capture_files)
+                visit_class(datum.get("secondaryFiles", []), ("File", "Directory"), _capture_files)
 
             if schema["type"] == "Directory":
                 ll = self.loadListing or (binding and binding.get("loadListing"))

cwltool/draft2tool.py

@@ -14,14 +14,14 @@
 import shellescape
 from schema_salad.ref_resolver import file_uri, uri_file_path
 from schema_salad.sourceline import SourceLine, indent
-from typing import Any, Callable, cast, Generator, Text, Union
+from typing import Any, Callable, cast, Generator, Text, Union, Dict
 
-from .builder import CONTENT_LIMIT, substitute, Builder, adjustFileObjs
-from .pathmapper import adjustDirObjs
+from .builder import CONTENT_LIMIT, substitute, Builder
+from .pathmapper import adjustFileObjs, adjustDirObjs, visit_class
 from .errors import WorkflowException
-from .job import CommandLineJob
+from .job import JobBase, CommandLineJob, DockerCommandLineJob
 from .pathmapper import PathMapper, get_listing, trim_listing
-from .process import Process, shortname, uniquename, normalizeFilesDirs, compute_checksums
+from .process import Process, shortname, uniquename, normalizeFilesDirs, compute_checksums, _logger_validation_warnings
 from .stdfsaccess import StdFsAccess
 from .utils import aslist
 
@@ -106,6 +106,8 @@ def revmap_file(builder, outdir, f):
             revmap_f = builder.pathmapper.reversemap(path)
             if revmap_f:
                 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:])
         return f
@@ -148,6 +150,7 @@ def run(self, **kwargs):
 # walk over input as implicit reassignment doesn't reach everything in builder.bindings
 def check_adjust(builder, f):
     # type: (Builder, Dict[Text, Any]) -> Dict[Text, Any]
+
     f["path"] = builder.pathmapper.mapper(f["location"])[1]
     f["dirname"], f["basename"] = os.path.split(f["path"])
     if f["class"] == "File":
@@ -156,26 +159,36 @@ def check_adjust(builder, f):
         raise WorkflowException("Invalid filename: '%s' contains illegal characters" % (f["basename"]))
     return f
 
+def check_valid_locations(fs_access, ob):
+    if ob["location"].startswith("_:"):
+        pass
+    if ob["class"] == "File" and not fs_access.isfile(ob["location"]):
+        raise validate.ValidationException("Does not exist or is not a File: '%s'" % ob["location"])
+    if ob["class"] == "Directory" and not fs_access.isdir(ob["location"]):
+        raise validate.ValidationException("Does not exist or is not a Directory: '%s'" % ob["location"])
 
 class CommandLineTool(Process):
     def __init__(self, toolpath_object, **kwargs):
         # type: (Dict[Text, Any], **Any) -> None
         super(CommandLineTool, self).__init__(toolpath_object, **kwargs)
 
-    def makeJobRunner(self):  # type: () -> CommandLineJob
-        return CommandLineJob()
+    def makeJobRunner(self):  # type: () -> JobBase
+        dockerReq, _ = self.get_requirement("DockerRequirement")
+        if dockerReq:
+            return DockerCommandLineJob()
+        else:
+            return CommandLineJob()
 
     def makePathMapper(self, reffiles, stagedir, **kwargs):
         # type: (List[Any], Text, **Any) -> PathMapper
-        dockerReq, _ = self.get_requirement("DockerRequirement")
         return PathMapper(reffiles, kwargs["basedir"], stagedir)
 
     def job(self,
             job_order,  # type: Dict[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             **kwargs  # type: Any
             ):
-        # type: (...) -> Generator[Union[CommandLineJob, CallbackJob], None, None]
+        # type: (...) -> Generator[Union[JobBase, CallbackJob], None, None]
 
         jobname = uniquename(kwargs.get("name", shortname(self.tool.get("id", "job"))))
 
@@ -190,10 +203,9 @@ def job(self,
                                                  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)
+            visit_class([cachebuilder.files, cachebuilder.bindings],
+                       ("File", "Directory"), _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:
@@ -290,10 +302,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
 
         _check_adjust = partial(check_adjust, builder)
 
-        adjustFileObjs(builder.files, _check_adjust)
-        adjustFileObjs(builder.bindings, _check_adjust)
-        adjustDirObjs(builder.files, _check_adjust)
-        adjustDirObjs(builder.bindings, _check_adjust)
+        visit_class([builder.files, builder.bindings], ("File", "Directory"), _check_adjust)
 
         if self.tool.get("stdin"):
             with SourceLine(self.tool, "stdin", validate.ValidationException):
@@ -363,8 +372,39 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
                         ls[i] = t["entry"]
             j.generatefiles[u"listing"] = ls
 
+        inplaceUpdateReq = self.get_requirement("http://commonwl.org/cwltool#InplaceUpdateRequirement")[0]
+
+        if inplaceUpdateReq:
+            j.inplace_update = inplaceUpdateReq["inplaceUpdate"]
         normalizeFilesDirs(j.generatefiles)
 
+        readers = {}
+        muts = set()
+
+        if builder.mutation_manager:
+            def register_mut(f):
+                muts.add(f["location"])
+                builder.mutation_manager.register_mutation(j.name, f)
+
+            def register_reader(f):
+                if f["location"] not in muts:
+                    builder.mutation_manager.register_reader(j.name, f)
+                    readers[f["location"]] = f
+
+            for li in j.generatefiles["listing"]:
+                li = cast(Dict[Text, Any], li)
+                if li.get("writable") and j.inplace_update:
+                    adjustFileObjs(li, register_mut)
+                    adjustDirObjs(li, register_mut)
+                else:
+                    adjustFileObjs(li, register_reader)
+                    adjustDirObjs(li, register_reader)
+
+            adjustFileObjs(builder.files, register_reader)
+            adjustFileObjs(builder.bindings, register_reader)
+            adjustDirObjs(builder.files, register_reader)
+            adjustDirObjs(builder.bindings, register_reader)
+
         j.environment = {}
         evr = self.get_requirement("EnvVarRequirement")[0]
         if evr:
@@ -386,16 +426,17 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
         j.pathmapper = builder.pathmapper
         j.collect_outputs = partial(
             self.collect_output_ports, self.tool["outputs"], builder,
-            compute_checksum=kwargs.get("compute_checksum", True))
+            compute_checksum=kwargs.get("compute_checksum", True),
+            jobname=jobname,
+            readers=readers)
         j.output_callback = output_callbacks
 
         yield j
 
-    def collect_output_ports(self, ports, builder, outdir, compute_checksum=True):
-        # type: (Set[Dict[Text, Any]], Builder, Text, bool) -> Dict[Text, Union[Text, List[Any], Dict[Text, Any]]]
+    def collect_output_ports(self, ports, builder, outdir, compute_checksum=True, jobname="", readers=None):
+        # type: (Set[Dict[Text, Any]], Builder, Text, bool, Text, Dict[Text, Any]) -> Dict[Text, Union[Text, List[Any], Dict[Text, Any]]]
         ret = {}  # type: Dict[Text, Union[Text, List[Any], Dict[Text, Any]]]
         try:
-
             fs_access = builder.make_fs_access(outdir)
             custom_output = fs_access.join(outdir, "cwl.output.json")
             if fs_access.exists(custom_output):
@@ -419,21 +460,27 @@ def collect_output_ports(self, ports, builder, outdir, compute_checksum=True):
                                 % (shortname(port["id"]), indent(u(str(e)))))
 
             if ret:
+                revmap = partial(revmap_file, builder, outdir)
                 adjustDirObjs(ret, trim_listing)
-                adjustFileObjs(ret,
-                               cast(Callable[[Any], Any],  # known bug in mypy
-                                    # https://github.com/python/mypy/issues/797
-                                    partial(revmap_file, builder, outdir)))
-                adjustFileObjs(ret, remove_path)
-                adjustDirObjs(ret, remove_path)
+                visit_class(ret, ("File", "Directory"), cast(Callable[[Any], Any], revmap))
+                visit_class(ret, ("File", "Directory"), remove_path)
                 normalizeFilesDirs(ret)
+                if builder.mutation_manager:
+                    adjustFileObjs(ret, builder.mutation_manager.set_generation)
+                visit_class(ret, ("File", "Directory"), partial(check_valid_locations, fs_access))
+
                 if compute_checksum:
                     adjustFileObjs(ret, partial(compute_checksums, fs_access))
 
-            validate.validate_ex(self.names.get_name("outputs_record_schema", ""), ret)
+            validate.validate_ex(self.names.get_name("outputs_record_schema", ""), ret,
+                                 strict=False, logger=_logger_validation_warnings)
             return ret if ret is not None else {}
         except validate.ValidationException as e:
-            raise WorkflowException("Error validating output record, " + Text(e) + "\n in " + json.dumps(ret, indent=4))
+            raise WorkflowException("Error validating output record. " + Text(e) + "\n in " + json.dumps(ret, indent=4))
+        finally:
+            if builder.mutation_manager and readers:
+                for r in readers.values():
+                    builder.mutation_manager.release_reader(jobname, r)
 
     def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=True):
         # type: (Dict[Text, Any], Builder, Text, StdFsAccess, bool) -> Union[Dict[Text, Any], List[Union[Dict[Text, Any], Text]]]

cwltool/extensions.yml

@@ -19,4 +19,18 @@ $graph:
       type:
         - type: enum
           name: LoadListingEnum
-          symbols: [no_listing, shallow_listing, deep_listing]
+          symbols: [no_listing, shallow_listing, deep_listing]
+
+- name: InplaceUpdateRequirement
+  type: record
+  inVocab: false
+  extends: cwl:ProcessRequirement
+  fields:
+    class:
+      type: string
+      doc: "Always 'InplaceUpdateRequirement'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    inplaceUpdate:
+      type: boolean