Validate that output files actually exist (#374)

* Add visit_class and refactor redundant traversals when both adjustFileObjs and
adjustDirObjs are called on the same objects with the same function.

* Validate file/directory locations on output.

* Fix output of outdir directly.

* Add test that outputs are better checked.

Author: tetron

cwltool/builder.py

@@ -8,7 +8,7 @@
 
 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
 
@@ -144,7 +144,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

@@ -16,8 +16,8 @@
 from schema_salad.sourceline import SourceLine, indent
 from typing import Any, Callable, cast, Generator, Text, Union
 
-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 .pathmapper import PathMapper, get_listing, trim_listing
@@ -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
@@ -156,6 +158,13 @@ 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):
@@ -190,10 +199,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 +298,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):
@@ -419,21 +424,19 @@ 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)
+                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)
             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))
 
     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/job.py

@@ -339,10 +339,6 @@ def linkoutdir(src, tgt):
             _logger.debug(u"[job %s] Removing temporary directory %s", self.name, self.tmpdir)
             shutil.rmtree(self.tmpdir, True)
 
-        if move_outputs == "move" and empty_subtree(self.outdir):
-            _logger.debug(u"[job %s] Removing empty output directory %s", self.name, self.outdir)
-            shutil.rmtree(self.outdir, True)
-
 
 def _job_popen(
         commands,  # type: List[str]

cwltool/load_tool.py

@@ -68,39 +68,36 @@ def _convert_stdstreams_to_files(workflowobj):
     # type: (Union[Dict[Text, Any], List[Dict[Text, Any]]]) -> None
 
     if isinstance(workflowobj, dict):
-        if ('class' in workflowobj
-            and workflowobj['class'] == 'CommandLineTool'):
-            if 'outputs' in workflowobj:
-                for out in workflowobj['outputs']:
-                    for streamtype in ['stdout', 'stderr']:
-                        if out['type'] == streamtype:
-                            if 'outputBinding' in out:
-                                raise ValidationException(
-                                    "Not allowed to specify outputBinding when"
-                                    " using %s shortcut." % streamtype)
-                            if streamtype in workflowobj:
-                                filename = workflowobj[streamtype]
-                            else:
-                                filename = Text(uuid.uuid4())
-                                workflowobj[streamtype] = filename
-                            out['type'] = 'File'
-                            out['outputBinding'] = {'glob': filename}
-            if 'inputs' in workflowobj:
-                for inp in workflowobj['inputs']:
-                    if inp['type'] == 'stdin':
-                        if 'inputBinding' in inp:
+        if workflowobj.get('class') == 'CommandLineTool':
+            for out in workflowobj.get('outputs', []):
+                for streamtype in ['stdout', 'stderr']:
+                    if out.get('type') == streamtype:
+                        if 'outputBinding' in out:
                             raise ValidationException(
-                                "Not allowed to specify inputBinding when"
-                                " using stdin shortcut.")
-                        if 'stdin' in workflowobj:
-                            raise ValidationException(
-                                "Not allowed to specify stdin path when"
-                                " using stdin type shortcut.")
+                                "Not allowed to specify outputBinding when"
+                                " using %s shortcut." % streamtype)
+                        if streamtype in workflowobj:
+                            filename = workflowobj[streamtype]
                         else:
-                            workflowobj['stdin'] = \
-                                "$(inputs.%s.path)" % \
-                                inp['id'].rpartition('#')[2]
-                            inp['type'] = 'File'
+                            filename = Text(uuid.uuid4())
+                            workflowobj[streamtype] = filename
+                        out['type'] = 'File'
+                        out['outputBinding'] = {'glob': filename}
+            for inp in workflowobj.get('inputs', []):
+                if inp.get('type') == 'stdin':
+                    if 'inputBinding' in inp:
+                        raise ValidationException(
+                            "Not allowed to specify inputBinding when"
+                            " using stdin shortcut.")
+                    if 'stdin' in workflowobj:
+                        raise ValidationException(
+                            "Not allowed to specify stdin path when"
+                            " using stdin type shortcut.")
+                    else:
+                        workflowobj['stdin'] = \
+                            "$(inputs.%s.path)" % \
+                            inp['id'].rpartition('#')[2]
+                        inp['type'] = 'File'
         else:
             for entry in itervalues(workflowobj):
                 _convert_stdstreams_to_files(entry)

