Unite test files.

Author: bencvdb

test/test_client_util.py

@@ -4,8 +4,12 @@
 import os
 import logging
 import subprocess
+import sys
 
-from wes_client.util import expand_globs
+pkg_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))  # noqa
+sys.path.insert(0, pkg_root)  # noqa
+
+from wes_client.util import expand_globs, wf_info
 
 logging.basicConfig(level=logging.INFO)
 
@@ -35,5 +39,63 @@ def test_expand_globs(self):
         assert set(files) == glob_files, '\n' + str(set(files)) + '\n' + str(glob_files)
 
 
+class WorkflowInfoTest(unittest.TestCase):
+
+    local = {'cwl': 'file://' + os.path.join(os.getcwd() + '/testdata/md5sum.cwl'),
+             'wdl': 'file://' + os.path.join(os.getcwd() + '/testdata/md5sum.wdl'),
+             'py': 'file://' + os.path.join(os.getcwd() + '/test/test_integration.py'),
+             'unsupported': 'fake.txt'}
+
+    remote = {'cwl': 'https://raw.githubusercontent.com/common-workflow-language/workflow-service/master/testdata/md5sum.cwl',
+              'wdl': 'https://raw.githubusercontent.com/common-workflow-language/workflow-service/master/testdata/md5sum.wdl',
+              'py': 'https://raw.githubusercontent.com/common-workflow-language/workflow-service/master/test/test_integration.py',
+              'unsupported': 'gs://topmed_workflow_testing/topmed_aligner/small_test_files_sbg/example_human_known_snp.py',  # TODO: find real external file of .py, .cwl, .wdl
+              'unreachable': 'https://fake.py'}
+
+    expected = {'cwl': ('v1.0', 'CWL'),
+                'wdl': ('draft-2', 'WDL'),
+                'py': ('2.7', 'PY'),
+                'pyWithPrefix': ('2.7', 'PY')}
+
+    def testSupportedFormatChecking(self):
+        """
+        Check that non-wdl, -python, -cwl files are rejected.
+
+        This test is run only on local files to avoid downloading and removing a new file.
+        """
+
+        for format, location in self.local.items():
+            if format != 'unsupported':
+                # Tests the behavior after receiving supported file types with and without the 'file://' prefix
+                self.assertEquals(wf_info(location), self.expected[format])
+                self.assertEquals(wf_info(location[7:]), self.expected[format])
+
+            else:
+                # Tests behavior after recieveing a non supported file type.
+                with self.assertRaises(TypeError):
+                    wf_info(location)
+
+    def testFileLocationChecking(self):
+        """
+        Check that the function rejects unsupported file locations.
+
+        This test needs to be run on remote files to test the location checking functionality of wf_info().
+        """
+
+        for format, location in self.remote.items():
+            if format == 'unsupported':
+                # Tests behavior after receiving a file hosted at an unsupported location.
+                with self.assertRaises(NotImplementedError):
+                    wf_info(location)
+
+            elif format == 'unreachable':
+                # Tests behavior after receiving a non-existent file.
+                with self.assertRaises(IOError):
+                    wf_info(location)
+
+            else:
+                self.assertEquals(wf_info(location), self.expected[format])
+                self.assertFalse(os.path.isfile(os.path.join(os.getcwd(), 'fetchedFromRemote.' + format)))
+
 if __name__ == '__main__':
     unittest.main()  # run all tests

test/test_wesclient_utils.py

@@ -12,7 +12,7 @@
 from urllib import urlopen
 
 
-def _two_seven_compatible(filePath):
+def two_seven_compatible(filePath):
     """Determines if a python file is 2.7 compatible by seeing if it compiles in a subprocess"""
     try:
         check_call(['python2', '-m', 'py_compile', filePath], stderr=DEVNULL)
@@ -21,9 +21,9 @@ def _two_seven_compatible(filePath):
     return True
 
 
-def _get_version(extension, workflow_file):
+def get_version(extension, workflow_file):
     '''Determines the version of a .py, .wdl, or .cwl file.'''
-    if extension == 'py' and _two_seven_compatible(workflow_file):
+    if extension == 'py' and two_seven_compatible(workflow_file):
         return '2.7'
     elif extension == 'cwl':
         return yaml.load(open(workflow_file))['cwlVersion']
@@ -50,7 +50,7 @@ def wf_info(workflow_path):
 
     if file_type in supported_formats:
         if workflow_path.startswith('file://'):
-            version = _get_version(file_type, workflow_path[7:])
+            version = get_version(file_type, workflow_path[7:])
         elif workflow_path.startswith('https://') or workflow_path.startswith('http://'):
             # If file not local go fetch it.
             html = urlopen(workflow_path).read()