Merge branch '12526-cwl-priority' closes #12526

[arvados.git] / sdk / cwl / arvados_cwl / pathmapper.py

diff --git a/sdk/cwl/arvados_cwl/pathmapper.py b/sdk/cwl/arvados_cwl/pathmapper.py
index c8ae77a8b69013718c51b6b579774b9d95d985ef..998890a31c50acac0513479d0fad9675fd790647 100644 (file)
--- a/sdk/cwl/arvados_cwl/pathmapper.py
+++ b/sdk/cwl/arvados_cwl/pathmapper.py
@@ -58,6 +58,8 @@ class ArvPathMapper(PathMapper):
         if isinstance(src, basestring) and ArvPathMapper.pdh_dirpath.match(src):
             self._pathmap[src] = MapperEnt(src, self.collection_pattern % urllib.unquote(src[5:]), srcobj["class"], True)
 
+        debug = logger.isEnabledFor(logging.DEBUG)
+
         if src not in self._pathmap:
             if src.startswith("file:"):
                 # Local FS ref, may need to be uploaded or may be on keep
@@ -67,7 +69,7 @@ class ArvPathMapper(PathMapper):
                                                    fnPattern="keep:%s/%s",
                                                    dirPattern="keep:%s/%s",
                                                    raiseOSError=True)
-                with SourceLine(srcobj, "location", WorkflowException):
+                with SourceLine(srcobj, "location", WorkflowException, debug):
                     if isinstance(st, arvados.commands.run.UploadFile):
                         uploadfiles.add((src, ab, st))
                     elif isinstance(st, arvados.commands.run.ArvFile):
@@ -82,25 +84,26 @@ class ArvPathMapper(PathMapper):
             else:
                 self._pathmap[src] = MapperEnt(src, src, srcobj["class"], True)
 
-        with SourceLine(srcobj, "secondaryFiles", WorkflowException):
+        with SourceLine(srcobj, "secondaryFiles", WorkflowException, debug):
             for l in srcobj.get("secondaryFiles", []):
                 self.visit(l, uploadfiles)
-        with SourceLine(srcobj, "listing", WorkflowException):
+        with SourceLine(srcobj, "listing", WorkflowException, debug):
             for l in srcobj.get("listing", []):
                 self.visit(l, uploadfiles)
 
-    def addentry(self, obj, c, path, subdirs):
+    def addentry(self, obj, c, path, remap):
         if obj["location"] in self._pathmap:
             src, srcpath = self.arvrunner.fs_access.get_collection(self._pathmap[obj["location"]].resolved)
             if srcpath == "":
                 srcpath = "."
             c.copy(srcpath, path + "/" + obj["basename"], source_collection=src, overwrite=True)
+            remap.append((obj["location"], path + "/" + obj["basename"]))
             for l in obj.get("secondaryFiles", []):
-                self.addentry(l, c, path, subdirs)
+                self.addentry(l, c, path, remap)
         elif obj["class"] == "Directory":
             for l in obj.get("listing", []):
-                self.addentry(l, c, path + "/" + obj["basename"], subdirs)
-            subdirs.append((obj["location"], path + "/" + obj["basename"]))
+                self.addentry(l, c, path + "/" + obj["basename"], remap)
+            remap.append((obj["location"], path + "/" + obj["basename"]))
         elif obj["location"].startswith("_:") and "contents" in obj:
             with c.open(path + "/" + obj["basename"], "w") as f:
                 f.write(obj["contents"].encode("utf-8"))
@@ -152,13 +155,13 @@ class ArvPathMapper(PathMapper):
             self._pathmap[loc] = MapperEnt(urllib.quote(fn, "/:+@"), self.collection_pattern % fn[5:], cls, True)
 
         for srcobj in referenced_files:
-            subdirs = []
+            remap = []
             if srcobj["class"] == "Directory" and srcobj["location"] not in self._pathmap:
                 c = arvados.collection.Collection(api_client=self.arvrunner.api,
                                                   keep_client=self.arvrunner.keep_client,
                                                   num_retries=self.arvrunner.num_retries)
                 for l in srcobj.get("listing", []):
-                    self.addentry(l, c, ".", subdirs)
+                    self.addentry(l, c, ".", remap)
 
                 check = self.arvrunner.api.collections().list(filters=[["portable_data_hash", "=", c.portable_data_hash()]], limit=1).execute(num_retries=self.arvrunner.num_retries)
                 if not check["items"]:
@@ -172,7 +175,7 @@ class ArvPathMapper(PathMapper):
                 c = arvados.collection.Collection(api_client=self.arvrunner.api,
                                                   keep_client=self.arvrunner.keep_client,
                                                   num_retries=self.arvrunner.num_retries                                                  )
-                self.addentry(srcobj, c, ".", subdirs)
+                self.addentry(srcobj, c, ".", remap)
 
                 check = self.arvrunner.api.collections().list(filters=[["portable_data_hash", "=", c.portable_data_hash()]], limit=1).execute(num_retries=self.arvrunner.num_retries)
                 if not check["items"]:
@@ -185,10 +188,13 @@ class ArvPathMapper(PathMapper):
                     ab = self.collection_pattern % c.portable_data_hash()
                     self._pathmap["_:" + unicode(uuid.uuid4())] = MapperEnt("keep:"+c.portable_data_hash(), ab, "Directory", True)
 
-            if subdirs:
-                for loc, sub in subdirs:
-                    # subdirs will all start with "./", strip it off
-                    ab = self.file_pattern % (c.portable_data_hash(), sub[2:])
+            if remap:
+                for loc, sub in remap:
+                    # subdirs start with "./", strip it off
+                    if sub.startswith("./"):
+                        ab = self.file_pattern % (c.portable_data_hash(), sub[2:])
+                    else:
+                        ab = self.file_pattern % (c.portable_data_hash(), sub)
                     self._pathmap[loc] = MapperEnt("keep:%s/%s" % (c.portable_data_hash(), sub[2:]),
                                                    ab, "Directory", True)
 
@@ -219,12 +225,16 @@ class StagingPathMapper(PathMapper):
         tgt = os.path.join(stagedir, obj["basename"])
         basetgt, baseext = os.path.splitext(tgt)
         n = 1
-        while tgt in self.targets:
-            n += 1
-            tgt = "%s_%i%s" % (basetgt, n, baseext)
+        if tgt in self.targets and (self.reversemap(tgt)[0] != loc):
+            while tgt in self.targets:
+                n += 1
+                tgt = "%s_%i%s" % (basetgt, n, baseext)
         self.targets.add(tgt)
         if obj["class"] == "Directory":
-            self._pathmap[loc] = MapperEnt(loc, tgt, "Directory", staged)
+            if obj.get("writable"):
+                self._pathmap[loc] = MapperEnt(loc, tgt, "WritableDirectory", staged)
+            else:
+                self._pathmap[loc] = MapperEnt(loc, tgt, "Directory", staged)
             if loc.startswith("_:") or self._follow_dirs:
                 self.visitlisting(obj.get("listing", []), tgt, basedir)
         elif obj["class"] == "File":
@@ -233,7 +243,7 @@ class StagingPathMapper(PathMapper):
             if "contents" in obj and loc.startswith("_:"):
                 self._pathmap[loc] = MapperEnt(obj["contents"], tgt, "CreateFile", staged)
             else:
-                if copy:
+                if copy or obj.get("writable"):
                     self._pathmap[loc] = MapperEnt(loc, tgt, "WritableFile", staged)
                 else:
                     self._pathmap[loc] = MapperEnt(loc, tgt, "File", staged)