cwltool/main.py

@@ -20,7 +20,7 @@
 
 from . import draft2tool
 from . import workflow
-from .pathmapper import adjustDirObjs, get_listing, adjustFileObjs, trim_listing
+from .pathmapper import adjustDirObjs, get_listing, adjustFileObjs, trim_listing, visit_class
 from .cwlrdf import printrdf, printdot
 from .errors import WorkflowException, UnsupportedRequirement
 from .load_tool import fetch_document, validate_document, make_tool
@@ -503,8 +503,7 @@ def pathToLoc(p):
             p["location"] = p["path"]
             del p["path"]
 
-    adjustDirObjs(job_order_object, pathToLoc)
-    adjustFileObjs(job_order_object, pathToLoc)
+    visit_class(job_order_object, ("File", "Directory"), pathToLoc)
     adjustDirObjs(job_order_object, trim_listing)
     normalizeFilesDirs(job_order_object)
 
@@ -548,8 +547,7 @@ def loadref(b, u):
         else:
             raise Exception(u"Unknown relative_deps %s" % relative_deps)
 
-        adjustFileObjs(deps, functools.partial(makeRelative, base))
-        adjustDirObjs(deps, functools.partial(makeRelative, base))
+        visit_class(deps, ("File", "Directory"), functools.partial(makeRelative, base))
 
     stdout.write(json.dumps(deps, indent=4))
 
@@ -774,8 +772,7 @@ def locToPath(p):
                     if p["location"].startswith("file://"):
                         p["path"] = uri_file_path(p["location"])
 
-                adjustDirObjs(out, locToPath)
-                adjustFileObjs(out, locToPath)
+                visit_class(out, ("File", "Directory"), locToPath)
 
                 if isinstance(out, basestring):
                     stdout.write(out)

cwltool/pathmapper.py

@@ -8,7 +8,7 @@
 import schema_salad.validate as validate
 from schema_salad.ref_resolver import uri_file_path
 from schema_salad.sourceline import SourceLine
-from typing import Any, Callable, Set, Text, Tuple, Union
+from typing import Any, Callable, Set, Text, Tuple, Union, Iterable
 from six.moves import urllib
 
 from .stdfsaccess import abspath, StdFsAccess
@@ -30,33 +30,26 @@ def adjustFiles(rec, op):  # type: (Any, Union[Callable[..., Any], partial[Any]]
         for d in rec:
             adjustFiles(d, op)
 
-
-def adjustFileObjs(rec, op):  # type: (Any, Union[Callable[..., Any], partial[Any]]) -> None
-    """Apply an update function to each File object in the object `rec`."""
+def visit_class(rec, cls, op):  # type: (Any, Iterable, Union[Callable[..., Any], partial[Any]]) -> None
+    """Apply a function to with "class" in cls."""
 
     if isinstance(rec, dict):
-        if rec.get("class") == "File":
+        if rec.get("class") in cls:
             op(rec)
         for d in rec:
-            adjustFileObjs(rec[d], op)
+            visit_class(rec[d], cls, op)
     if isinstance(rec, list):
         for d in rec:
-            adjustFileObjs(d, op)
+            visit_class(d, cls, op)
 
+def adjustFileObjs(rec, op):  # type: (Any, Union[Callable[..., Any], partial[Any]]) -> None
+    """Apply an update function to each File object in the object `rec`."""
+    visit_class(rec, ("File",), op)
 
 def adjustDirObjs(rec, op):
     # type: (Any, Union[Callable[..., Any], partial[Any]]) -> None
     """Apply an update function to each Directory object in the object `rec`."""
-
-    if isinstance(rec, dict):
-        if rec.get("class") == "Directory":
-            op(rec)
-        for key in rec:
-            adjustDirObjs(rec[key], op)
-    if isinstance(rec, list):
-        for d in rec:
-            adjustDirObjs(d, op)
-
+    visit_class(rec, ("Directory",), op)
 
 def normalizeFilesDirs(job):
     # type: (Union[List[Dict[Text, Any]], Dict[Text, Any]]) -> None
