# Copyright (c) 2008-2009 AG Projects
# Author: Denis Bilenko
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

# package is named greentest, not test, so it won't be confused with test in stdlib
# pylint:disable=broad-except,unused-argument,no-member,too-many-branches,unused-variable
# pylint:disable=attribute-defined-outside-init,abstract-method
import sys
import types
import unittest
from unittest import TestCase as BaseTestCase
from unittest.util import safe_repr
import time
import os
from os.path import basename, splitext
import gevent
import gevent.core
from patched_tests_setup import get_switch_expected
from gevent.hub import _get_hub
from functools import wraps
import contextlib
import gc
import _six as six


PYPY = hasattr(sys, 'pypy_version_info')
VERBOSE = sys.argv.count('-v') > 1
WIN = sys.platform.startswith("win")
LINUX = sys.platform.startswith('linux')

# XXX: Formalize this better
LIBUV = os.getenv('GEVENT_CORE_CFFI_ONLY') == 'libuv' or (PYPY and WIN) or hasattr(gevent.core, 'libuv')
CFFI_BACKEND = bool(os.getenv('GEVENT_CORE_CFFI_ONLY')) or PYPY

if '--debug-greentest' in sys.argv:
    sys.argv.remove('--debug-greentest')
    DEBUG = True
else:
    DEBUG = False

RUN_LEAKCHECKS = os.getenv('GEVENTTEST_LEAKCHECK')
OPTIONAL_MODULES = ['resolver_ares']

# Generally, ignore the portions that are only implemented
# on particular platforms; they generally contain partial
# implementations completed in different modules.
PLATFORM_SPECIFIC_SUFFIXES = ['2', '279', '3']
if WIN:
    PLATFORM_SPECIFIC_SUFFIXES.append('posix')

PY2 = None
PY3 = None
PY34 = None
PY36 = None
PY37 = None

NON_APPLICABLE_SUFFIXES = []
if sys.version_info[0] == 3:
    # Python 3
    NON_APPLICABLE_SUFFIXES.extend(('2', '279'))
    PY2 = False
    PY3 = True
    if sys.version_info[1] >= 4:
        PY34 = True
    if sys.version_info[1] >= 6:
        PY36 = True
    if sys.version_info[1] >= 7:
        PY37 = True

elif sys.version_info[0] == 2:
    # Any python 2
    PY3 = False
    PY2 = True
    NON_APPLICABLE_SUFFIXES.append('3')
    if (sys.version_info[1] < 7
            or (sys.version_info[1] == 7 and sys.version_info[2] < 9)):
        # Python 2, < 2.7.9
        NON_APPLICABLE_SUFFIXES.append('279')

PYPY3 = PYPY and PY3

if WIN:
    NON_APPLICABLE_SUFFIXES.append("posix")
    # This is intimately tied to FileObjectPosix
    NON_APPLICABLE_SUFFIXES.append("fileobject2")
    SHARED_OBJECT_EXTENSION = ".pyd"
else:
    SHARED_OBJECT_EXTENSION = ".so"


RUNNING_ON_TRAVIS = os.environ.get('TRAVIS')
RUNNING_ON_APPVEYOR = os.environ.get('APPVEYOR')
RUNNING_ON_CI = RUNNING_ON_TRAVIS or RUNNING_ON_APPVEYOR

def _do_not_skip(reason):
    def dec(f):
        return f
    return dec

if WIN:
    skipOnWindows = unittest.skip
else:
    skipOnWindows = _do_not_skip

if RUNNING_ON_APPVEYOR:
    # See comments scattered around about timeouts and the timer
    # resolution available on appveyor (lots of jitter). this
    # seems worse with the 62-bit builds.
    # Note that we skip/adjust these tests only on AppVeyor, not
    # win32---we don't think there's gevent related problems but
    # environment related problems. These can be tested and debugged
    # separately on windows in a more stable environment.
    skipOnAppVeyor = unittest.skip

    # We can't exec corecext on appveyor if we haven't run setup.py in
    # 'develop' mode (i.e., we install)
    NON_APPLICABLE_SUFFIXES.append('corecext')
