Merge pull request #49 from achave11/tests-fixes

Tests fixes for #48, includes #48 changes and fixes for #48

Author: achave11-ucsc

test/test_integration.py

@@ -15,6 +15,7 @@
 
 class IntegrationTest(unittest.TestCase):
     """A baseclass that's inherited for use with different cwl backends."""
+
     def setUp(self):
         """Start a (local) wes-service server to make requests against."""
         raise NotImplementedError
@@ -43,31 +44,53 @@ def test_dockstore_md5sum(self):
     def test_local_md5sum(self):
         """Pass a local md5sum cwl to the wes-service server, and check for the correct output."""
         cwl_local_path = os.path.abspath('testdata/md5sum.cwl')
-        output_filepath, _ = run_cwl_md5sum(cwl_input='file://' + cwl_local_path)
+        workflow_attachment_path = os.path.abspath('testdata/dockstore-tool-md5sum.cwl')
+        output_filepath, _ = run_cwl_md5sum(cwl_input='file://' + cwl_local_path,
+                                            workflow_attachment='file://' + workflow_attachment_path)
 
         self.assertTrue(check_for_file(output_filepath), 'Output file was not found: ' + str(output_filepath))
 
     def test_multipart_upload(self):
         """Pass a local md5sum cwl to the wes-service server, and check for uploaded file in service."""
         cwl_local_path = os.path.abspath('testdata/md5sum.cwl')
-        _, run_id = run_cwl_md5sum(cwl_input='file://' + cwl_local_path)
+        workflow_attachment_path = os.path.abspath('testdata/dockstore-tool-md5sum.cwl')
+        out_file_path, run_id = run_cwl_md5sum(cwl_input='file://' + cwl_local_path,
+                                               workflow_attachment='file://' + workflow_attachment_path)
 
         get_response = get_log_request(run_id)["request"]
 
+        self.assertTrue(check_for_file(out_file_path), 'Output file was not found: '
+                        + get_response["workflow_attachment"])
         self.assertTrue(check_for_file(get_response["workflow_url"][7:]), 'Output file was not found: '
                         + get_response["workflow_url"][:7])
 
+    def test_run_attachments(self):
+        """Pass a local md5sum cwl to the wes-service server, check for attachments."""
+        cwl_local_path = os.path.abspath('testdata/md5sum.cwl')
+        workflow_attachment_path = os.path.abspath('testdata/dockstore-tool-md5sum.cwl')
+        out_file_path, run_id = run_cwl_md5sum(cwl_input='file://' + cwl_local_path,
+                                               workflow_attachment='file://' + workflow_attachment_path)
+
+        get_response = get_log_request(run_id)["request"]
+        attachment_tool_path = get_response["workflow_attachment"][7:] + "/dockstore-tool-md5sum.cwl"
+        self.assertTrue(check_for_file(out_file_path), 'Output file was not found: '
+                        + get_response["workflow_attachment"])
+        self.assertTrue(check_for_file(attachment_tool_path), 'Attachment file was not found: '
+                        + get_response["workflow_attachment"])
+
 
-def run_cwl_md5sum(cwl_input):
+def run_cwl_md5sum(cwl_input, workflow_attachment=None):
     """Pass a local md5sum cwl to the wes-service server, and return the path of the output file that was created."""
     endpoint = 'http://localhost:8080/ga4gh/wes/v1/runs'
     params = {'output_file': {'path': '/tmp/md5sum.txt', 'class': 'File'},
               'input_file': {'path': os.path.abspath('testdata/md5sum.input'), 'class': 'File'}}
 
     parts = [("workflow_params", json.dumps(params)), ("workflow_type", "CWL"), ("workflow_type_version", "v1.0")]
     if cwl_input.startswith("file://"):
-        parts.append(("workflow_descriptor", ("md5sum.cwl", open(cwl_input[7:], "rb"))))
+        parts.append(("workflow_attachment", ("md5sum.cwl", open(cwl_input[7:], "rb"))))
         parts.append(("workflow_url", os.path.basename(cwl_input[7:])))
+        if workflow_attachment:
+            parts.append(("workflow_attachment", ("dockstore-tool-md5sum.cwl", open(workflow_attachment[7:], "rb"))))
     else:
         parts.append(("workflow_url", cwl_input))
     response = requests.post(endpoint, files=parts).json()

testdata/md5sum.cwl

@@ -11,8 +11,7 @@ outputs:
 
 steps:
   md5sum:
-    run: https://raw.githubusercontent.com/common-workflow-language/workflow-service/master/testdata/dockstore-tool-md5sum.cwl
+    run: dockstore-tool-md5sum.cwl
     in:
       input_file: input_file
     out: [output_file]
-

wes_client/wes_client_main.py

@@ -128,7 +128,7 @@ def fixpaths(d):
     ]
     if workflow_url.startswith("file://"):
         # with open(workflow_url[7:], "rb") as f:
-        #     body["workflow_descriptor"] = f.read()
+        #     body["workflow_attachment"] = f.read()
         rootdir = os.path.dirname(workflow_url[7:])
         dirpath = rootdir
         # for dirpath, dirnames, filenames in os.walk(rootdir):
@@ -137,7 +137,7 @@ def fixpaths(d):
                 continue
             fn = os.path.join(dirpath, f)
             if os.path.isfile(fn):