@@ -84,8 +77,7 @@ def addLocation(d):
         if "basename" not in d:
             d["basename"] = os.path.basename(urllib.request.url2pathname(path))
 
-    adjustFileObjs(job, addLocation)
-    adjustDirObjs(job, addLocation)
+    visit_class(job, ("File", "Directory"), addLocation)
 
 
 def dedup(listing):  # type: (List[Any]) -> List[Any]

cwltool/process.py

@@ -26,10 +26,10 @@
 from typing import (Any, AnyStr, Callable, cast, Dict, List, Generator, Text,
                     Tuple, Union)
 
-from .builder import Builder, adjustFileObjs
+from .builder import Builder
 from .pathmapper import adjustDirObjs, get_listing
 from .errors import WorkflowException, UnsupportedRequirement
-from .pathmapper import PathMapper, normalizeFilesDirs
+from .pathmapper import PathMapper, normalizeFilesDirs, visit_class
 from .stdfsaccess import StdFsAccess
 from .utils import aslist, get_feature
 
@@ -189,7 +189,7 @@ def stageFiles(pm, stageFunc, ignoreWritable=False):
             continue
         if not os.path.exists(os.path.dirname(p.target)):
             os.makedirs(os.path.dirname(p.target), 0o0755)
-        if p.type in ("File", "Directory") and p.resolved.startswith("/"):
+        if p.type in ("File", "Directory") and (p.resolved.startswith("/") or p.resolved.startswith("file:///")):
             stageFunc(p.resolved, p.target)
         elif p.type == "Directory" and not os.path.exists(p.target) and p.resolved.startswith("_:"):
             os.makedirs(p.target, 0o0755)
@@ -234,7 +234,10 @@ def moveIt(src, dst):
                     shutil.move(src, dst)
                     return
         _logger.debug("Copying %s to %s", src, dst)
-        shutil.copy(src, dst)
+        if os.path.isdir(src):
+            shutil.copytree(src, dst)
+        else:
+            shutil.copy(src, dst)
 
     outfiles = []  # type: List[Dict[Text, Any]]
     collectFilesAndDirs(outputObj, outfiles)
@@ -245,12 +248,11 @@ def _check_adjust(f):
         f["location"] = file_uri(pm.mapper(f["location"])[1])
         if "contents" in f:
             del f["contents"]
-        if f["class"] == "File":
-            compute_checksums(fs_access, f)
         return f
 
-    adjustFileObjs(outputObj, _check_adjust)
-    adjustDirObjs(outputObj, _check_adjust)
+    visit_class(outputObj, ("File", "Directory"), _check_adjust)
+
+    visit_class(outputObj, ("File",), functools.partial(compute_checksums, fs_access))
 
     return outputObj
 

tests/test_check.py

@@ -0,0 +1,15 @@
+import unittest
+
+import cwltool.expression as expr
+import cwltool.factory
+import cwltool.pathmapper
+import cwltool.process
+import cwltool.workflow
+from .util import get_data
+from cwltool.main import main
+
+class TestCheck(unittest.TestCase):
+    def test_output_checking(self):
+        self.assertEquals(main([get_data('tests/wf/badout1.cwl')]), 1)
+        self.assertEquals(main([get_data('tests/wf/badout2.cwl')]), 1)
+        self.assertEquals(main([get_data('tests/wf/badout3.cwl')]), 1)

tests/test_pack.py

@@ -7,7 +7,7 @@
 import cwltool.workflow
 from cwltool.load_tool import fetch_document, validate_document
 from cwltool.main import makeRelative
-from cwltool.process import adjustFileObjs, adjustDirObjs
+from cwltool.pathmapper import adjustFileObjs, adjustDirObjs
 from .util import get_data
 
 class TestPack(unittest.TestCase):

tests/wf/badout1.cwl

@@ -0,0 +1,13 @@
+class: CommandLineTool
+cwlVersion: v1.0
+baseCommand: touch
+arguments: [file1]
+requirements:
+  InlineJavascriptRequirement: {}
+inputs: []
+outputs:
+  out:
+    type: File
+    outputBinding:
+      outputEval: |
+        $({"class": "File", "path": runtime.outdir+"/file2"})