else:
    skipOnAppVeyor = _do_not_skip

if PYPY3 and RUNNING_ON_CI:
    # Same as above, for PyPy3.3-5.5-alpha and 3.5-5.7.1-beta and 3.5-5.8
    skipOnPyPy3OnCI = unittest.skip
else:
    skipOnPyPy3OnCI = _do_not_skip

if PYPY:
    skipOnPyPy = unittest.skip
else:
    skipOnPyPy = _do_not_skip

if PYPY3:
    skipOnPyPy3 = unittest.skip
else:
    skipOnPyPy3 = _do_not_skip

skipIf = unittest.skipIf

EXPECT_POOR_TIMER_RESOLUTION = PYPY3 or RUNNING_ON_APPVEYOR

skipOnLibuv = _do_not_skip
skipOnLibuvOnCI = _do_not_skip
skipOnLibuvOnCIOnPyPy = _do_not_skip

if LIBUV:
    skipOnLibuv = unittest.skip

    if RUNNING_ON_CI:
        skipOnLibuvOnCI = unittest.skip
        if PYPY:
            skipOnLibuvOnCIOnPyPy = unittest.skip

class ExpectedException(Exception):
    """An exception whose traceback should be ignored by the hub"""


# The next exceptions allow us to raise them in a highly
# greppable way so that we can debug them later.

class FlakyTest(unittest.SkipTest):
    """
    A unittest exception that causes the test to be skipped when raised.

    Use this carefully, it is a code smell and indicates an undebugged problem.
    """

class FlakyTestRaceCondition(FlakyTest):
    """
    Use this when the flaky test is definitely caused by a race condition.
    """

class FlakyTestTimeout(FlakyTest):
    """
    Use this when the flaky test is definitely caused by an
    unexpected timeout.
    """


if RUNNING_ON_CI:
    def reraiseFlakyTestRaceCondition():
        six.reraise(FlakyTestRaceCondition,
                    FlakyTestRaceCondition('\n'.join(dump_stacks())),
                    sys.exc_info()[2])

    def reraiseFlakyTestTimeout():
        six.reraise(FlakyTestTimeout,
                    FlakyTestTimeout(),
                    sys.exc_info()[2])

else:
    def reraiseFlakyTestRaceCondition():
        six.reraise(*sys.exc_info())

    reraiseFlakyTestTimeout = reraiseFlakyTestRaceCondition

def wrap_switch_count_check(method):
    @wraps(method)
    def wrapper(self, *args, **kwargs):
        initial_switch_count = getattr(_get_hub(), 'switch_count', None)
        self.switch_expected = getattr(self, 'switch_expected', True)
        if initial_switch_count is not None:
            fullname = getattr(self, 'fullname', None)
            if self.switch_expected == 'default' and fullname:
                self.switch_expected = get_switch_expected(fullname)
        result = method(self, *args, **kwargs)
        if initial_switch_count is not None and self.switch_expected is not None:
            switch_count = _get_hub().switch_count - initial_switch_count
            if self.switch_expected is True:
                assert switch_count >= 0
                if not switch_count:
                    raise AssertionError('%s did not switch' % fullname)
            elif self.switch_expected is False:
                if switch_count:
                    raise AssertionError('%s switched but not expected to' % fullname)
            else:
                raise AssertionError('Invalid value for switch_expected: %r' % (self.switch_expected, ))
        return result
    return wrapper


def wrap_timeout(timeout, method):
    if timeout is None:
        return method

    @wraps(method)
    def wrapper(self, *args, **kwargs):
        with gevent.Timeout(timeout, 'test timed out', ref=False):
            return method(self, *args, **kwargs)

    return wrapper

def ignores_leakcheck(func):
    func.ignore_leakcheck = True
    return func

