10401: Add integration tests for directory listing behavior.

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

diff --git a/sdk/cwl/arvados_cwl/runner.py b/sdk/cwl/arvados_cwl/runner.py
index ddeac3ab0691e0d33b05b2ce774a2805a8a13deb..1c10fe8196aba940495934169d559bb756cfda31 100644 (file)
--- a/sdk/cwl/arvados_cwl/runner.py
+++ b/sdk/cwl/arvados_cwl/runner.py
@@ -3,10 +3,9 @@ import urlparse
 from functools import partial
 import logging
 import json
-import re
 import subprocess
 
-from cStringIO import StringIO
+from StringIO import StringIO
 
 from schema_salad.sourceline import SourceLine
 
@@ -30,8 +29,6 @@ from . import done
 
 logger = logging.getLogger('arvados.cwl-runner')
 
-cwltool.draft2tool.ACCEPTLIST_RE = re.compile(r".*")
-
 def trim_listing(obj):
     """Remove 'listing' field from Directory objects that are keep references.
 
@@ -98,6 +95,10 @@ def upload_dependencies(arvrunner, name, document_loader,
     if include_primary and "id" in workflowobj:
         sc.append({"class": "File", "location": workflowobj["id"]})
 
+    if "$schemas" in workflowobj:
+        for s in workflowobj["$schemas"]:
+            sc.append({"class": "File", "location": s})
+
     mapper = ArvPathMapper(arvrunner, sc, "",
                            "keep:%s",
                            "keep:%s/%s",
@@ -109,11 +110,18 @@ def upload_dependencies(arvrunner, name, document_loader,
     adjustFileObjs(workflowobj, setloc)
     adjustDirObjs(workflowobj, setloc)
 
+    if "$schemas" in workflowobj:
+        sch = []
+        for s in workflowobj["$schemas"]:
+            sch.append(mapper.mapper(s).resolved)
+        workflowobj["$schemas"] = sch
+
     return mapper
 
 
 def upload_docker(arvrunner, tool):
-    """Visitor which uploads Docker images referenced in CommandLineTool objects."""
+    """Uploads Docker images used in CommandLineTool objects."""
+
     if isinstance(tool, CommandLineTool):
         (docker_req, docker_is_req) = get_feature(tool, "DockerRequirement")
         if docker_req:
@@ -122,6 +130,9 @@ def upload_docker(arvrunner, tool):
                 raise SourceLine(docker_req, "dockerOutputDirectory", UnsupportedRequirement).makeError(
                     "Option 'dockerOutputDirectory' of DockerRequirement not supported.")
             arv_docker_get_image(arvrunner.api, docker_req, True, arvrunner.project_uuid)
+    elif isinstance(tool, cwltool.workflow.Workflow):
+        for s in tool.steps:
+            upload_docker(arvrunner, s.embedded_tool)
 
 def packed_workflow(arvrunner, tool):
     """Create a packed workflow.
@@ -179,19 +190,21 @@ def upload_job_order(arvrunner, name, tool, job_order):
 
 def upload_workflow_deps(arvrunner, tool):
     # Ensure that Docker images needed by this workflow are available
-    tool.visit(partial(upload_docker, arvrunner))
+
+    upload_docker(arvrunner, tool)
 
     document_loader = tool.doc_loader
 
     def upload_tool_deps(deptool):
-        upload_dependencies(arvrunner,
-                            "%s dependencies" % (shortname(deptool["id"])),
-                            document_loader,
-                            deptool,
-                            deptool["id"],
-                            False,
-                            include_primary=False)
-        document_loader.idx[deptool["id"]] = deptool
+        if "id" in deptool:
+            upload_dependencies(arvrunner,
+                                "%s dependencies" % (shortname(deptool["id"])),
+                                document_loader,
+                                deptool,
+                                deptool["id"],
+                                False,
+                                include_primary=False)
+            document_loader.idx[deptool["id"]] = deptool
 
     tool.visit(upload_tool_deps)
 
@@ -204,6 +217,31 @@ def arvados_jobs_image(arvrunner, img):
         raise Exception("Docker image %s is not available\n%s" % (img, e) )
     return img
 
+def upload_workflow_collection(arvrunner, name, packed):
+    collection = arvados.collection.Collection(api_client=arvrunner.api,
+                                               keep_client=arvrunner.keep_client,
+                                               num_retries=arvrunner.num_retries)
+    with collection.open("workflow.cwl", "w") as f:
+        f.write(json.dumps(packed, indent=2, sort_keys=True, separators=(',',': ')))
+
+    filters = [["portable_data_hash", "=", collection.portable_data_hash()],
+               ["name", "like", name+"%"]]
+    if arvrunner.project_uuid:
+        filters.append(["owner_uuid", "=", arvrunner.project_uuid])
+    exists = arvrunner.api.collections().list(filters=filters).execute(num_retries=arvrunner.num_retries)
+
+    if exists["items"]:
+        logger.info("Using collection %s", exists["items"][0]["uuid"])
+    else:
+        collection.save_new(name=name,
+                            owner_uuid=arvrunner.project_uuid,
+                            ensure_unique_name=True,
+                            num_retries=arvrunner.num_retries)
+        logger.info("Uploaded to %s", collection.manifest_locator())
+
+    return collection.portable_data_hash()
+
+
 class Runner(object):
     """Base class for runner processes, which submit an instance of
     arvados-cwl-runner and wait for the final result."""