Symlink attachments. Add util to compose post.

Author: DailyDreaming

test/test_integration.py

@@ -10,11 +10,21 @@
 import shutil
 import logging
 
+from wes_client.util import build_wes_request
+
 logging.basicConfig(level=logging.INFO)
 
 
 class IntegrationTest(unittest.TestCase):
     """A baseclass that's inherited for use with different cwl backends."""
+    @classmethod
+    def setUpClass(cls):
+
+        cls.cwl_dockstore_url = 'https://dockstore.org:8443/api/ga4gh/v2/tools/quay.io%2Fbriandoconnor%2Fdockstore-tool-md5sum/versions/master/plain-CWL/descriptor/%2FDockstore.cwl'
+        cls.cwl_local_path = os.path.abspath('testdata/md5sum.cwl')
+        cls.json_input = "file://" + os.path.abspath('testdata/md5sum.json')
+        cls.attachments = ['file://' + os.path.abspath('testdata/md5sum.input'),
+                           'file://' + os.path.abspath('testdata/dockstore-tool-md5sum.cwl')]
 
     def setUp(self):
         """Start a (local) wes-service server to make requests against."""
@@ -35,72 +45,54 @@ def tearDown(self):
         unittest.TestCase.tearDown(self)
 
     def test_dockstore_md5sum(self):
-        """Fetch the md5sum cwl from dockstore, run it on the wes-service server, and check for the correct output."""
-        cwl_dockstore_url = 'https://dockstore.org:8443/api/ga4gh/v2/tools/quay.io%2Fbriandoconnor%2Fdockstore-tool-md5sum/versions/master/plain-CWL/descriptor/%2FDockstore.cwl'
-        output_filepath, _ = run_cwl_md5sum(cwl_input=cwl_dockstore_url)
-
-        self.assertTrue(check_for_file(output_filepath), 'Output file was not found: ' + str(output_filepath))
+        """HTTP md5sum cwl (dockstore), run it on the wes-service server, and check for the correct output."""
+        outfile_path, _ = run_cwl_md5sum(cwl_input=self.cwl_dockstore_url,
+                                         json_input=self.json_input,
+                                         workflow_attachment=self.attachments)
+        self.assertTrue(check_for_file(outfile_path), 'Output file was not found: ' + str(outfile_path))
 
     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')
-        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))
+        """LOCAL md5sum cwl to the wes-service server, and check for the correct output."""
+        outfile_path, run_id = run_cwl_md5sum(cwl_input=self.cwl_local_path,
+                                              json_input=self.json_input,
+                                              workflow_attachment=self.attachments)
+        self.assertTrue(check_for_file(outfile_path), 'Output file was not found: ' + str(outfile_path))
 
     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')
-        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)
-
+        """LOCAL md5sum cwl to the wes-service server, and check for uploaded file in service."""
+        outfile_path, run_id = run_cwl_md5sum(cwl_input=self.cwl_local_path,
+                                              json_input=self.json_input,
+                                              workflow_attachment=self.attachments)
         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])