def wrap_refcount(method):
    if not RUN_LEAKCHECKS:
        return method

    if getattr(method, 'ignore_leakcheck', False):
        return method

    # Some builtin things that we ignore
    IGNORED_TYPES = (tuple, dict, types.FrameType, types.TracebackType)

    def type_hist():
        import collections
        d = collections.defaultdict(int)
        for x in gc.get_objects():
            k = type(x)
            if k in IGNORED_TYPES:
                continue
            if k == gevent.core.callback and x.callback is None and x.args is None:
                # these represent callbacks that have been stopped, but
                # the event loop hasn't cycled around to run them. The only
                # known cause of this is killing greenlets before they get a chance
                # to run for the first time.
                continue
            d[k] += 1
        return d

    def report_diff(a, b):
        diff_lines = []
        for k, v in sorted(a.items(), key=lambda i: i[0].__name__):
            if b[k] != v:
                diff_lines.append("%s: %s != %s" % (k, v, b[k]))

        if not diff_lines:
            return None
        diff = '\n'.join(diff_lines)
        return diff

    @wraps(method)
    def wrapper(self, *args, **kwargs):
        gc.collect()
        gc.collect()
        gc.collect()
        deltas = []
        d = None
        gc.disable()
        try:
            while True:

                # Grab current snapshot
                hist_before = type_hist()
                d = sum(hist_before.values())

                self.setUp()
                try:
                    method(self, *args, **kwargs)
                finally:
                    self.tearDown()

                # Grab post snapshot
                if 'urlparse' in sys.modules:
                    sys.modules['urlparse'].clear_cache()
                if 'urllib.parse' in sys.modules:
                    sys.modules['urllib.parse'].clear_cache()
                hist_after = type_hist()
                d = sum(hist_after.values()) - d
                deltas.append(d)

                # Reset and check for cycles
                gc.collect()
                if gc.garbage:
                    raise AssertionError("Generated uncollectable garbage %r" % (gc.garbage,))

                # the following configurations are classified as "no leak"
                # [0, 0]
                # [x, 0, 0]
                # [... a, b, c, d]  where a+b+c+d = 0
                #
                # the following configurations are classified as "leak"
                # [... z, z, z]  where z > 0
                if deltas[-2:] == [0, 0] and len(deltas) in (2, 3):
                    break
                elif deltas[-3:] == [0, 0, 0]:
                    break
                elif len(deltas) >= 4 and sum(deltas[-4:]) == 0:
                    break
                elif len(deltas) >= 3 and deltas[-1] > 0 and deltas[-1] == deltas[-2] and deltas[-2] == deltas[-3]:
                    diff = report_diff(hist_before, hist_after)
                    raise AssertionError('refcount increased by %r\n%s' % (deltas, diff))
                # OK, we don't know for sure yet. Let's search for more
                if sum(deltas[-3:]) <= 0 or sum(deltas[-4:]) <= 0 or deltas[-4:].count(0) >= 2:
                    # this is suspicious, so give a few more runs
                    limit = 11
                else:
                    limit = 7
                if len(deltas) >= limit:
                    raise AssertionError('refcount increased by %r\n%s' % (deltas, report_diff(hist_before, hist_after)))
        finally:
            gc.enable()
        self.skipTearDown = True

    return wrapper


def wrap_error_fatal(method):
    @wraps(method)
    def wrapper(self, *args, **kwargs):
        # XXX should also be able to do gevent.SYSTEM_ERROR = object
        # which is a global default to all hubs
        SYSTEM_ERROR = gevent.get_hub().SYSTEM_ERROR
        gevent.get_hub().SYSTEM_ERROR = object
        try:
            return method(self, *args, **kwargs)
        finally:
            gevent.get_hub().SYSTEM_ERROR = SYSTEM_ERROR
    return wrapper


def wrap_restore_handle_error(method):
    @wraps(method)
    def wrapper(self, *args, **kwargs):
        old = gevent.get_hub().handle_error
        try:
            return method(self, *args, **kwargs)
        finally:
            gevent.get_hub().handle_error = old
        if self.peek_error()[0] is not None:
            gevent.getcurrent().throw(*self.peek_error()[1:])
    return wrapper


def _get_class_attr(classDict, bases, attr, default=AttributeError):
    NONE = object()
    value = classDict.get(attr, NONE)
    if value is not NONE:
        return value
    for base in bases:
        value = getattr(bases[0], attr, NONE)
        if value is not NONE:
            return value
    if default is AttributeError:
        raise AttributeError('Attribute %r not found\n%s\n%s\n' % (attr, classDict, bases))
    return default


