When on_error=continue, do not execute downstream steps connected to an (#283)

* When on_error=continue, do not execute downstream steps connected to an
upstream step that failed.

* Log errors from fs_access while collecting output.

* Pin typed-ast to 0.6.3

Author: tetron

cwltool/draft2tool.py

@@ -461,6 +461,9 @@ def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=Tr
                                       for g in fs_access.glob(fs_access.join(outdir, gb))])
                         except (OSError, IOError) as e:
                             _logger.warn(Text(e))
+                        except:
+                            _logger.error("Unexpected error from fs_access", exc_info=True)
+                            raise
 
                 for files in r:
                     if files["class"] == "Directory" and "listing" not in files:

cwltool/workflow.py

@@ -19,7 +19,7 @@
 
 _logger = logging.getLogger("cwltool")
 
-WorkflowStateItem = namedtuple('WorkflowStateItem', ['parameter', 'value'])
+WorkflowStateItem = namedtuple('WorkflowStateItem', ['parameter', 'value', 'success'])
 
 
 def defaultMakeTool(toolpath_object,  # type: Dict[Text, Any]
@@ -161,7 +161,7 @@ def object_from_state(state, parms, frag_only, supportsMultipleInput, sourceFiel
                     "declared.")
             connections = aslist(inp[sourceField])
             for src in connections:
-                if src in state and state[src] is not None:
+                if src in state and state[src] is not None and (state[src].success == "success" or incomplete):
                     if not match_types(
                             inp["type"], state[src], iid, inputobj,
                             inp.get("linkMerge", ("merge_nested"
@@ -232,7 +232,7 @@ def receive_output(self, step, outputparms, jobout, processStatus):
         for i in outputparms:
             if "id" in i:
                 if i["id"] in jobout:
-                    self.state[i["id"]] = WorkflowStateItem(i, jobout[i["id"]])
+                    self.state[i["id"]] = WorkflowStateItem(i, jobout[i["id"]], processStatus)
                 else:
                     _logger.error(u"[%s] Output is missing expected field %s", step.name, i["id"])
                     processStatus = "permanentFail"
@@ -359,9 +359,9 @@ def job(self, joborder, output_callback, **kwargs):
             with SourceLine(self.tool["inputs"], e, WorkflowException):
                 iid = shortname(i["id"])
                 if iid in joborder:
-                    self.state[i["id"]] = WorkflowStateItem(i, copy.deepcopy(joborder[iid]))
+                    self.state[i["id"]] = WorkflowStateItem(i, copy.deepcopy(joborder[iid]), "success")
                 elif "default" in i:
-                    self.state[i["id"]] = WorkflowStateItem(i, copy.deepcopy(i["default"]))
+                    self.state[i["id"]] = WorkflowStateItem(i, copy.deepcopy(i["default"]), "success")
                 else:
                     raise WorkflowException(
                         u"Input '%s' not in input object and does not have a default value." % (i["id"]))

tests/wf/echo.cwl

@@ -12,6 +12,8 @@ inputs:
       else:
         f = open("foo"+sys.argv[1]+".txt", "w")
         f.write(sys.argv[1]+"\n")
+      if sys.argv[1] == "5":
+        exit(1)
 outputs:
   out:
     type: File

tests/wf/scatterfail.cwl

@@ -2,6 +2,7 @@ class: Workflow
 cwlVersion: v1.0
 requirements:
   ScatterFeatureRequirement: {}
+  SubworkflowFeatureRequirement: {}
 inputs:
   range:
     type: string[]
@@ -16,10 +17,22 @@ steps:
       r: range
     scatter: r
     out: [out]
-    run: echo.cwl
-  step2:
-    in:
-      r: step1/out
-    scatter: r
-    out: []
-    run: cat.cwl
+    run:
+      class: Workflow
+      inputs:
+        r: string
+      outputs:
+        out:
+          type: File
+          outputSource: sstep1/out
+      steps:
+        sstep1:
+          in:
+            r: r
+          out: [out]
+          run: echo.cwl
+        sstep2:
+          in:
+            r: sstep1/out
+          out: []
+          run: cat.cwl

tests/wf/wffail.cwl

@@ -1,13 +1,18 @@
 class: Workflow
 cwlVersion: v1.0
 inputs: []
+requirements:
+  StepInputExpressionRequirement: {}
 outputs:
   out1:
     type: File
     outputSource: step1/out
   out2:
     type: File
     outputSource: step2/out
+  out4:
+    type: File
+    outputSource: step4/out
 steps:
   step1:
     in:
@@ -19,3 +24,15 @@ steps:
       r:  {default: "2"}
     out: [out]
     run: echo.cwl
+  step3:
+    in:
+      r:  {default: "5"}
+    out: [out]
+    run: echo.cwl
+  step4:
+    in:
+      r:
+        source: step3/out
+        valueFrom: $(inputs.r.basename)
+    out: [out]
+    run: echo.cwl

tox.ini

@@ -15,7 +15,7 @@ commands = make mypy
 whitelist_externals = make
 deps =
   mypy>=0.470
-  typed-ast
+  typed-ast==0.6.3
   -rrequirements.txt
 
 [testenv:py35-lint]