Docstrings, --version, and a wes-client test. (#25)

* flake8 ignore line length.

* Pass flake8.

* Newline in .flake8

* Basic help descriptions.

* Docstrings to util.

* More docstrings.

* More docstrings.

* More docstrings.

* Refactor.

* Add --version to service.

* Add test for client.

* Add test to travis.

* Remove test cruft.

* Remove service test stub.

* flake8 fix.

* Change travis yml test run command.

* Remove test cruft.

* pip install future added to reqs.

* Different relative path for travis.

* Remove debugging ls from travis script.

* Remove redundant SIGTERM.

* Amend CLI entrypoints.

* Add nosetests to dev-requirements.txt and rename the test file.

* Add nosetests to the yml.

* Pytest instead of nosetests.

* Update requirements.

* subprocess32

* Test server stdout to DEVNULL.

* Test checks for output file and deletes outputs.

* Run the correct test in travis.

* Pytest.

* Add local file test.

* Amend local file path.

* Explicit dockstore url.

* Requirements given versions.  Add pycache to gitignore.

* Requirements.txt.

* Add testing for Toil.

* Minor naming.

* Dependencies in setup.py.

* Update .travis.yml

* Update .travis.yml

* Pytest in dev-requirements.txt

* Revert new files.

* Move files back out of init.

* flake8 changes.

* Better check.

Author: DailyDreaming

.gitignore

@@ -52,6 +52,9 @@ coverage.xml
 *.mo
 *.pot
 
+# PyTest
+.pytest_cache
+
 # PyCharm
 .idea/
 

.travis.yml

@@ -5,8 +5,10 @@ before_install:
 - sudo apt-get update -qq
 - pip install . --process-dependency-links
 - pip install -r dev-requirements.txt
+- pip install toil[all]==3.16.0
 script:
 - flake8 wes_service wes_client
+- pytest
 deploy:
   provider: pypi
   on:

dev-requirements.txt

@@ -1 +1,2 @@
 flake8
+pytest

passenger_wsgi.py

@@ -1,3 +1,3 @@
-import wes_service
+from wes_service.wes_service_main import setup
 
-application = wes_service.setup()
+application = setup()

setup.py

@@ -23,18 +23,22 @@
       package_data={'wes_service': ['openapi/workflow_execution_service.swagger.yaml']},
       include_package_data=True,
       install_requires=[
-          'connexion',
-          'bravado',
+          'future',
+          'connexion==1.4.2',
+          'bravado==10.1.0',
           'ruamel.yaml >= 0.12.4, < 0.15',
-          'cwlref-runner',
-          'schema-salad'
+          'cwlref-runner==1.0',
+          'schema-salad>=2.6, <3',
+          'subprocess32==3.5.2'
                         ],
       entry_points={
-          'console_scripts': ["wes-server=wes_service:main",
-                              "wes-client=wes_client:main"]
+          'console_scripts': ["wes-server=wes_service.wes_service_main:main",
+                              "wes-client=wes_client.wes_client_main:main"]
                     },
       extras_require={
           "arvados": ["arvados-cwl-runner"
-                      ]},
+                      ],
+          "toil": ["toil[all]==3.16.0"
+                   ]},
       zip_safe=False
       )

test/__init__.py