class TestCaseMetaClass(type):
    # wrap each test method with
    # a) timeout check
    # b) fatal error check
    # c) restore the hub's error handler (see expect_one_error)
    # d) totalrefcount check
    def __new__(cls, classname, bases, classDict):
        # pylint and pep8 fight over what this should be called (mcs or cls).
        # pylint gets it right, but we cant scope disable pep8, so we go with
        # its convention.
        # pylint: disable=bad-mcs-classmethod-argument
        timeout = classDict.get('__timeout__', 'NONE')
        if timeout == 'NONE':
            timeout = getattr(bases[0], '__timeout__', None)
            if RUN_LEAKCHECKS and timeout is not None:
                timeout *= 6
        check_totalrefcount = _get_class_attr(classDict, bases, 'check_totalrefcount', True)
        error_fatal = _get_class_attr(classDict, bases, 'error_fatal', True)
        # Python 3: must copy, we mutate the classDict. Interestingly enough,
        # it doesn't actually error out, but under 3.6 we wind up wrapping
        # and re-wrapping the same items over and over and over.
        for key, value in list(classDict.items()):
            if key.startswith('test') and callable(value):
                classDict.pop(key)
                #value = wrap_switch_count_check(value)
                value = wrap_timeout(timeout, value)
                my_error_fatal = getattr(value, 'error_fatal', None)
                if my_error_fatal is None:
                    my_error_fatal = error_fatal
                if my_error_fatal:
                    value = wrap_error_fatal(value)
                value = wrap_restore_handle_error(value)
                if check_totalrefcount:
                    value = wrap_refcount(value)
                classDict[key] = value
        return type.__new__(cls, classname, bases, classDict)

# Travis is slow and overloaded; Appveyor used to be faster, but
# as of Dec 2015 it's almost always slower and/or has much worse timer
# resolution
CI_TIMEOUT = 10
if (PY3 and PYPY) or (PYPY and WIN and LIBUV):
    # pypy3 is very slow right now,
    # as is PyPy2 on windows (which only has libuv)
    CI_TIMEOUT = 15
if PYPY and LIBUV:
    # slow and flaky timeouts
    LOCAL_TIMEOUT = CI_TIMEOUT
else:
    LOCAL_TIMEOUT = 1

DEFAULT_LOCAL_HOST_ADDR = 'localhost'
DEFAULT_LOCAL_HOST_ADDR6 = DEFAULT_LOCAL_HOST_ADDR
DEFAULT_BIND_ADDR = ''

if RUNNING_ON_TRAVIS:
    # As of November 2017 (probably Sept or Oct), after a
    # Travis upgrade, using "localhost" no longer works,
    # producing 'OSError: [Errno 99] Cannot assign
    # requested address'. This is apparently something to do with
    # docker containers. Sigh.
    DEFAULT_LOCAL_HOST_ADDR = '127.0.0.1'
    DEFAULT_LOCAL_HOST_ADDR6 = '::1'
    # Likewise, binding to '' appears to work, but it cannot be
    # connected to with the same error.
    DEFAULT_BIND_ADDR = '127.0.0.1'

