Merge branch 'master' into prov_decoupling

Author: mr-c

cwltool/builder.py

@@ -192,10 +192,13 @@ def bind_input(self,
             binding = CommentedMap(schema["inputBinding"].items())
             assert binding is not None
 
+            bp = list(aslist(lead_pos))
             if "position" in binding:
-                binding["position"] = aslist(lead_pos) + aslist(binding["position"]) + aslist(tail_pos)
+                bp.extend(aslist(binding["position"]))
             else:
-                binding["position"] = aslist(lead_pos) + [0] + aslist(tail_pos)
+                bp.append(0)
+            bp.extend(aslist(tail_pos))
+            binding["position"] = bp
 
             binding["datum"] = datum
             if "valueFrom" in binding:

cwltool/executors.py

@@ -77,11 +77,12 @@ def execute(self,
         runtime_context.workflow_eval_lock = threading.Condition(threading.RLock())
 
         job_reqs = None
-        if "cwl:requirements" in job_order_object:
-            job_reqs = job_order_object["cwl:requirements"]
+        if "https://w3id.org/cwl/cwl#requirements" in job_order_object:
+            job_reqs = job_order_object["https://w3id.org/cwl/cwl#requirements"]
         elif ("cwl:defaults" in process.metadata
-              and "cwl:requirements" in process.metadata["cwl:defaults"]):
-            job_reqs = process.metadata["cwl:defaults"]["cwl:requirements"]
+              and "https://w3id.org/cwl/cwl#requirements"
+              in process.metadata["cwl:defaults"]):
+            job_reqs = process.metadata["cwl:defaults"]["https://w3id.org/cwl/cwl#requirements"]
         if job_reqs is not None:
             for req in job_reqs:
                 process.requirements.append(req)
@@ -95,7 +96,7 @@ def execute(self,
                 getdefault(runtime_context.compute_checksum, True),
                 path_mapper=runtime_context.path_mapper)
 
-        if runtime_context.rm_tmpdir is not None:
+        if runtime_context.rm_tmpdir:
             if runtime_context.cachedir is None:
                 output_dirs = self.output_dirs  # type: Iterable[Any]
             else:

cwltool/job.py

@@ -18,26 +18,23 @@
 
 import shellescape
 from prov.model import PROV
-from schema_salad.sourceline import SourceLine
 from six import PY2, with_metaclass
 from typing_extensions import (TYPE_CHECKING,  # pylint: disable=unused-import
                                Text)
-# move to a regular typing import when Python 3.3-3.6 is no longer supported
+from schema_salad.sourceline import SourceLine
 
 from .builder import Builder, HasReqsHints  # pylint: disable=unused-import
 from .context import RuntimeContext  # pylint: disable=unused-import
 from .context import getdefault
 from .errors import WorkflowException
 from .loghandler import _logger
-from .pathmapper import (MapperEnt, PathMapper, ensure_writable,
-                         ensure_non_writable)
-from .process import UnsupportedRequirement, stageFiles
+from .pathmapper import (MapperEnt, PathMapper,  # pylint: disable=unused-import
+                         ensure_writable, ensure_non_writable)
+from .process import UnsupportedRequirement, stage_files
 from .secrets import SecretStore  # pylint: disable=unused-import
-from .utils import \
-    bytes2str_in_dicts  # pylint: disable=unused-import; pylint: disable=unused-import
-from .utils import (DEFAULT_TMP_PREFIX, Directory, copytree_with_merge,
-                    json_dump, json_dumps, onWindows, processes_to_kill,
-                    subprocess)
+from .utils import (DEFAULT_TMP_PREFIX, Directory, bytes2str_in_dicts,
+                    copytree_with_merge, json_dump, json_dumps, onWindows,
+                    processes_to_kill, subprocess)
 
 if TYPE_CHECKING:
     from .provenance import ProvenanceProfile  # pylint: disable=unused-import
@@ -265,7 +262,6 @@ def _execute(self,
                 else:
                     stdin_path = rmap[1]
 
-
             stderr_path = None
             if self.stderr is not None:
                 abserr = os.path.join(self.outdir, self.stderr)
@@ -378,7 +374,7 @@ def _execute(self,
             _logger.debug(u"[job %s] Removing input staging directory %s", self.name, self.stagedir)
             shutil.rmtree(self.stagedir, True)
 
-        if runtimeContext.rm_tmpdir is not None:
+        if runtimeContext.rm_tmpdir:
             _logger.debug(u"[job %s] Removing temporary directory %s", self.name, self.tmpdir)
             shutil.rmtree(self.tmpdir, True)
 
@@ -407,14 +403,17 @@ def run(self,
         if "PATH" not in env:
             env["PATH"] = str(os.environ["PATH"]) if onWindows() else os.environ["PATH"]
         if "SYSTEMROOT" not in env and "SYSTEMROOT" in os.environ:
-            env["SYSTEMROOT"] = str(os.environ["SYSTEMROOT"]) if onWindows() else os.environ["SYSTEMROOT"]
+            env["SYSTEMROOT"] = str(os.environ["SYSTEMROOT"]) if onWindows() \
+                    else os.environ["SYSTEMROOT"]
 
-        stageFiles(self.pathmapper, ignoreWritable=True, symLink=True, secret_store=runtimeContext.secret_store)
+        stage_files(self.pathmapper, ignore_writable=True, symlink=True,
+                    secret_store=runtimeContext.secret_store)
         if self.generatemapper is not None:
-            stageFiles(self.generatemapper, ignoreWritable=self.inplace_update,
-                       symLink=True, secret_store=runtimeContext.secret_store)
-            relink_initialworkdir(self.generatemapper, self.outdir,
-                                  self.builder.outdir, inplace_update=self.inplace_update)
+            stage_files(self.generatemapper, ignore_writable=self.inplace_update,
+                        symlink=True, secret_store=runtimeContext.secret_store)
+            relink_initialworkdir(
+                self.generatemapper, self.outdir, self.builder.outdir,
+                inplace_update=self.inplace_update)
 
         self._execute([], env, runtimeContext)
 
@@ -478,12 +477,12 @@ def create_file_and_add_volume(self,
                                    volume,           # type: MapperEnt
                                    host_outdir_tgt,  # type: Optional[Text]
                                    secret_store      # type: Optional[SecretStore]
-                                  ):  # type: (...) -> None
+                                  ):  # type: (...) -> Text
         """Create the file and add a mapping."""
         if not host_outdir_tgt:
             new_file = os.path.join(
                 tempfile.mkdtemp(dir=self.tmpdir),
-                os.path.basename(volume.resolved))
+                os.path.basename(volume.target))
         writable = True if volume.type == "CreateWritableFile" else False
         if secret_store:
             contents = secret_store.retrieve(volume.resolved)
@@ -501,7 +500,7 @@ def create_file_and_add_volume(self,
             ensure_writable(host_outdir_tgt or new_file)
         else:
             ensure_non_writable(host_outdir_tgt or new_file)
-
+        return host_outdir_tgt or new_file
 
 
     def add_volumes(self,
@@ -513,7 +512,7 @@ def add_volumes(self,
         """Append volume mappings to the runtime option list."""
 
         container_outdir = self.builder.outdir
-        for vol in (itm[1] for itm in pathmapper.items() if itm[1].staged):
+        for key, vol in (itm for itm in pathmapper.items() if itm[1].staged):
             host_outdir_tgt = None  # type: Optional[Text]
             if vol.target.startswith(container_outdir + "/"):
                 host_outdir_tgt = os.path.join(
@@ -532,11 +531,12 @@ def add_volumes(self,
                 self.add_writable_directory_volume(
                     runtime, vol, host_outdir_tgt)
             elif vol.type in ["CreateFile", "CreateWritableFile"]:
-                self.create_file_and_add_volume(
+                new_path = self.create_file_and_add_volume(
                     runtime, vol, host_outdir_tgt, secret_store)
+                pathmapper.update(
+                    key, new_path, vol.target, vol.type, vol.staged)
 
-    def run(self, runtimeContext):
-        # type: (RuntimeContext) -> None
+    def run(self, runtimeContext):  # type: (RuntimeContext) -> None
         if not os.path.exists(self.tmpdir):
             os.makedirs(self.tmpdir)
         (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")

cwltool/load_tool.py

@@ -48,7 +48,7 @@
 overrides_ctx = {
     u"overrideTarget": {u"@type": u"@id"},
     u"cwltool": "http://commonwl.org/cwltool#",
-    u"overrides": {
+    u"http://commonwl.org/cwltool#overrides": {
         "@id": "cwltool:overrides",
         "mapSubject": "overrideTarget",
     },
@@ -226,7 +226,7 @@ def validate_document(document_loader,           # type: Loader
         uri = urllib.parse.urljoin(uri, workflowobj["https://w3id.org/cwl/cwl#tool"])
         del cast(dict, jobobj)["https://w3id.org/cwl/cwl#tool"]
 
-        if "http://commonwl.org/cwltool#overrides" in jobobj:
+        if isinstance(jobobj, CommentedMap) and "http://commonwl.org/cwltool#overrides" in jobobj:
             overrides.extend(resolve_overrides(jobobj, uri, uri))
             del jobobj["http://commonwl.org/cwltool#overrides"]
 
@@ -272,7 +272,7 @@ def validate_document(document_loader,           # type: Loader
     if isinstance(avsc_names, Exception):
         raise avsc_names
 
-    processobj = None  # type: Union[CommentedMap, CommentedSeq, Text]
+    processobj = None  # type: Union[CommentedMap, CommentedSeq, Text, None]
     document_loader = Loader(sch_document_loader.ctx, schemagraph=sch_document_loader.graph,
                              idx=document_loader.idx, cache=sch_document_loader.cache,
                              fetcher_constructor=fetcher_constructor, skip_schemas=skip_schemas)
@@ -285,7 +285,7 @@ def validate_document(document_loader,           # type: Loader
     if not isinstance(processobj, (CommentedMap, CommentedSeq)):
         raise ValidationException("Workflow must be a dict or list.")
 
-    if not new_metadata:
+    if not new_metadata and isinstance(processobj, CommentedMap):
         new_metadata = cast(CommentedMap, cmap(
             {"$namespaces": processobj.get("$namespaces", {}),
              "$schemas": processobj.get("$schemas", []),
@@ -387,7 +387,7 @@ def resolve_overrides(ov,      # Type: CommentedMap
         raise Exception("Expected CommentedMap, got %s" % type(ret))
     cwl_docloader = get_schema("v1.0")[0]
     cwl_docloader.resolve_all(ret, ov_uri)
-    return ret["overrides"]
+    return ret["http://commonwl.org/cwltool#overrides"]
 
 def load_overrides(ov, base_url):  # type: (Text, Text) -> List[Dict[Text, Any]]
     ovloader = Loader(overrides_ctx)