Attachment modifications in cwltool and toil.

Author: DailyDreaming

test/test_integration.py

@@ -40,8 +40,8 @@ def tearDown(self):
                     time.sleep(3)
                 except OSError as e:
                     print(e)
-        if os.path.exists('workflows'):
-            shutil.rmtree('workflows')
+        # if os.path.exists('workflows'):
+        #     shutil.rmtree('workflows')
         unittest.TestCase.tearDown(self)
 
     def test_dockstore_md5sum(self):
@@ -85,7 +85,7 @@ def run_cwl_md5sum(cwl_input, json_input, workflow_attachment=None):
                               json_input,
                               attachments=workflow_attachment)
     response = requests.post(endpoint, files=parts).json()
-
+    assert 'run_id' in response, str(response.json())
     output_dir = os.path.abspath(os.path.join('workflows', response['run_id'], 'outdir'))
     return os.path.join(output_dir, 'md5sum.txt'), response['run_id']
 

wes_client/util.py

@@ -33,7 +33,7 @@ def build_wes_request(workflow_file, json_path, attachments=None):
 
     :return: A list of tuples formatted to be sent in a post to the wes-server (Swagger API).
     """
-    workflow_file = "file://" + workflow_file if "://" not in workflow_file else workflow_file
+    workflow_file = "file://" + workflow_file if ":" not in workflow_file else workflow_file
     json_path = json_path[7:] if json_path.startswith("file://") else json_path
 
     parts = [("workflow_params", json.dumps(json.load(open(json_path)))),

wes_client/wes_client_main.py

@@ -12,6 +12,7 @@
 import requests
 from requests.exceptions import InvalidSchema, MissingSchema
 from wes_service.util import visit
+from wes_client.util import build_wes_request
 from bravado.client import SwaggerClient
 from bravado.requests_client import RequestsClient
 
@@ -25,6 +26,7 @@ def main(argv=sys.argv[1:]):
                         help="Options: [http, https].  Defaults to WES_API_PROTO (https).")
     parser.add_argument("--quiet", action="store_true", default=False)
     parser.add_argument("--outdir", type=str)
+    parser.add_argument("--attachments", type=list, default=None)
     parser.add_argument("--page", type=str, default=None)
     parser.add_argument("--page-size", type=int, default=None)
 
@@ -81,15 +83,8 @@ def main(argv=sys.argv[1:]):
         json.dump(response.result(), sys.stdout, indent=4)
         return 0
 
-    if args.workflow_url.lower().endswith('wdl'):
-        wf_type = 'WDL'
-    elif args.workflow_url.lower().endswith('cwl'):
-        wf_type = 'CWL'
-    elif args.workflow_url.lower().endswith('py'):
-        wf_type = 'PY'
-
     if not args.job_order:
-        logging.error("Missing job order")
+        logging.error("Missing json/yaml file.")
         return 1
 
     loader = schema_salad.ref_resolver.Loader({
@@ -112,35 +107,12 @@ def fixpaths(d):
                 del d["path"]
     visit(input_dict, fixpaths)
 
-    workflow_url = args.workflow_url
-    if ":" not in workflow_url:
-        workflow_url = "file://" + os.path.abspath(workflow_url)
-
     if args.quiet:
         logging.basicConfig(level=logging.WARNING)
     else:
         logging.basicConfig(level=logging.INFO)
 
-    parts = [
-        ("workflow_params", json.dumps(input_dict)),
-        ("workflow_type", wf_type),
-        ("workflow_type_version", "v1.0")
-    ]
-    if workflow_url.startswith("file://"):
-        # with open(workflow_url[7:], "rb") as f:
-        #     body["workflow_attachment"] = f.read()
-        rootdir = os.path.dirname(workflow_url[7:])
-        dirpath = rootdir
-        # for dirpath, dirnames, filenames in os.walk(rootdir):
-        for f in os.listdir(rootdir):
-            if f.startswith("."):
-                continue
-            fn = os.path.join(dirpath, f)
-            if os.path.isfile(fn):
-                parts.append(('workflow_attachment', (fn[len(rootdir)+1:], open(fn, "rb"))))
-        parts.append(("workflow_url", os.path.basename(workflow_url[7:])))
-    else:
-        parts.append(("workflow_url", workflow_url))
+    parts = build_wes_request(args.workflow_url, args.job_order, attachments=args.attachments)
 
     postresult = http_client.session.post("%s://%s/ga4gh/wes/v1/runs" % (args.proto, args.host),
                                           files=parts,

wes_service/cwl_runner.py

@@ -8,21 +8,15 @@
 
 
 class Workflow(object):
-    def __init__(self, run_id, tempdir=None):
+    def __init__(self, run_id):
         super(Workflow, self).__init__()
         self.run_id = run_id
         self.workdir = os.path.join(os.getcwd(), "workflows", self.run_id)
         self.outdir = os.path.join(self.workdir, 'outdir')
         if not os.path.exists(self.outdir):
             os.makedirs(self.outdir)
 
-        if tempdir:
-            # tempdir is the folder where attachments were downloaded, if there were any
-            # symlink everything inside into self.workdir
-            for attachment in os.listdir(tempdir):
-                os.symlink(os.path.join(tempdir, attachment), os.path.join(self.workdir, attachment))
-
-    def run(self, request, opts):
+    def run(self, request, tempdir, opts):
         """
         Constructs a command to run a cwl/json from requests and opts,
         runs it, and deposits the outputs in outdir.
@@ -46,8 +40,7 @@ def run(self, request, opts):
         with open(os.path.join(self.workdir, "request.json"), "w") as f:
             json.dump(request, f)
 
-        with open(os.path.join(
-                self.workdir, "cwl.input.json"), "w") as inputtemp:
+        with open(os.path.join(self.workdir, "cwl.input.json"), "w") as inputtemp:
             json.dump(request["workflow_params"], inputtemp)
 
         workflow_url = request.get("workflow_url")  # Will always be local path to descriptor cwl, or url.
@@ -57,12 +50,27 @@ def run(self, request, opts):
 
         runner = opts.getopt("runner", default="cwl-runner")
         extra = opts.getoptlist("extra")
-        command_args = [runner] + extra + [workflow_url, inputtemp.name]
+
+        # replace any locally specified outdir with the default
+        for e in extra:
+            if e.startswith('--outdir='):
+                extra.remove(e)
+        extra.append('--outdir=' + self.outdir)
+
+        # link the cwl and json into the tempdir/cwd
+        if workflow_url.startswith('file://'):
+            os.link(workflow_url[7:], os.path.join(tempdir, "wes_workflow.cwl"))
+            workflow_url = os.path.join(tempdir, "wes_workflow.cwl")
+        os.link(inputtemp.name, os.path.join(tempdir, "cwl.input.json"))
+        jsonpath = os.path.join(tempdir, "cwl.input.json")
+
+        # build args and run
+        command_args = [runner] + extra + [workflow_url, jsonpath]
         proc = subprocess.Popen(command_args,
                                 stdout=output,
                                 stderr=stderr,
                                 close_fds=True,
-                                cwd=self.outdir)
+                                cwd=tempdir)
         output.close()
         stderr.close()
         with open(os.path.join(self.workdir, "pid"), "w") as pid:
@@ -180,9 +188,9 @@ def RunWorkflow(self, **args):
         tempdir, body = self.collect_attachments()
 
         run_id = uuid.uuid4().hex
-        job = Workflow(run_id, tempdir)
+        job = Workflow(run_id)
 
-        job.run(body, self)
+        job.run(body, tempdir, self)
         return {"run_id": run_id}
 
     def GetRunLog(self, run_id):