class TestCase(TestCaseMetaClass("NewBase", (BaseTestCase,), {})):
    __timeout__ = LOCAL_TIMEOUT if not RUNNING_ON_CI else CI_TIMEOUT
    switch_expected = 'default'
    error_fatal = True
    close_on_teardown = ()

    def run(self, *args, **kwargs):
        # pylint:disable=arguments-differ
        if self.switch_expected == 'default':
            self.switch_expected = get_switch_expected(self.fullname)
        return BaseTestCase.run(self, *args, **kwargs)

    def setUp(self):
        super(TestCase, self).setUp()
        self.close_on_teardown = []

    def tearDown(self):
        if getattr(self, 'skipTearDown', False):
            return
        if hasattr(self, 'cleanup'):
            self.cleanup()
        self._error = self._none
        self._tearDownCloseOnTearDown()
        self.close_on_teardown = []
        super(TestCase, self).tearDown()

    def _tearDownCloseOnTearDown(self):
        # XXX: Should probably reverse this
        for x in self.close_on_teardown:
            close = getattr(x, 'close', x)
            try:
                close()
            except Exception:
                pass

    @classmethod
    def setUpClass(cls):
        import warnings
        cls._warning_cm = warnings.catch_warnings()
        cls._warning_cm.__enter__()
        if not sys.warnoptions:
            warnings.simplefilter('default')
        super(TestCase, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        cls._warning_cm.__exit__(None, None, None)
        super(TestCase, cls).tearDownClass()

    def _close_on_teardown(self, resource):
        """
        *resource* either has a ``close`` method, or is a
        callable.
        """
        self.close_on_teardown.append(resource)
        return resource

    @property
    def testname(self):
        return getattr(self, '_testMethodName', '') or getattr(self, '_TestCase__testMethodName')

    @property
    def testcasename(self):
        return self.__class__.__name__ + '.' + self.testname

    @property
    def modulename(self):
        return os.path.basename(sys.modules[self.__class__.__module__].__file__).rsplit('.', 1)[0]

    @property
    def fullname(self):
        return splitext(basename(self.modulename))[0] + '.' + self.testcasename

    _none = (None, None, None)
    # (context, kind, value)
    _error = _none

    def expect_one_error(self):
        assert self._error == self._none, self._error
        self._old_handle_error = gevent.get_hub().handle_error
        gevent.get_hub().handle_error = self._store_error

    def _store_error(self, where, t, value, tb):
        del tb
        if self._error != self._none:
            gevent.get_hub().parent.throw(t, value)
        else:
            self._error = (where, t, value)

    def peek_error(self):
        return self._error

    def get_error(self):
        try:
            return self._error
        finally:
            self._error = self._none

    def assert_error(self, kind=None, value=None, error=None, where_type=None):
        if error is None:
            error = self.get_error()
        econtext, ekind, evalue = error
        if kind is not None:
            self.assertIsInstance(kind, type)
            try:
                assert issubclass(ekind, kind), error
            except TypeError as e:
                # Seen on PyPy on Windows
                print("TYPE ERROR", e, ekind, kind, type(kind))
                raise
        if value is not None:
            if isinstance(value, str):
                self.assertEqual(str(evalue), value)
            else:
                self.assertIs(evalue, value)
        if where_type is not None:
            self.assertIsInstance(econtext, where_type)
        return error

    if RUNNING_ON_APPVEYOR:
        # appveyor timeouts are unreliable; seems to be very slow wakeups
        def assertTimeoutAlmostEqual(self, *args, **kwargs):
            return

        def assertTimeWithinRange(self, delay, min_time, max_time):
            return
    else:
        def assertTimeoutAlmostEqual(self, *args, **kwargs):
            self.assertAlmostEqual(*args, **kwargs)

        def assertTimeWithinRange(self, delay, min_time, max_time):
            self.assertLessEqual(delay, max_time)
            self.assertGreaterEqual(delay, min_time)

    if not hasattr(BaseTestCase, 'assertIsNot'):
        # Methods added in 3.1, backport for 2.7
        def assertIs(self, expr1, expr2, msg=None):
            """Just like self.assertTrue(a is b), but with a nicer default message."""
            if expr1 is not expr2:
                standardMsg = '%s is not %s' % (safe_repr(expr1),
                                                safe_repr(expr2))
                self.fail(self._formatMessage(msg, standardMsg))

        def assertIsNot(self, expr1, expr2, msg=None):
            """Just like self.assertTrue(a is not b), but with a nicer default message."""
            if expr1 is expr2:
                standardMsg = 'unexpectedly identical: %s' % (safe_repr(expr1),)
                self.fail(self._formatMessage(msg, standardMsg))

    def assertMonkeyPatchedFuncSignatures(self, mod_name, func_names=(), exclude=()):
        # We use inspect.getargspec because it's the only thing available
        # in Python 2.7, but it is deprecated
        # pylint:disable=deprecated-method
        import inspect
        import warnings
        from gevent.monkey import get_original
        # XXX: Very similar to gevent.monkey.patch_module. Should refactor?
        gevent_module = getattr(__import__('gevent.' + mod_name), mod_name)
        module_name = getattr(gevent_module, '__target__', mod_name)

        funcs_given = True
        if not func_names:
            funcs_given = False
            func_names = getattr(gevent_module, '__implements__')

        for func_name in func_names:
            if func_name in exclude:
                continue
            gevent_func = getattr(gevent_module, func_name)
            if not inspect.isfunction(gevent_func) and not funcs_given:
                continue

            func = get_original(module_name, func_name)

            try:
                with warnings.catch_warnings():
                    warnings.simplefilter("ignore")
                    gevent_sig = inspect.getargspec(gevent_func)
                    sig = inspect.getargspec(func)
            except TypeError:
                if funcs_given:
                    raise
                # Can't do this one. If they specifically asked for it,
                # it's an error, otherwise it's not.
                # Python 3 can check a lot more than Python 2 can.
                continue
            self.assertEqual(sig.args, gevent_sig.args, func_name)
            # The next three might not actually matter?
            self.assertEqual(sig.varargs, gevent_sig.varargs, func_name)
            self.assertEqual(sig.keywords, gevent_sig.keywords, func_name)
            self.assertEqual(sig.defaults, gevent_sig.defaults, func_name)

if not hasattr(TestCase, 'assertRaisesRegex'):
    TestCase.assertRaisesRegex = TestCase.assertRaisesRegexp

main = unittest.main
_original_Hub = gevent.hub.Hub


class CountingHub(_original_Hub):

    EXPECTED_TEST_ERROR = (ExpectedException,)

    switch_count = 0

    def switch(self, *args):
        # pylint:disable=arguments-differ
        self.switch_count += 1
        return _original_Hub.switch(self, *args)

    def handle_error(self, context, type, value, tb):
        if issubclass(type, self.EXPECTED_TEST_ERROR):
            # Don't print these to cut down on the noise in the test logs
            return
        return _original_Hub.handle_error(self, context, type, value, tb)

gevent.hub.Hub = CountingHub


class _DelayWaitMixin(object):

    _default_wait_timeout = 0.01
    _default_delay_min_adj = 0.001
    if not RUNNING_ON_APPVEYOR:
        _default_delay_max_adj = 0.11
    else:
        # Timing resolution is extremely poor on Appveyor
        # and subject to jitter.
        _default_delay_max_adj = 1.5

    def wait(self, timeout):
        raise NotImplementedError('override me in subclass')

    def _check_delay_bounds(self, timeout, delay,
                            delay_min_adj=None,
                            delay_max_adj=None):
        delay_min_adj = self._default_delay_min_adj if not delay_min_adj else delay_min_adj
        delay_max_adj = self._default_delay_max_adj if not delay_max_adj else delay_max_adj
        self.assertGreaterEqual(delay, timeout - delay_min_adj)
        self.assertLess(delay, timeout + delay_max_adj)

    def _wait_and_check(self, timeout=None):
        if timeout is None:
            timeout = self._default_wait_timeout

        # gevent.timer instances have a 'seconds' attribute,
        # otherwise it's the raw number
        seconds = getattr(timeout, 'seconds', timeout)

        start = time.time()
        try:
            result = self.wait(timeout)
        finally:
            self._check_delay_bounds(seconds, time.time() - start,
                                     self._default_delay_min_adj,
                                     self._default_delay_max_adj)
        return result

    def test_outer_timeout_is_not_lost(self):
        timeout = gevent.Timeout.start_new(0.001, ref=False)
        try:
            try:
                result = self.wait(timeout=1)
            except gevent.Timeout as ex:
                assert ex is timeout, (ex, timeout)
            else:
                raise AssertionError('must raise Timeout (returned %r)' % (result, ))
        finally:
            timeout.cancel()


class GenericWaitTestCase(_DelayWaitMixin, TestCase):

    _default_wait_timeout = 0.2
    _default_delay_min_adj = 0.1
    if not RUNNING_ON_APPVEYOR:
        _default_delay_max_adj = 0.11
    else:
        # Timing resolution is very poor on Appveyor
        # and subject to jitter
        _default_delay_max_adj = 1.5

    @ignores_leakcheck # waiting checks can be very sensitive to timing
    def test_returns_none_after_timeout(self):
        result = self._wait_and_check()
        # join and wait simply return after timeout expires
        assert result is None, repr(result)


class GenericGetTestCase(_DelayWaitMixin, TestCase):

    Timeout = gevent.Timeout

    def cleanup(self):
        pass

    def test_raises_timeout_number(self):
        self.assertRaises(self.Timeout, self._wait_and_check, timeout=0.01)
        # get raises Timeout after timeout expired
        self.cleanup()

    def test_raises_timeout_Timeout(self):
        timeout = gevent.Timeout(self._default_wait_timeout)
        try:
            self._wait_and_check(timeout=timeout)
        except gevent.Timeout as ex:
            assert ex is timeout, (ex, timeout)
        self.cleanup()

    def test_raises_timeout_Timeout_exc_customized(self):
        error = RuntimeError('expected error')
        timeout = gevent.Timeout(self._default_wait_timeout, exception=error)
        try:
            self._wait_and_check(timeout=timeout)
        except RuntimeError as ex:
            assert ex is error, (ex, error)
        self.cleanup()


def walk_modules(basedir=None, modpath=None, include_so=False, recursive=False):
    if PYPY:
        include_so = False
    if basedir is None:
        basedir = os.path.dirname(gevent.__file__)
        if modpath is None:
            modpath = 'gevent.'
    else:
        if modpath is None:
            modpath = ''
    for fn in sorted(os.listdir(basedir)):
        path = os.path.join(basedir, fn)
        if os.path.isdir(path):
            if not recursive:
                continue
            pkg_init = os.path.join(path, '__init__.py')
            if os.path.exists(pkg_init):
                yield pkg_init, modpath + fn
                for p, m in walk_modules(path, modpath + fn + "."):
                    yield p, m
            continue
        if fn.endswith('.py'):
            x = fn[:-3]
            if x.endswith('_d'):
                x = x[:-2]
            if x in ['__init__', 'core', 'ares', '_util', '_semaphore',
                     'corecffi', '_corecffi', '_corecffi_build']:
                continue
            if x in OPTIONAL_MODULES:
                try:
                    six.exec_("import %s" % x, {})
                except ImportError:
                    continue
            yield path, modpath + x
        elif include_so and fn.endswith(SHARED_OBJECT_EXTENSION):
            if '.pypy-' in fn:
                continue
            if fn.endswith('_d.so'):
                yield path, modpath + fn[:-5]
            else:
                yield path, modpath + fn[:-3]


def bind_and_listen(sock, address=('', 0), backlog=50, reuse_addr=True):
    from socket import SOL_SOCKET, SO_REUSEADDR, error
    if reuse_addr:
        try:
            sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, sock.getsockopt(SOL_SOCKET, SO_REUSEADDR) | 1)
        except error:
            pass
    sock.bind(address)
    sock.listen(backlog)