-                parts.append(('workflow_descriptor', (fn[len(rootdir)+1:], open(fn, "rb"))))
+                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))

wes_service/arvados_wes.py

@@ -109,7 +109,7 @@ def ListRuns(self, page_size=None, page_token=None, state_search=None):
         }
 
     def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
-                          env, workflow_descriptor_file, project_uuid,
+                          env, project_uuid,
                           tempdir):
         api = arvados.api_from_config(version="v1", apiconfig={
             "ARVADOS_API_HOST": env["ARVADOS_API_HOST"],
@@ -150,9 +150,6 @@ def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
         except subprocess.CalledProcessError as e:
             api.container_requests().update(uuid=cr_uuid, body={"priority": 0,
                                                                 "properties": {"arvados-cwl-runner-log": str(e)}}).execute()
-        finally:
-            if workflow_descriptor_file is not None:
-                workflow_descriptor_file.close()
 
     @catch_exceptions
     def RunWorkflow(self, **args):
@@ -184,20 +181,13 @@ def RunWorkflow(self, **args):
                                                     "priority": 500}}).execute()
 
         workflow_url = body.get("workflow_url")
-        workflow_descriptor_file = None
-        if body.get("workflow_descriptor"):
-            workflow_descriptor_file = tempfile.NamedTemporaryFile()
-            workflow_descriptor_file.write(body.get('workflow_descriptor'))
-            workflow_descriptor_file.flush()
-            workflow_url = workflow_descriptor_file.name
 
         project_uuid = body.get("workflow_engine_parameters", {}).get("project_uuid")
 
         threading.Thread(target=self.invoke_cwl_runner, args=(cr["uuid"],
                                                               workflow_url,
                                                               body["workflow_params"],
                                                               env,
-                                                              workflow_descriptor_file,
                                                               project_uuid,
                                                               tempdir)).start()
 

wes_service/cwl_runner.py

@@ -2,7 +2,6 @@
 import json
 import os
 import subprocess
-import urllib
 import uuid
 
 from wes_service.util import WESBackend
@@ -25,7 +24,7 @@ def run(self, request, opts):
         CWL (url):
         request["workflow_url"] == a url to a cwl file
         or
-        request["workflow_descriptor"] == input cwl text (written to a file and a url constructed for that file)
+        request["workflow_attachment"] == input cwl text (written to a file and a url constructed for that file)
 
         JSON File:
         request["workflow_params"] == input json text (to be written to a file)
@@ -46,13 +45,7 @@ def run(self, request, opts):
                 self.workdir, "cwl.input.json"), "w") as inputtemp:
             json.dump(request["workflow_params"], inputtemp)
 
-        if request.get("workflow_descriptor"):
-            workflow_descriptor = request.get('workflow_descriptor')
-            with open(os.path.join(self.workdir, "workflow.cwl"), "w") as f:
-                f.write(workflow_descriptor)
-            workflow_url = urllib.pathname2url(os.path.join(self.workdir, "workflow.cwl"))
-        else:
-            workflow_url = request.get("workflow_url")
+        workflow_url = request.get("workflow_url")  # Will always be local path to descriptor cwl, or url.
 
         output = open(os.path.join(self.workdir, "cwl.output.json"), "w")
         stderr = open(os.path.join(self.workdir, "stderr"), "w")

wes_service/toil_wes.py

@@ -4,7 +4,6 @@
 import subprocess
 import time
 import logging
-import urllib
 import uuid
 
 from multiprocessing import Process
@@ -38,13 +37,13 @@ def write_workflow(self, request, opts, wftype='cwl'):
         """Writes a cwl, wdl, or python file as appropriate from the request dictionary."""
         self.input_wf_filename = os.path.join(self.workdir, 'workflow.' + wftype)
 
-        if request.get("workflow_descriptor"):
-            workflow_descriptor = request.get('workflow_descriptor')
+        if request.get("workflow_attachment"):
+            workflow_attachment = request.get('workflow_attachment')
             with open(self.input_wf_filename, "w") as f:
-                f.write(workflow_descriptor)
-            workflow_url = urllib.pathname2url(self.input_wf_filename)
-        else:
-            workflow_url = request.get("workflow_url")
+                f.write(workflow_attachment)
+            # workflow_url = urllib.pathname2url(self.input_wf_filename)
+
+        workflow_url = request.get("workflow_url")
 
         extra = opts.getoptlist("extra")
         if wftype == 'cwl':
@@ -147,7 +146,7 @@ def run(self, request, opts):
         CWL (url):
         request["workflow_url"] == a url to a cwl file
         or
-        request["workflow_descriptor"] == input cwl text (written to a file and a url constructed for that file)
+        request["workflow_attachment"] == input cwl text (written to a file and a url constructed for that file)
 
         JSON File:
         request["workflow_params"] == input json text (to be written to a file)

wes_service/util.py

@@ -47,9 +47,10 @@ def collect_attachments(self):
         body = {}
         for k, ls in connexion.request.files.iterlists():
             for v in ls:
-                if k == "workflow_descriptor":
+                if k == "workflow_attachment":
                     filename = secure_filename(v.filename)
                     v.save(os.path.join(tempdir, filename))
+                    body[k] = "file://%s" % os.path.join(tempdir)  # Reference to tem working dir.
                 elif k in ("workflow_params", "tags", "workflow_engine_parameters"):
                     body[k] = json.loads(v.read())
                 else: