Merge pull request #47 from common-workflow-language/default-timeout

Add default timeout

Author: mr-c

cwltest/__init__.py

@@ -3,66 +3,46 @@
 from __future__ import absolute_import
 from __future__ import print_function
 
-from six.moves import range
-from six.moves import zip
-
 import argparse
 import json
+import logging
 import os
-import subprocess
-import sys
+import pipes
 import shutil
+import sys
 import tempfile
+import threading
+import time
+
 import junit_xml
 import ruamel.yaml as yaml
 import ruamel.yaml.scanner as yamlscanner
-import pipes
-import logging
 import schema_salad.ref_resolver
-import time
-import threading
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Dict, List, Text
+from six.moves import range
+from six.moves import zip
+from typing import Any, Dict, List
 
-from cwltest.utils import compare, CompareFail
+from cwltest.utils import compare, CompareFail, TestResult
 
 _logger = logging.getLogger("cwltest")
 _logger.addHandler(logging.StreamHandler())
 _logger.setLevel(logging.INFO)
 
 UNSUPPORTED_FEATURE = 33
-RUNTIME = sys.version_info.major
-
-
-class TestResult(object):
-
-    """Encapsulate relevant test result data."""
-
-    def __init__(self, return_code, standard_output, error_output, duration, classname, message=''):
-        # type: (int, Text, Text, float, Text, str) -> None
-        self.return_code = return_code
-        self.standard_output = standard_output
-        self.error_output = error_output
-        self.duration = duration
-        self.message = message
-        self.classname = classname
-
-    def create_test_case(self, test):
-        # type: (Dict[Text, Any]) -> junit_xml.TestCase
-        doc = test.get(u'doc', 'N/A').strip()
-        case = junit_xml.TestCase(
-            doc, elapsed_sec=self.duration, classname=self.classname,
-            stdout=self.standard_output, stderr=self.error_output,
-        )
-        if self.return_code > 0:
-            case.failure_message = self.message
-        return case
+DEFAULT_TIMEOUT = 900  # 15 minutes
 
+if sys.version_info < (3, 0):
+    import subprocess32 as subprocess
+else:
+    import subprocess
 
 templock = threading.Lock()
 
 
-def run_test(args, i, tests):  # type: (argparse.Namespace, int, List[Dict[str, str]]) -> TestResult
+def run_test(args, i, tests, timeout):
+    # type: (argparse.Namespace, int, List[Dict[str, str]], int) -> TestResult
+
     global templock
 
     out = {}  # type: Dict[str,Any]
@@ -105,7 +85,7 @@ def run_test(args, i, tests):  # type: (argparse.Namespace, int, List[Dict[str,
         start_time = time.time()
         stderr = subprocess.PIPE if not args.verbose else None
         process = subprocess.Popen(test_command, stdout=subprocess.PIPE, stderr=stderr)
-        outstr, outerr = [var.decode('utf-8') for var in process.communicate()]
+        outstr, outerr = [var.decode('utf-8') for var in process.communicate(timeout=timeout)]
         return_code = process.poll()
         duration = time.time() - start_time
         if return_code:
@@ -135,6 +115,10 @@ def run_test(args, i, tests):  # type: (argparse.Namespace, int, List[Dict[str,
     except KeyboardInterrupt:
         _logger.error(u"""Test interrupted: %s""", " ".join([pipes.quote(tc) for tc in test_command]))
         raise
+    except subprocess.TimeoutExpired:
+        _logger.error(u"""Test timed out: %s""", " ".join([pipes.quote(tc) for tc in test_command]))
+        _logger.error(t.get("doc"))
+        return TestResult(2, outstr, outerr, timeout, args.classname, "Test timed out")
 
     fail_message = ''
 
@@ -177,6 +161,9 @@ def main():  # type: () -> int
                                                         "(defaults to one).")
     parser.add_argument("--verbose", action="store_true", help="More verbose output during test run.")
     parser.add_argument("--classname", type=str, default="", help="Specify classname for the Test Suite.")
+    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="Time of execution in seconds after "
+                                                                             "which the test will be skipped."
+                                                                             "Defaults to 900 sec (15 minutes)")
 
     args = parser.parse_args()
     if '--' in args.args:
@@ -229,7 +216,7 @@ def main():  # type: () -> int
 
     total = 0
     with ThreadPoolExecutor(max_workers=args.j) as executor:
