Tests working with latest toil.

Author: DailyDreaming

.travis.yml

@@ -3,7 +3,7 @@ python:
 - '2.7'
 before_install:
 - sudo apt-get update -qq
-- pip install toil[all]==3.16.0
+- virtualenv venv && . venv/bin/activate && git clone https://github.com/DataBiosphere/toil.git && cd toil && make prepare && make develop extras=[all] && cd ..
 - pip install . --process-dependency-links
 - pip install -r dev-requirements.txt
 script:

test/test_integration.py

@@ -19,12 +19,16 @@ class IntegrationTest(unittest.TestCase):
     """A baseclass that's inherited for use with different cwl backends."""
     @classmethod
     def setUpClass(cls):
-
+        # cwl
         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')]
+        cls.cwl_json_input = "file://" + os.path.abspath('testdata/md5sum.json')
+        cls.cwl_attachments = ['file://' + os.path.abspath('testdata/md5sum.input'),
+                               'file://' + os.path.abspath('testdata/dockstore-tool-md5sum.cwl')]
+        # wdl
+        cls.wdl_local_path = os.path.abspath('testdata/md5sum.wdl')
+        cls.wdl_json_input = "file://" + os.path.abspath('testdata/md5sum.wdl.json')
+        cls.wdl_attachments = ['file://' + os.path.abspath('testdata/md5sum.input')]
 
     def setUp(self):
         """Start a (local) wes-service server to make requests against."""
@@ -40,41 +44,32 @@ 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):
         """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)
+                                         json_input=self.cwl_json_input,
+                                         workflow_attachment=self.cwl_attachments)
         self.assertTrue(check_for_file(outfile_path), 'Output file was not found: ' + str(outfile_path))
 
     def test_local_md5sum(self):
         """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)
+                                              json_input=self.cwl_json_input,
+                                              workflow_attachment=self.cwl_attachments)
         self.assertTrue(check_for_file(outfile_path), 'Output file was not found: ' + str(outfile_path))
 
-    def test_multipart_upload(self):
-        """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(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):
         """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)
+                                              json_input=self.cwl_json_input,
+                                              workflow_attachment=self.cwl_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(outfile_path), 'Output file was not found: ' + get_response["workflow_attachment"])
+        attachment_tool_path = get_response["workflow_attachment"][7:] + "/dockstore-tool-md5sum.cwl"
         self.assertTrue(check_for_file(attachment_tool_path), 'Attachment file was not found: ' + get_response["workflow_attachment"])
 
 
@@ -90,25 +85,6 @@ def run_cwl_md5sum(cwl_input, json_input, workflow_attachment=None):
     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/runs'
-    params = '{"ga4ghMd5.inputFile": "' + os.path.abspath('testdata/md5sum.input') + '"}'
-    parts = [("workflow_params", params),
-             ("workflow_type", "WDL"),
-             ("workflow_type_version", "v1.0"),
-             ("workflow_url", wdl_input)]
-    response = requests.post(endpoint, files=parts).json()
-    output_dir = os.path.abspath(os.path.join('workflows', response['workflow_id'], 'outdir'))
-    check_travis_log = os.path.join(output_dir, 'stderr')
-    with open(check_travis_log, 'r') as f:
-        logging.info(f.read())
-    logging.info(subprocess.check_output(['ls', os.path.join('workflows', response['workflow_id'])]))
-    logging.info('\n')
-    logging.info(subprocess.check_output(['ls', output_dir]))
-    return os.path.join(output_dir, 'md5sum.txt'), response['workflow_id']
-
-
 def get_log_request(run_id):
     endpoint = 'http://localhost:8080/ga4gh/wes/v1/runs/{}'.format(run_id)
     return requests.get(endpoint).json()
@@ -159,6 +135,13 @@ def setUp(self):
                                                    shell=True)
         time.sleep(5)
 
+    def test_local_wdl(self):
+        """LOCAL md5sum wdl to the wes-service server, and check for the correct output."""
+        outfile_path, run_id = run_cwl_md5sum(cwl_input=self.wdl_local_path,
+                                              json_input=self.wdl_json_input,
+                                              workflow_attachment=self.wdl_attachments)
+        self.assertTrue(check_for_file(outfile_path), 'Output file was not found: ' + str(outfile_path))
+
 
 # Prevent pytest/unittest's discovery from attempting to discover the base test class.
 del IntegrationTest

wes_client/util.py

@@ -49,6 +49,8 @@ def build_wes_request(workflow_file, json_path, attachments=None):
     if attachments:
         for attachment in attachments:
             attachment = attachment[7:] if attachment.startswith("file://") else attachment
+            if ':' in attachment:
+                raise TypeError('Only local files supported for attachment: %s' % attachment)
             parts.append(("workflow_attachment", (os.path.basename(attachment), open(attachment, "rb"))))
 
     return parts

