manyi/venv/Lib/site-packages/pytest_timeout.py

"""Timeout for tests to stop hanging testruns

This plugin will dump the stack and terminate the test.  This can be
useful when running tests on a continuous integration server.

If the platform supports SIGALRM this is used to raise an exception in
the test, otherwise os._exit(1) is used.
"""

import os
import signal
import sys
import threading
import traceback
from collections import namedtuple
from distutils.version import StrictVersion

import py
import pytest


HAVE_SIGALRM = hasattr(signal, 'SIGALRM')
if HAVE_SIGALRM:
    DEFAULT_METHOD = 'signal'
else:
    DEFAULT_METHOD = 'thread'
TIMEOUT_DESC = """
Timeout in seconds before dumping the stacks.  Default is 0 which
means no timeout.
""".strip()
METHOD_DESC = """
Timeout mechanism to use.  'signal' uses SIGALRM if available,
'thread' uses a timer thread.  The default is to use 'signal' and fall
back to 'thread'.
""".strip()
FUNC_ONLY_DESC = """
When set to True, defers the timeout evaluation to only the test
function body, ignoring the time it takes when evaluating any fixtures
used in the test.
""".strip()

Settings = namedtuple('Settings', ['timeout', 'method', 'func_only'])


@pytest.hookimpl
def pytest_addoption(parser):
    """Add options to control the timeout plugin"""
    group = parser.getgroup(
        'timeout', 'Interrupt test run and dump stacks of all threads after '
        'a test times out')
    group.addoption('--timeout',
                    type=float,
                    help=TIMEOUT_DESC)
    group.addoption('--timeout_method',
                    action='store',
                    choices=['signal', 'thread'],
                    help='Deprecated, use --timeout-method')
    group.addoption('--timeout-method',
                    dest='timeout_method',
                    action='store',
                    choices=['signal', 'thread'],
                    help=METHOD_DESC)
    parser.addini('timeout', TIMEOUT_DESC)
    parser.addini('timeout_method', METHOD_DESC)
    parser.addini('timeout_func_only', FUNC_ONLY_DESC, type='bool')


@pytest.hookimpl
def pytest_configure(config):
    # Register the marker so it shows up in --markers output.
    config.addinivalue_line(
        'markers',
        'timeout(timeout, method=None, func_only=False): Set a timeout, timeout '
        'method and func_only evaluation on just one test item.  The first '
        'argument, *timeout*, is the timeout in seconds while the keyword, '
        '*method*, takes the same values as the --timeout_method option. The '
        '*func_only* keyword, when set to True, defers the timeout evaluation '
        'to only the test function body, ignoring the time it takes when '
        'evaluating any fixrures used in the test.')

    settings = get_env_settings(config)
    config._env_timeout = settings.timeout
    config._env_timeout_method = settings.method
    config._env_timeout_func_only = settings.func_only



@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_protocol(item):
    func_only = get_func_only_setting(item)
    if func_only is False:
        timeout_setup(item)
    outcome = yield
    if func_only is False:
        timeout_teardown(item)


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
    func_only = get_func_only_setting(item)
    if func_only is True:
        timeout_setup(item)
    outcome = yield
    if func_only is True:
        timeout_teardown(item)


@pytest.hookimpl(tryfirst=True)
def pytest_report_header(config):
    if config._env_timeout:
        return ["timeout: %ss\ntimeout method: %s\ntimeout func_only: %s" %
                (config._env_timeout,
                 config._env_timeout_method,
                 config._env_timeout_func_only)]


@pytest.hookimpl(tryfirst=True)
def pytest_exception_interact(node):
    timeout_teardown(node)


@pytest.hookimpl
def pytest_enter_pdb():
    # Since pdb.set_trace happens outside of any pytest control, we don't have
    # any pytest ``item`` here, so we cannot use timeout_teardown. Thus, we
    # need another way to signify that the timeout should not be performed.
    global SUPPRESS_TIMEOUT
    SUPPRESS_TIMEOUT = True
SUPPRESS_TIMEOUT = False