def tcp_listener(address, backlog=50, reuse_addr=True):
    """A shortcut to create a TCP socket, bind it and put it into listening state."""
    from gevent import socket
    sock = socket.socket()
    bind_and_listen(sock)
    return sock


@contextlib.contextmanager
def disabled_gc():
    was_enabled = gc.isenabled()
    gc.disable()
    try:
        yield
    finally:
        if was_enabled:
            gc.enable()


import re
# Linux/OS X/BSD platforms can implement this by calling out to lsof


if WIN:
    def _run_lsof():
        raise unittest.SkipTest("lsof not expected on Windows")
else:
    def _run_lsof():
        import tempfile
        pid = os.getpid()
        fd, tmpname = tempfile.mkstemp('get_open_files')
        os.close(fd)
        lsof_command = 'lsof -p %s > %s' % (pid, tmpname)
        if os.system(lsof_command):
            # XXX: This prints to the console an annoying message: 'lsof is not recognized'
            raise unittest.SkipTest("lsof failed")
        with open(tmpname) as fobj:
            data = fobj.read().strip()
        os.remove(tmpname)
        return data

def default_get_open_files(pipes=False):
    data = _run_lsof()
    results = {}
    for line in data.split('\n'):
        line = line.strip()
        if not line or line.startswith("COMMAND"):
            # Skip header and blank lines
            continue
        split = re.split(r'\s+', line)
        command, pid, user, fd = split[:4]
        # Pipes (on OS X, at least) get an fd like "3" while normal files get an fd like "1u"
        if fd[:-1].isdigit() or fd.isdigit():
            if not pipes and fd[-1].isdigit():
                continue
            fd = int(fd[:-1]) if not fd[-1].isdigit() else int(fd)
            if fd in results:
                params = (fd, line, split, results.get(fd), data)
                raise AssertionError('error when parsing lsof output: duplicate fd=%r\nline=%r\nsplit=%r\nprevious=%r\ndata:\n%s' % params)
            results[fd] = line
    if not results:
        raise AssertionError('failed to parse lsof:\n%s' % (data, ))
    results['data'] = data
    return results