wes_service/toil_wes.py

@@ -28,11 +28,12 @@ def __init__(self, run_id):
         self.endtime = os.path.join(self.workdir, 'endtime')
         self.pidfile = os.path.join(self.workdir, 'pid')
         self.cmdfile = os.path.join(self.workdir, 'cmd')
+        self.jobstorefile = os.path.join(self.workdir, 'jobstore')
         self.request_json = os.path.join(self.workdir, 'request.json')
         self.output_json = os.path.join(self.workdir, "output.json")
         self.input_wf_filename = os.path.join(self.workdir, "wes_workflow.cwl")
         self.input_json = os.path.join(self.workdir, "wes_input.json")
-        self.jobstore_default = os.path.join(self.workdir, 'toiljobstore')
+        self.jobstore_default = os.path.join(self.workdir, 'file:toiljobstore')
         self.jobstore = None
 
     def sort_toil_options(self, extra):
@@ -43,13 +44,18 @@ def sort_toil_options(self, extra):
                 self.jobstore = e[11:]
                 if self.jobstore.startswith(('aws', 'google', 'azure')):
                     cloud = True
-            if e.startswith('--outdir='):
+            if e.startswith(('--outdir=', '-o=')):
                 extra.remove(e)
         if not cloud:
             extra.append('--outdir=' + self.outdir)
         if not self.jobstore:
-            extra.append('--jobStore=file:' + self.jobstore_default)
+            extra.append('--jobStore=' + self.jobstore_default)
             self.jobstore = self.jobstore_default
+
+        # store the jobstore location
+        with open(self.jobstorefile, 'w') as f:
+            f.write(self.jobstore)
+
         return extra
 
     def write_workflow(self, request, opts, cwd, wftype='cwl'):
@@ -79,8 +85,8 @@ def write_workflow(self, request, opts, cwd, wftype='cwl'):
 
     def write_json(self, request_dict):
         input_json = os.path.join(self.workdir, 'input.json')
-        with open(input_json, 'w') as inputtemp:
-            json.dump(request_dict['workflow_params'], inputtemp)
+        with open(input_json, 'w') as f:
+            json.dump(request_dict['workflow_params'], f)
         return input_json
 
     def call_cmd(self, cmd, cwd):
@@ -128,8 +134,8 @@ def getlog(self):
 
         outputobj = {}
         if state == "COMPLETE":
-            with open(self.output_json, "r") as outputtemp:
-                outputobj = json.load(outputtemp)
+            with open(self.output_json, "r") as f:
+                outputobj = json.load(f)
 
         return {
             "run_id": self.run_id,
@@ -209,42 +215,19 @@ def getstate(self):
         state = "RUNNING"
         exit_code = -1
 
-        exitcode_file = os.path.join(self.workdir, "exit_code")
-        pid_file = os.path.join(self.workdir, "pid")
-
-        if os.path.exists(exitcode_file):
-            with open(exitcode_file) as f:
-                exit_code = int(f.read())
-        elif os.path.exists(pid_file):
-            with open(pid_file, "r") as pid:
-                pid = int(pid.read())
-            try:
-                (_pid, exit_status) = os.waitpid(pid, os.WNOHANG)
-                if _pid != 0:
-                    exit_code = exit_status >> 8
-                    with open(exitcode_file, "w") as f:
-                        f.write(str(exit_code))
-                    os.unlink(pid_file)
-            except OSError:
-                os.unlink(pid_file)
-                exit_code = 255
-
-        if exit_code == 0:
-            state = "COMPLETE"
-        elif exit_code != -1:
-            state = "EXECUTOR_ERROR"
+        with open(self.jobstorefile, 'r') as f:
+            self.jobstore = f.read()
 
-        # Uncomment once https://github.com/DataBiosphere/toil/pull/2330 is merged
-        # logs = subprocess.check_output(['toil', 'status', 'file:' + self.jobstore, '--printLogs'])
-        # if 'ERROR:toil.worker:Exiting' in logs:
-        #     state = "EXECUTOR_ERROR"
-        #     exit_code = 255
-        # elif 'Root job is absent.  The workflow may have completed successfully.' in logs:
-        #     state = "COMPLETE"
-        #     exit_code = 0
-        # elif 'No job store found.' in logs:
-        #     state = "INITIALIZING"
-        #     exit_code = -1
+        logs = subprocess.check_output(['toil', 'status', 'file:' + self.jobstore, '--printLogs'])
+        if 'ERROR:toil.worker:Exiting' in logs:
+            state = "EXECUTOR_ERROR"
+            exit_code = 255
+        elif 'Root job is absent.  The workflow may have completed successfully.' in logs:
+            state = "COMPLETE"
+            exit_code = 0
+        elif 'No job store found.' in logs:
+            state = "INITIALIZING"
+            exit_code = -1
 
         return state, exit_code