Fix updater (#1051)

* Internally update and use v1.1 semantics

* Simplify interfaces of load_tool methods to make them less error prone

* Get loadListing from parameter schema, not inputBinding.

* Don't accidentally relocate symlinked files

* Fix loadListing on output parameter schema.

Author: tetron

cwltool/builder.py

@@ -135,8 +135,7 @@ def __init__(self,
                  loadListing,          # type: Text
                  outdir,               # type: Text
                  tmpdir,               # type: Text
-                 stagedir,             # type: Text
-                 cwl_version,          # type: Text
+                 stagedir             # type: Text
                 ):  # type: (...) -> None
 
         self.job = job
@@ -146,7 +145,6 @@ def __init__(self,
         self.names = names
         self.requirements = requirements
         self.hints = hints
-        self.cwl_version = cwl_version
         self.resources = resources
         self.mutation_manager = mutation_manager
         self.formatgraph = formatgraph
@@ -291,14 +289,10 @@ def _capture_files(f):
                     if "secondaryFiles" not in datum:
                         datum["secondaryFiles"] = []
                     for sf in aslist(schema["secondaryFiles"]):
-                        sf_required = True
-                        if isinstance(sf, MutableMapping) and "pattern" in sf and self.cwl_version in ['v1.1.0-dev1']:
-                            if 'required' in sf:
-                                sf_required = self.do_eval(sf['required'], context=datum)
-                        elif isinstance(sf, string_types):
-                            sf = {"pattern": sf}
+                        if 'required' in sf:
+                            sf_required = self.do_eval(sf['required'], context=datum)
                         else:
-                            raise validate.ValidationException("Not a secondary file definition: %s" % sf)
+                            sf_required = True
 
                         if "$(" in sf["pattern"] or "${" in sf["pattern"]:
                             sfpath = self.do_eval(sf["pattern"], context=datum)
@@ -318,7 +312,7 @@ def _capture_files(f):
                                 sf_location = datum["location"][0:datum["location"].rindex("/")+1]+sfname
                                 if isinstance(sfname, MutableMapping):
                                     datum["secondaryFiles"].append(sfname)
-                                elif discover_secondaryFiles and os.path.exists(uri_file_path(sf_location)):
+                                elif discover_secondaryFiles and self.fs_access.exists(sf_location):
                                     datum["secondaryFiles"].append({
                                         "location": sf_location,
                                         "basename": sfname,
@@ -341,7 +335,7 @@ def _capture_files(f):
                 visit_class(datum.get("secondaryFiles", []), ("File", "Directory"), _capture_files)
 
             if schema["type"] == "Directory":
-                ll = self.loadListing or (binding and binding.get("loadListing"))
+                ll = schema.get("loadListing") or self.loadListing
                 if ll and ll != "no_listing":
                     get_listing(self.fs_access, datum, (ll == "deep_listing"))
                 self.files.append(datum)

cwltool/checker.py

@@ -126,6 +126,14 @@ def _rec_fields(rec):  # type: (MutableMapping[Text, Any]) -> MutableMapping[Tex
             return False
     return True
 
+def missing_subset(fullset, subset):
+    # type: (List, List) -> List
+    missing = []
+    for i in subset:
+        if i not in fullset:
+            missing.append(i)
+    return missing
+
 def static_checker(workflow_inputs, workflow_outputs, step_inputs, step_outputs, param_to_step):
     # type: (List[Dict[Text, Any]], List[Dict[Text, Any]], List[Dict[Text, Any]], List[Dict[Text, Any]], Dict[Text, Dict[Text, Any]]) -> None
     """Check if all source and sink types of a workflow are compatible before run time.
@@ -152,19 +160,18 @@ def static_checker(workflow_inputs, workflow_outputs, step_inputs, step_outputs,
         src = warning.src
         sink = warning.sink
         linkMerge = warning.linkMerge
-        if sink.get("secondaryFiles") and sorted(
-                sink.get("secondaryFiles", [])) != sorted(src.get("secondaryFiles", [])):
-            msg1 = "Sink '%s'" % (shortname(sink["id"]))
-            msg2 = SourceLine(sink.get("_tool_entry", sink), "secondaryFiles").makeError(
-                "expects secondaryFiles: %s but" % (sink.get("secondaryFiles")))
-            if "secondaryFiles" in src:
-                msg3 = SourceLine(src, "secondaryFiles").makeError(
-                    "source '%s' has secondaryFiles %s." % (shortname(src["id"]), src.get("secondaryFiles")))
-            else:
-                msg3 = SourceLine(src, "id").makeError(
-                    "source '%s' does not include secondaryFiles." % (shortname(src["id"])))
-            msg4 = SourceLine(src, "id").makeError("To fix, add secondaryFiles: %s to definition of '%s'." % (sink.get("secondaryFiles"), shortname(src["id"])))
-            msg = SourceLine(sink).makeError("%s\n%s" % (msg1, bullets([msg2, msg3, msg4], "  ")))
+        sinksf = sorted([p["pattern"] for p in sink.get("secondaryFiles", []) if p.get("required", True)])
+        srcsf = sorted([p["pattern"] for p in src.get("secondaryFiles", [])])
+        # Every secondaryFile required by the sink, should be declared
+        # by the source
+        missing = missing_subset(srcsf, sinksf)
+        if missing:
+            msg1 = "Parameter '%s' requires secondaryFiles %s but" % (shortname(sink["id"]), missing)
+            msg3 = SourceLine(src, "id").makeError(
+                "source '%s' does not provide those secondaryFiles." % (shortname(src["id"])))
+            msg4 = SourceLine(src.get("_tool_entry", src), "secondaryFiles").makeError("To resolve, add missing secondaryFiles patterns to definition of '%s' or" % (shortname(src["id"])))
+            msg5 = SourceLine(sink.get("_tool_entry", sink), "secondaryFiles").makeError("mark missing secondaryFiles in definition of '%s' as optional." % shortname(sink["id"]))
+            msg = SourceLine(sink).makeError("%s\n%s" % (msg1, bullets([msg3, msg4, msg5], "  ")))
         elif sink.get("not_connected"):
             msg = SourceLine(sink, "type").makeError(
                 "'%s' is not an input parameter of %s, expected %s"

cwltool/command_line_tool.py

@@ -292,11 +292,7 @@ def job(self,
            ):
         # type: (...) -> Generator[Union[JobBase, CallbackJob], None, None]
 
-        require_prefix = ""
-        if self.metadata["cwlVersion"] == "v1.0":
-            require_prefix = "http://commonwl.org/cwltool#"
-
-        workReuse, _ = self.get_requirement(require_prefix + "WorkReuse")
+        workReuse, _ = self.get_requirement("WorkReuse")
         enableReuse = workReuse.get("enableReuse", True) if workReuse else True
 
         jobname = uniquename(runtimeContext.name or shortname(self.tool.get("id", "job")))
@@ -518,10 +514,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
             j.tmpdir = builder.tmpdir
             j.stagedir = builder.stagedir
 
-        inplaceUpdateReq, _ = self.get_requirement("http://commonwl.org/cwltool#InplaceUpdateRequirement")
-        if not inplaceUpdateReq and self.metadata["cwlVersion"] == "v1.1.0-dev1":
-            inplaceUpdateReq, _ = self.get_requirement("InplaceUpdateRequirement")
-
+        inplaceUpdateReq, _ = self.get_requirement("InplaceUpdateRequirement")
         if inplaceUpdateReq is not None:
             j.inplace_update = inplaceUpdateReq["inplaceUpdate"]
         normalizeFilesDirs(j.generatefiles)
@@ -553,16 +546,14 @@ def register_reader(f):
             adjustDirObjs(builder.files, register_reader)
             adjustDirObjs(builder.bindings, register_reader)
 
-        timelimit, _ = self.get_requirement(require_prefix + "TimeLimit")
+        timelimit, _ = self.get_requirement("TimeLimit")
         if timelimit is not None:
             with SourceLine(timelimit, "timelimit", validate.ValidationException, debug):
                 j.timelimit = builder.do_eval(timelimit["timelimit"])
                 if not isinstance(j.timelimit, int) or j.timelimit < 0:
                     raise Exception("timelimit must be an integer >= 0, got: %s" % j.timelimit)
 
-        if self.metadata["cwlVersion"] == "v1.0":
-            j.networkaccess = True
-        networkaccess, _ = self.get_requirement(require_prefix + "NetworkAccess")
+        networkaccess, _ = self.get_requirement("NetworkAccess")
         if networkaccess is not None:
             with SourceLine(networkaccess, "networkAccess", validate.ValidationException, debug):
                 j.networkaccess = builder.do_eval(networkaccess["networkAccess"])
@@ -711,7 +702,7 @@ def collect_output(self,
                     rfile = files.copy()
                     revmap(rfile)
                     if files["class"] == "Directory":
-                        ll = builder.loadListing or (binding and binding.get("loadListing"))
+                        ll = schema.get("loadListing") or builder.loadListing
                         if ll and ll != "no_listing":
                             get_listing(fs_access, files, (ll == "deep_listing"))
                     else:
@@ -762,34 +753,22 @@ def collect_output(self,
                             primary.setdefault("secondaryFiles", [])
                             pathprefix = primary["path"][0:primary["path"].rindex("/")+1]
                             for sf in aslist(schema["secondaryFiles"]):
-                                if isinstance(sf, MutableMapping) and 'pattern' in sf:
-                                    if 'required' in sf:
-                                        sf_required = sf['required']
-                                    else:
-                                        sf_required = False
-                                    sf = sf['pattern']
+                                if 'required' in sf:
+                                    sf_required = builder.do_eval(sf['required'], context=primary)
                                 else:
                                     sf_required = False
 
-                                if isinstance(sf, MutableMapping) or "$(" in sf or "${" in sf:
-                                    sfpath = builder.do_eval(sf, context=primary)
-                                    subst = False
+                                if "$(" in sf["pattern"] or "${" in sf["pattern"]:
+                                    sfpath = builder.do_eval(sf["pattern"], context=primary)
                                 else:
-                                    if sf.endswith('?') and \
-                                            self.metadata['cwlVersion'] in ['v1.1.0-dev1']:
-                                        sf_required = False
-                                        sf = sf[:-1]
-                                    sfpath = sf
-                                    subst = True
+                                    sfpath = substitute(primary["basename"], sf["pattern"])
+
                                 for sfitem in aslist(sfpath):
                                     if not sfitem:
                                         continue
                                     if isinstance(sfitem, string_types):
-                                        if subst:
-                                            sfitem = {"path": substitute(primary["path"], sfitem)}
-                                        else:
-                                            sfitem = {"path": pathprefix+sfitem}
-                                    if not os.path.exists(sfitem['path']) and sf_required:
+                                        sfitem = {"path": pathprefix+sfitem}
+                                    if not fs_access.exists(sfitem['path']) and sf_required:
                                         raise WorkflowException(
                                             "Missing required secondary file '%s'" % (
                                                 sfitem["path"]))

cwltool/context.py

@@ -64,6 +64,8 @@ def __init__(self, kwargs=None):
         self.host_provenance = False       # type: bool
         self.user_provenance = False       # type: bool
         self.prov_obj = None               # type: Optional[ProvenanceProfile]
+        self.do_update = True              # type: bool
+        self.jobdefaults = None            # type: Optional[MutableMapping[Text, Any]]
 
         super(LoadingContext, self).__init__(kwargs)
 

cwltool/executors.py

@@ -77,7 +77,7 @@ def execute(self,
 
         job_reqs = None
         if "https://w3id.org/cwl/cwl#requirements" in job_order_object:
-            if process.metadata["cwlVersion"] == 'v1.0':
+            if process.metadata.get("http://commonwl.org/cwltool#original_cwlVersion") == 'v1.0':
                 raise WorkflowException(
                     "`cwl:requirements` in the input object is not part of CWL "
                     "v1.0. You can adjust to use `cwltool:overrides` instead; or you "
@@ -87,7 +87,7 @@ def execute(self,
         elif ("cwl:defaults" in process.metadata
               and "https://w3id.org/cwl/cwl#requirements"
               in process.metadata["cwl:defaults"]):
-            if process.metadata["cwlVersion"] == 'v1.0':
+            if process.metadata.get("http://commonwl.org/cwltool#original_cwlVersion") == 'v1.0':
                 raise WorkflowException(
                     "`cwl:requirements` in the input object is not part of CWL "
                     "v1.0. You can adjust to use `cwltool:overrides` instead; or you "