-        jobs = [executor.submit(run_test, args, i, tests)
+        jobs = [executor.submit(run_test, args, i, tests, args.timeout)
                 for i in ntest]
         try:
             for i, job in zip(ntest, jobs):

cwltest/utils.py

@@ -1,9 +1,36 @@
 import json
-from typing import Any, Dict, Set
+
+import junit_xml
+from typing import Any, Dict, Set, Text
 
 from six.moves import range
 
 
+class TestResult(object):
+
+    """Encapsulate relevant test result data."""
+
+    def __init__(self, return_code, standard_output, error_output, duration, classname, message=''):
+        # type: (int, Text, Text, float, Text, str) -> None
+        self.return_code = return_code
+        self.standard_output = standard_output
+        self.error_output = error_output
+        self.duration = duration
+        self.message = message
+        self.classname = classname
+
+    def create_test_case(self, test):
+        # type: (Dict[Text, Any]) -> junit_xml.TestCase
+        doc = test.get(u'doc', 'N/A').strip()
+        case = junit_xml.TestCase(
+            doc, elapsed_sec=self.duration, classname=self.classname,
+            stdout=self.standard_output, stderr=self.error_output,
+        )
+        if self.return_code > 0:
+            case.failure_message = self.message
+        return case
+
+
 class CompareFail(Exception):
 
     @classmethod

requirements.txt

@@ -1,4 +1,6 @@
 schema-salad >= 1.14
 typing>=3.6,<3.7; python_version < '3.5'
 futures >= 3.0.5; python_version == '2.7'
+subprocess32; python_version < '3'
 junit-xml >= 1.7
+

setup.py

@@ -23,12 +23,11 @@
 ]
 
 if sys.version_info.major == 2:
-    install_requires.append('futures >= 3.0.5')
+    install_requires.extend(['futures >= 3.0.5', 'subprocess32'])
 
 if sys.version_info[:2] < (3, 5):
     install_requires.append('typing >= 3.5.2')
 
-
 setup(name='cwltest',
       version='1.0',
       description='Common workflow language testing framework',

typeshed/2.7/subprocess32.pyi

@@ -0,0 +1,53 @@
+# Stubs for subprocess32 (Python 2)
+#
+# NOTE: This dynamically typed stub was automatically generated by stubgen.
+
+from typing import Any, Optional
+
+class CalledProcessError(Exception):
+    returncode: Any = ...
+    cmd: Any = ...
+    output: Any = ...
+    def __init__(self, returncode, cmd, output: Optional[Any] = ...) -> None: ...
+
+class TimeoutExpired(Exception):
+    cmd: Any = ...
+    timeout: Any = ...
+    output: Any = ...
+    def __init__(self, cmd, timeout, output: Optional[Any] = ...) -> None: ...
+
+class STARTUPINFO:
+    dwFlags: int = ...
+    hStdInput: Any = ...
+    hStdOutput: Any = ...
+    hStdError: Any = ...
+    wShowWindow: int = ...
+
+class pywintypes:
+    error: Any = ...
+
+PIPE: int
+STDOUT: int
+
+def call(*popenargs, **kwargs): ...
+def check_call(*popenargs, **kwargs): ...
+def check_output(*popenargs, **kwargs): ...
+
+class Popen:
+    args: Any = ...
+    stdin: Any = ...
+    stdout: Any = ...
+    stderr: Any = ...
+    pid: Any = ...
+    returncode: Any = ...
+    universal_newlines: Any = ...
+    def __init__(self, args, bufsize: int = ..., executable: Optional[Any] = ..., stdin: Optional[Any] = ..., stdout: Optional[Any] = ..., stderr: Optional[Any] = ..., preexec_fn: Optional[Any] = ..., close_fds: Any = ..., shell: bool = ..., cwd: Optional[Any] = ..., env: Optional[Any] = ..., universal_newlines: bool = ..., startupinfo: Optional[Any] = ..., creationflags: int = ..., restore_signals: bool = ..., start_new_session: bool = ..., pass_fds: Any = ...) -> None: ...
+    def __enter__(self): ...
+    def __exit__(self, type, value, traceback): ...
+    def __del__(self, _maxint: Any = ..., _active: Any = ...): ...
+    def communicate(self, input: Optional[Any] = ..., timeout: Optional[Any] = ...): ...
+    def poll(self) -> int: ...
+    def wait(self, timeout: Optional[Any] = ..., endtime: Optional[Any] = ...): ...
+    def send_signal(self, sig): ...
+    def terminate(self): ...
+    def kill(self): ...