# package is named tests, not test, so it won't be confused with test in stdlib
from __future__ import print_function
+import contextlib
import errno
+import functools
import gc
+import json
import os
try:
import resource
import unittest
import warnings
+from nose.plugins.skip import SkipTest
+
import eventlet
from eventlet import tpool
main = unittest.main
-def skipped(func):
- """ Decorator that marks a function as skipped. Uses nose's SkipTest exception
- if installed. Without nose, this will count skipped tests as passing tests."""
+@contextlib.contextmanager
+def assert_raises(exc_type):
try:
- from nose.plugins.skip import SkipTest
+ yield
+ except exc_type:
+ pass
+ else:
+ name = str(exc_type)
+ try:
+ name = exc_type.__name__
+ except AttributeError:
+ pass
+ assert False, 'Expected exception {0}'.format(name)
- def skipme(*a, **k):
- raise SkipTest()
- skipme.__name__ = func.__name__
- return skipme
- except ImportError:
- # no nose, we'll just skip the test ourselves
- def skipme(*a, **k):
- print(("Skipping {0}".format(func.__name__)))
- skipme.__name__ = func.__name__
- return skipme
+
+def skipped(func, *decorator_args):
+ """Decorator that marks a function as skipped.
+ """
+ @functools.wraps(func)
+ def wrapped(*a, **k):
+ raise SkipTest(*decorator_args)
+
+ return wrapped
def skip_if(condition):
should return True to skip the test.
"""
def skipped_wrapper(func):
+ @functools.wraps(func)
def wrapped(*a, **kw):
if isinstance(condition, bool):
result = condition
else:
result = condition(func)
if result:
- return skipped(func)(*a, **kw)
+ raise SkipTest()
else:
return func(*a, **kw)
- wrapped.__name__ = func.__name__
return wrapped
return skipped_wrapper
should return True if the condition is satisfied.
"""
def skipped_wrapper(func):
+ @functools.wraps(func)
def wrapped(*a, **kw):
if isinstance(condition, bool):
result = condition
else:
result = condition(func)
if not result:
- return skipped(func)(*a, **kw)
+ raise SkipTest()
else:
return func(*a, **kw)
- wrapped.__name__ = func.__name__
return wrapped
return skipped_wrapper
def verify_hub_empty():
+
+ def format_listener(listener):
+ return 'Listener %r for greenlet %r with run callback %r' % (
+ listener, listener.greenlet, getattr(listener.greenlet, 'run', None))
+
from eventlet import hubs
hub = hubs.get_hub()
- num_readers = len(hub.get_readers())
- num_writers = len(hub.get_writers())
+ readers = hub.get_readers()
+ writers = hub.get_writers()
+ num_readers = len(readers)
+ num_writers = len(writers)
num_timers = hub.get_timers_count()
- assert num_readers == 0 and num_writers == 0, "Readers: %s Writers: %s" % (
- num_readers, num_writers)
+ assert num_readers == 0 and num_writers == 0, \
+ "Readers: %s (%d) Writers: %s (%d)" % (
+ ', '.join(map(format_listener, readers)), num_readers,
+ ', '.join(map(format_listener, writers)), num_writers,
+ )
def find_command(command):
".test_dbauth", which contains a json map of parameters to the
connect function.
"""
- import os
retval = {
'MySQLdb': {'host': 'localhost', 'user': 'root', 'passwd': ''},
'psycopg2': {'user': 'test'},
}
- try:
- import json
- except ImportError:
- try:
- import simplejson as json
- except ImportError:
- print("No json implementation, using baked-in db credentials.")
- return retval
if 'EVENTLET_DB_TEST_AUTH' in os.environ:
return json.loads(os.environ.get('EVENTLET_DB_TEST_AUTH'))
try:
auth_utf8 = json.load(open(f))
# Have to convert unicode objects to str objects because
- # mysqldb is dum. Using a doubly-nested list comprehension
+ # mysqldb is dumb. Using a doubly-nested list comprehension
# because we know that the structure is a two-level dict.
return dict(
[(str(modname), dict(
if not path.endswith('.py'):
path += '.py'
path = os.path.abspath(path)
- dir_ = os.path.dirname(path)
+ src_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
new_env = os.environ.copy()
- new_env['PYTHONPATH'] = os.pathsep.join(sys.path + [dir_])
+ new_env['PYTHONPATH'] = os.pathsep.join(sys.path + [src_dir])
p = subprocess.Popen(
[sys.executable, path],
env=new_env,
return output
+def run_isolated(path, prefix='tests/isolated/'):
+ output = run_python(prefix + path).rstrip()
+ if output.startswith(b'skip'):
+ parts = output.split(b':', 1)
+ skip_args = []
+ if len(parts) > 1:
+ skip_args.append(parts[1])
+ raise SkipTest(*skip_args)
+ assert output == b'pass', output
+
+
certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt')
private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key')