@@ -0,0 +1,109 @@
+from __future__ import absolute_import
+import unittest
+import time
+import os
+import subprocess32 as subprocess
+import signal
+import requests
+import shutil
+
+
+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
+
+    def tearDown(self):
+        """Kill the wes-service server."""
+        os.kill(self.wes_server_process.pid, signal.SIGTERM)
+        while get_server_pids():
+            for pid in get_server_pids():
+                try:
+                    os.kill(int(pid), signal.SIGKILL)
+                    time.sleep(3)
+                except OSError as e:
+                    print(e)
+
+        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_md5sum(cwl_input=cwl_dockstore_url)
+
+        self.assertTrue(check_for_file(output_filepath), 'Output file was not found: ' + str(output_filepath))
+        shutil.rmtree('workflows')
+
+    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_md5sum(cwl_input='file://' + cwl_local_path)
+
+        self.assertTrue(check_for_file(output_filepath), 'Output file was not found: ' + str(output_filepath))
+        shutil.rmtree('workflows')
+
+
+def run_md5sum(cwl_input):
+    """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/workflows'
+    params = {'output_file': {'path': '/tmp/md5sum.txt', 'class': 'File'}, 'input_file': {'path': '../../testdata/md5sum.input', 'class': 'File'}}
+    body = {'workflow_url': cwl_input, 'workflow_params': params, 'workflow_type': 'CWL', 'workflow_type_version': 'v1.0'}
+    response = requests.post(endpoint, json=body).json()
+    output_dir = os.path.abspath(os.path.join('workflows', response['workflow_id'], 'outdir'))
+    return os.path.join(output_dir, 'md5sum.txt')
+
+
+def get_server_pids():
+    try:
+        pids = subprocess.check_output(['pgrep', '-f', 'wes_service_main.py']).strip().split()
+    except subprocess.CalledProcessError:
+        return None
+    return pids
+
+
+def check_for_file(filepath, seconds=20):
+    """Return True if a file exists within a certain amount of time."""
+    wait_counter = 0
+    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
+
+
+class CwltoolTest(IntegrationTest):
+    """Test using cwltool."""
+    def setUp(self):
+        """
+        Start a (local) wes-service server to make requests against.
+        Use cwltool as the wes-service server 'backend'.
+        """
+        self.wes_server_process = subprocess.Popen('python {}'.format(os.path.abspath('wes_service/wes_service_main.py')),
+                                                   shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        time.sleep(5)
+
+
+class ToilTest(IntegrationTest):
+    """Test using Toil."""
+    def setUp(self):
+        """
+        Start a (local) wes-service server to make requests against.
+        Use toil as the wes-service server 'backend'.
+        """
+        self.wes_server_process = subprocess.Popen('python {} '
+                                                   '--opt runner=cwltoil --opt extra=--logLevel=CRITICAL'
+                                                   ''.format(os.path.abspath('wes_service/wes_service_main.py')),
+                                                   shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        time.sleep(5)
+
+
+# Prevent pytest/unittest's discovery from attempting to discover the base test class.
+del IntegrationTest
+
+
+if __name__ == '__main__':
+    unittest.main()  # run all tests

test/test_integration.py

@@ -1,183 +0,0 @@
-#!/usr/bin/env python
-import urlparse
-import pkg_resources  # part of setuptools
-import urllib
-import json
-import time
-import sys
-import os
-import argparse
-import logging
-import schema_salad.ref_resolver
-import requests
-from wes_service.util import visit
-from bravado.client import SwaggerClient
-from bravado.requests_client import RequestsClient
-
-def main(argv=sys.argv[1:]):
-    parser = argparse.ArgumentParser(description='Workflow Execution Service')
-    parser.add_argument("--host", type=str, default=os.environ.get("WES_API_HOST"))
-    parser.add_argument("--auth", type=str, default=os.environ.get("WES_API_AUTH"))
-    parser.add_argument("--proto", type=str, default=os.environ.get("WES_API_PROTO", "https"))
-    parser.add_argument("--quiet", action="store_true", default=False)
-    parser.add_argument("--outdir", type=str)
-    parser.add_argument("--page", type=str, default=None)
-    parser.add_argument("--page-size", type=int, default=None)
-
-    exgroup = parser.add_mutually_exclusive_group()
-    exgroup.add_argument("--run", action="store_true", default=False)
-    exgroup.add_argument("--get", type=str, default=None)
-    exgroup.add_argument("--log", type=str, default=None)
-    exgroup.add_argument("--list", action="store_true", default=False)
-    exgroup.add_argument("--info", action="store_true", default=False)
-    exgroup.add_argument("--version", action="store_true", default=False)
-
-    exgroup = parser.add_mutually_exclusive_group()
-    exgroup.add_argument("--wait", action="store_true", default=True, dest="wait")
-    exgroup.add_argument("--no-wait", action="store_false", default=True, dest="wait")
-
-    parser.add_argument("workflow_url", type=str, nargs="?", default=None)
-    parser.add_argument("job_order", type=str, nargs="?", default=None)
-    args = parser.parse_args(argv)
-
-    if args.version:
-        pkg = pkg_resources.require("wes_service")
-        print(u"%s %s" % (sys.argv[0], pkg[0].version))
-        exit(0)
-
-    http_client = RequestsClient()
-    split = urlparse.urlsplit("%s://%s/" % (args.proto, args.host))
-
-    http_client.set_api_key(
-        split.hostname, args.auth,
-        param_name='Authorization', param_in='header')
-    client = SwaggerClient.from_url(
-        "%s://%s/ga4gh/wes/v1/swagger.json" % (args.proto, args.host),
-        http_client=http_client, config={'use_models': False})
-
-    if args.list:
-        response = client.WorkflowExecutionService.ListWorkflows(page_token=args.page, page_size=args.page_size)
-        json.dump(response.result(), sys.stdout, indent=4)
-        return 0
-
-    if args.log:
-        response = client.WorkflowExecutionService.GetWorkflowLog(
-            workflow_id=args.log)
-        sys.stdout.write(response.result()["workflow_log"]["stderr"])
-        return 0
-
-    if args.get:
-        response = client.WorkflowExecutionService.GetWorkflowLog(
-            workflow_id=args.get)
-        json.dump(response.result(), sys.stdout, indent=4)
-        return 0
-
-    if args.info:
-        response = client.WorkflowExecutionService.GetServiceInfo()
-        json.dump(response.result(), sys.stdout, indent=4)
-        return 0
-
-    loader = schema_salad.ref_resolver.Loader({
-        "location": {"@type": "@id"},
-        "path": {"@type": "@id"}
-    })
-    input, _ = loader.resolve_ref(args.job_order)
-
-    basedir = os.path.dirname(args.job_order)
-
-    def fixpaths(d):
-        if isinstance(d, dict):
-            if "path" in d:
-                if ":" not in d["path"]:
-                    local_path = os.path.normpath(
-                        os.path.join(os.getcwd(), basedir, d["path"]))
-                    d["location"] = urllib.pathname2url(local_path)
-                else:
-                    d["location"] = d["path"]
-                del d["path"]
-            loc = d.get("location", "")
-            if d.get("class") == "Directory":
-                if loc.startswith("http:") or loc.startswith("https:"):
-                    logging.error("Directory inputs not supported with http references")
-                    exit(33)
-            # if not (loc.startswith("http:") or loc.startswith("https:")
-            #         or args.job_order.startswith("http:") or args.job_order.startswith("https:")):
-            #     logging.error("Upload local files not supported, must use http: or https: references.")
-            #     exit(33)
-
-    visit(input, fixpaths)
-
-    workflow_url = args.workflow_url
-    if not workflow_url.startswith("/") and ":" 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)),
-        ("workflow_type", "CWL"),
-        ("workflow_type_version", "v1.0")
-    ]
-
-    if workflow_url.startswith("file://"):
-        # with open(workflow_url[7:], "rb") as f:
-        #     body["workflow_descriptor"] = 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_descriptor', (fn[len(rootdir)+1:], open(fn, "rb"))))
-        parts.append(("workflow_url", os.path.basename(workflow_url[7:])))
-    else:
-        parts.append(("workflow_url", workflow_url))
-
-    postresult = http_client.session.post("%s://%s/ga4gh/wes/v1/workflows" % (args.proto, args.host),
-                                          files=parts,
-                                          headers={"Authorization": args.auth})
-
-    r = json.loads(postresult.text)
-
-    if postresult.status_code != 200:
-        logging.error("%s", r)
-        exit(1)
-
-    if args.wait:
-        logging.info("Workflow id is %s", r["workflow_id"])
-    else:
-        sys.stdout.write(r["workflow_id"]+"\n")
-        exit(0)
-
-    r = client.WorkflowExecutionService.GetWorkflowStatus(
-        workflow_id=r["workflow_id"]).result()
-    while r["state"] in ("QUEUED", "INITIALIZING", "RUNNING"):
-        time.sleep(8)
-        r = client.WorkflowExecutionService.GetWorkflowStatus(
-            workflow_id=r["workflow_id"]).result()
-
-    logging.info("State is %s", r["state"])
-
-    s = client.WorkflowExecutionService.GetWorkflowLog(
-        workflow_id=r["workflow_id"]).result()
-    logging.info("%s", s["workflow_log"]["stderr"])
-    logs = requests.get(s["workflow_log"]["stderr"], headers={"Authorization": args.auth}).text
-    logging.info("Workflow log:\n"+logs)
-
-    if "fields" in s["outputs"] and s["outputs"]["fields"] is None:
-        del s["outputs"]["fields"]
-    json.dump(s["outputs"], sys.stdout, indent=4)
-
-    if r["state"] == "COMPLETE":
-        return 0
-    else:
-        return 1
-
-
-if __name__ == "__main__":
-    sys.exit(main(sys.argv[1:]))