def default_get_number_open_files():
    if os.path.exists('/proc/'):
        # Linux only
        fd_directory = '/proc/%d/fd' % os.getpid()
        return len(os.listdir(fd_directory))
    else:
        try:
            return len(get_open_files(pipes=True)) - 1
        except (OSError, AssertionError, unittest.SkipTest):
            return 0

lsof_get_open_files = default_get_open_files

try:
    # psutil import subprocess which on Python 3 imports selectors.
    # This can expose issues with monkey-patching.
    import psutil
except ImportError:
    get_open_files = default_get_open_files
    get_number_open_files = default_get_number_open_files
else:
    # If psutil is available (it is cross-platform) use that.
    # It is *much* faster than shelling out to lsof each time
    # (Running 14 tests takes 3.964s with lsof and 0.046 with psutil)
    # However, it still doesn't completely solve the issue on Windows: fds are reported
    # as -1 there, so we can't fully check those.

    def get_open_files():
        """
        Return a list of popenfile and pconn objects.

        Note that other than `fd`, they have different attributes.

        .. important:: If you want to find open sockets, on Windows
           and linux, it is important that the socket at least be listening
           (socket.listen(1)). Unlike the lsof implementation, this will only
           return sockets in a state like that.
        """
        results = dict()
        process = psutil.Process()
        results['data'] = process.open_files() + process.connections('all')
        for x in results['data']:
            results[x.fd] = x
        results['data'] += ['From psutil', process]
        return results

    def get_number_open_files():
        process = psutil.Process()
        try:
            return process.num_fds()
        except AttributeError:
            # num_fds is unix only. Is num_handles close enough on Windows?
            return 0


if PYPY:

    def getrefcount(*args):
        pass

else:

    def getrefcount(*args):
        return sys.getrefcount(*args)

def dump_stacks():
    """
    Request information about the running threads of the current process.

    :return: A sequence of text lines detailing the stacks of running
            threads and greenlets. (One greenlet will duplicate one thread,
            the current thread and greenlet.)
    """
    dump = []

    # threads
    import threading  # Late import this stuff because it may get monkey-patched
    import traceback
    from greenlet import greenlet

    threads = {th.ident: th.name for th in threading.enumerate()}

    for thread, frame in sys._current_frames().items():
        dump.append('Thread 0x%x (%s)\n' % (thread, threads.get(thread)))
        dump.append(''.join(traceback.format_stack(frame)))
        dump.append('\n')

    # greenlets

    # if greenlet is present, let's dump each greenlet stack
    # Use the gc module to inspect all objects to find the greenlets
    # since there isn't a global registry
    for ob in gc.get_objects():
        if not isinstance(ob, greenlet):
            continue
        if not ob:
            continue  # not running anymore or not started
        dump.append('Greenlet %s\n' % ob)
        dump.append(''.join(traceback.format_stack(ob.gr_frame)))
        dump.append('\n')

    return dump