def timeout_setup(item):
    """Setup up a timeout trigger and handler"""
    params = get_params(item)
    if params.timeout is None or params.timeout <= 0:
        return
    if params.method == 'signal':
        def handler(signum, frame):
            __tracebackhide__ = True
            timeout_sigalrm(item, params.timeout)
        def cancel():
            signal.setitimer(signal.ITIMER_REAL, 0)
            signal.signal(signal.SIGALRM, signal.SIG_DFL)
        item.cancel_timeout = cancel
        signal.signal(signal.SIGALRM, handler)
        signal.setitimer(signal.ITIMER_REAL, params.timeout)
    elif params.method == 'thread':
        timer = threading.Timer(params.timeout, timeout_timer,
                                (item, params.timeout))
        def cancel():
            timer.cancel()
            timer.join()
        item.cancel_timeout = cancel
        timer.start()


def timeout_teardown(item):
    """Cancel the timeout trigger if it was set"""
    # When skipping is raised from a pytest_runtest_setup function
    # (as is the case when using the pytest.mark.skipif marker) we
    # may be called without our setup counterpart having been
    # called.
    cancel = getattr(item, 'cancel_timeout', None)
    if cancel:
        cancel()


def get_env_settings(config):
    timeout = config.getvalue('timeout')
    if timeout is None:
        timeout = _validate_timeout(
            os.environ.get('PYTEST_TIMEOUT'),
            'PYTEST_TIMEOUT environment variable')
    if timeout is None:
        ini = config.getini('timeout')
        if ini:
            timeout = _validate_timeout(ini, 'config file')

    method = config.getvalue('timeout_method')
    if method is None:
        ini = config.getini('timeout_method')
        if ini:
            method = _validate_method(ini, 'config file')
    if method is None:
        method = DEFAULT_METHOD

    func_only = config.getini('timeout_func_only')
    if func_only == []:
        # No value set
        func_only = None
    if func_only is not None:
        func_only = _validate_func_only(func_only, 'config file')
    return Settings(timeout, method, func_only or False)


def get_func_only_setting(item):
    """Return the func_only setting for an item"""
    func_only = None
    marker = item.get_closest_marker('timeout')
    if marker:
        settings = get_params(item, marker=marker)
        func_only = _validate_func_only(settings.func_only, 'marker')
    if func_only is None:
        func_only = item.config._env_timeout_func_only
    if func_only is None:
        func_only = False
    return func_only


def get_params(item, marker=None):
    """Return (timeout, method) for an item"""
    timeout = method = func_only = None
    if not marker:
        marker = item.get_closest_marker('timeout')
    if marker is not None:
        settings = _parse_marker(item.get_closest_marker(name='timeout'))
        timeout = _validate_timeout(settings.timeout, 'marker')
        method = _validate_method(settings.method, 'marker')
        func_only = _validate_func_only(settings.func_only, 'marker')
    if timeout is None:
        timeout = item.config._env_timeout
    if method is None:
        method = item.config._env_timeout_method
    if func_only is None:
        func_only = item.config._env_timeout_func_only
    return Settings(timeout, method, func_only)


def _parse_marker(marker):
    """Return (timeout, method) tuple from marker

    Either could be None.  The values are not interpreted, so
    could still be bogus and even the wrong type.
    """
    if not marker.args and not marker.kwargs:
        raise TypeError('Timeout marker must have at least one argument')
    timeout = method = func_only = NOTSET = object()
    for kw, val in marker.kwargs.items():
        if kw == 'timeout':
            timeout = val
        elif kw == 'method':
            method = val
        elif kw == 'func_only':
            func_only = val
        else:
            raise TypeError(
                'Invalid keyword argument for timeout marker: %s' % kw)
    if len(marker.args) >= 1 and timeout is not NOTSET:
        raise TypeError(
            'Multiple values for timeout argument of timeout marker')
    elif len(marker.args) >= 1:
        timeout = marker.args[0]
    if len(marker.args) >= 2 and method is not NOTSET:
        raise TypeError(
            'Multiple values for method argument of timeout marker')
    elif len(marker.args) >= 2:
        method = marker.args[1]
    if len(marker.args) > 2:
        raise TypeError('Too many arguments for timeout marker')
    if timeout is NOTSET:
        timeout = None
    if method is NOTSET:
        method = None
    if func_only is NOTSET:
        func_only = None
    return Settings(timeout, method, func_only)