+        self.assertTrue(check_for_file(outfile_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)
-
+        """LOCAL md5sum cwl to the wes-service server, check for attachments."""
+        outfile_path, run_id = run_cwl_md5sum(cwl_input=self.cwl_local_path,
+                                              json_input=self.json_input,
+                                              workflow_attachment=self.attachments)
         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"])
+        self.assertTrue(check_for_file(outfile_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, workflow_attachment=None):
+def run_cwl_md5sum(cwl_input, json_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_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))
+    parts = build_wes_request(cwl_input,
+                              json_input,
+                              attachments=workflow_attachment)
     response = requests.post(endpoint, files=parts).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']
 
 
 def run_wdl_md5sum(wdl_input):
     """Pass a local md5sum wdl to the wes-service server, and return the path of the output file that was created."""
-    endpoint = 'http://localhost:8080/ga4gh/wes/v1/workflows'
+    endpoint = 'http://localhost:8080/ga4gh/wes/v1/runs'
     params = '{"ga4ghMd5.inputFile": "' + os.path.abspath('testdata/md5sum.input') + '"}'
     parts = [("workflow_params", params),
              ("workflow_type", "WDL"),
@@ -136,8 +128,6 @@ def check_for_file(filepath, seconds=40):
     while not os.path.exists(filepath):
         time.sleep(1)
         wait_counter += 1
-        if os.path.exists(filepath):
-            return True
         if wait_counter > seconds:
             return False
     return True

testdata/md5sum.cwl

@@ -15,3 +15,4 @@ steps:
     in:
       input_file: input_file
     out: [output_file]
+

testdata/md5sum.json

@@ -0,0 +1,2 @@
+{"output_file": {"path": "/tmp/md5sum.txt", "class": "File"},
+ "input_file": {"path": "md5sum.input", "class": "File"}}

wes_client/util.py

@@ -0,0 +1,54 @@
+import os
+import json
+
+
+def wf_type(workflow_file):
+    if workflow_file.lower().endswith('wdl'):
+        return 'WDL'
+    elif workflow_file.lower().endswith('cwl'):
+        return 'CWL'
+    elif workflow_file.lower().endswith('py'):
+        return 'PY'
+    else:
+        raise ValueError('Unrecognized/unsupported workflow file extension: %s' % workflow_file.lower().split('.')[-1])
+
+
+def wf_version(workflow_file):
+    # TODO: Check inside of the file, handling local/http/etc.
+    if wf_type(workflow_file) == 'PY':
+        return '2.7'
+    # elif wf_type(workflow_file) == 'CWL':
+    #     # only works locally
+    #     return yaml.load(open(workflow_file))['cwlVersion']
+    else:
+        # TODO: actually check the wdl file
+        return "v1.0"
+
+
+def build_wes_request(workflow_file, json_path, attachments=None):
+    """
+    :param str workflow_file: Path to cwl/wdl file.  Can be http/https/file.
+    :param json_path: Path to accompanying json file.  Currently must be local.
+    :param attachments: Any other files needing to be uploaded to the server.
+
+    :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
+    json_path = json_path[7:] if json_path.startswith("file://") else json_path
+
+    parts = [("workflow_params", json.dumps(json.load(open(json_path)))),
+             ("workflow_type", wf_type(workflow_file)),
+             ("workflow_type_version", wf_version(workflow_file))]
+
+    if workflow_file.startswith("file://") or '://' not in workflow_file:
+        parts.append(("workflow_attachment", (os.path.basename(workflow_file[7:]), open(workflow_file[7:], "rb"))))
+        parts.append(("workflow_url", os.path.basename(workflow_file[7:])))
+    else:
+        parts.append(("workflow_url", workflow_file))
+
+    if attachments:
+        for attachment in attachments:
+            attachment = attachment[7:] if attachment.startswith("file://") else attachment
+            parts.append(("workflow_attachment", (os.path.basename(attachment), open(attachment, "rb"))))
+
+    return parts

wes_service/cwl_runner.py

@@ -8,10 +8,19 @@
 
 
 class Workflow(object):
-    def __init__(self, run_id):
+    def __init__(self, run_id, tempdir=None):
         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):
         """
@@ -34,10 +43,6 @@ def run(self, request, opts):
                                                  specifically the runner and runner options
         :return: {"run_id": self.run_id, "state": state}
         """
-        os.makedirs(self.workdir)
-        outdir = os.path.join(self.workdir, "outdir")
-        os.mkdir(outdir)
-
         with open(os.path.join(self.workdir, "request.json"), "w") as f:
             json.dump(request, f)
 
@@ -57,7 +62,7 @@ def run(self, request, opts):
                                 stdout=output,
                                 stderr=stderr,
                                 close_fds=True,
-                                cwd=outdir)
+                                cwd=self.outdir)
         output.close()
         stderr.close()
         with open(os.path.join(self.workdir, "pid"), "w") as pid:
@@ -175,7 +180,7 @@ def RunWorkflow(self, **args):
         tempdir, body = self.collect_attachments()
 
         run_id = uuid.uuid4().hex
-        job = Workflow(run_id)
+        job = Workflow(run_id, tempdir)
 
         job.run(body, self)
         return {"run_id": run_id}

wes_service/toil_wes.py

@@ -13,7 +13,7 @@
 
 
 class ToilWorkflow(object):
-    def __init__(self, run_id):
+    def __init__(self, run_id, tempdir=None):
         super(ToilWorkflow, self).__init__()
         self.run_id = run_id
 
@@ -22,6 +22,12 @@ def __init__(self, run_id):
         if not os.path.exists(self.outdir):
             os.makedirs(self.outdir)
 
+        if tempdir:
+            # tempdir is where attachments were downloaded, if 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))
+
         self.outfile = os.path.join(self.workdir, 'stdout')
         self.errfile = os.path.join(self.workdir, 'stderr')
         self.starttime = os.path.join(self.workdir, 'starttime')
@@ -266,7 +272,7 @@ def RunWorkflow(self):
         tempdir, body = self.collect_attachments()
 
         run_id = uuid.uuid4().hex
-        job = ToilWorkflow(run_id)
+        job = ToilWorkflow(run_id, tempdir)
         p = Process(target=job.run, args=(body, self))
         p.start()
         self.processes[run_id] = p

wes_service/util.py

@@ -50,17 +50,13 @@ def collect_attachments(self):
                 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.
+                    body[k] = "file://%s" % tempdir  # Reference to tem working dir.
                 elif k in ("workflow_params", "tags", "workflow_engine_parameters"):
                     body[k] = json.loads(v.read())
                 else:
                     body[k] = v.read()
 
-        if body['workflow_type'] != "CWL" or \
-                body['workflow_type_version'] != "v1.0":
-            return
-
         if ":" not in body["workflow_url"]:
             body["workflow_url"] = "file://%s" % os.path.join(tempdir, secure_filename(body["workflow_url"]))
 
-        return (tempdir, body)
+        return tempdir, body