def _validate_timeout(timeout, where):
    """Helper for get_params()"""
    if timeout is None:
        return None
    try:
        return float(timeout)
    except ValueError:
        raise ValueError('Invalid timeout %s from %s' % (timeout, where))


def _validate_method(method, where):
    """Helper for get_params()"""
    if method is None:
        return None
    if method not in ['signal', 'thread']:
        raise ValueError('Invalid method %s from %s' % (method, where))
    return method


def _validate_func_only(func_only, where):
    """Helper for get_params()"""
    if func_only is None:
        return False
    if not isinstance(func_only, bool):
        raise ValueError('Invalid func_only value %s from %s' % (func_only, where))
    return func_only


def timeout_sigalrm(item, timeout):
    """Dump stack of threads and raise an exception

    This will output the stacks of any threads other then the
    current to stderr and then raise an AssertionError, thus
    terminating the test.
    """
    if SUPPRESS_TIMEOUT:
        return
    __tracebackhide__ = True
    nthreads = len(threading.enumerate())
    if nthreads > 1:
        write_title('Timeout', sep='+')
    dump_stacks()
    if nthreads > 1:
        write_title('Timeout', sep='+')
    pytest.fail('Timeout >%ss' % timeout)


def timeout_timer(item, timeout):
    """Dump stack of threads and call os._exit()

    This disables the capturemanager and dumps stdout and stderr.
    Then the stacks are dumped and os._exit(1) is called.
    """
    if SUPPRESS_TIMEOUT:
        return
    try:
        capman = item.config.pluginmanager.getplugin('capturemanager')
        if capman:
            pytest_version = StrictVersion(pytest.__version__)
            if pytest_version >= StrictVersion('3.7.3'):
                capman.suspend_global_capture(item)
                stdout, stderr = capman.read_global_capture()
            else:
                stdout, stderr = capman.suspend_global_capture(item)
        else:
            stdout, stderr = None, None
        write_title('Timeout', sep='+')
        caplog = item.config.pluginmanager.getplugin('_capturelog')
        if caplog and hasattr(item, 'capturelog_handler'):
            log = item.capturelog_handler.stream.getvalue()
            if log:
                write_title('Captured log')
                write(log)
        if stdout:
            write_title('Captured stdout')
            write(stdout)
        if stderr:
            write_title('Captured stderr')
            write(stderr)
        dump_stacks()
        write_title('Timeout', sep='+')
    except Exception:
        traceback.print_exc()
    finally:
        sys.stdout.flush()
        sys.stderr.flush()
        os._exit(1)


def dump_stacks():
    """Dump the stacks of all threads except the current thread"""
    current_ident = threading.current_thread().ident
    for thread_ident, frame in sys._current_frames().items():
        if thread_ident == current_ident:
            continue
        for t in threading.enumerate():
            if t.ident == thread_ident:
                thread_name = t.name
                break
        else:
            thread_name = '<unknown>'
        write_title('Stack of %s (%s)' % (thread_name, thread_ident))
        write(''.join(traceback.format_stack(frame)))


def write_title(title, stream=None, sep='~'):
    """Write a section title

    If *stream* is None sys.stderr will be used, *sep* is used to
    draw the line.
    """
    if stream is None:
        stream = sys.stderr
    width = py.io.get_terminal_width()
    fill = int((width - len(title) - 2) / 2)
    line = ' '.join([sep * fill, title, sep * fill])
    if len(line) < width:
        line += sep * (width - len(line))
    stream.write('\n' + line + '\n')


def write(text, stream=None):
    """Write text to stream

    Pretty stupid really, only here for symetry with .write_title().
    """
    if stream is None:
        stream = sys.stderr
    stream.write(text)