X-Git-Url: https://review.fuel-infra.org/gitweb?a=blobdiff_plain;f=eventlet%2Ftests%2F__init__.py;fp=eventlet%2Ftests%2F__init__.py;h=0000000000000000000000000000000000000000;hb=358bd9258c2b6d2ee74de4dfd07a5123107abad4;hp=3dba2420dea1f4787d243bf0e37b2e84fa7d32ad;hpb=376ff3bfe7071cc0793184a378c4e74508fb0d97;p=packages%2Ftrusty%2Fpython-eventlet.git diff --git a/eventlet/tests/__init__.py b/eventlet/tests/__init__.py deleted file mode 100644 index 3dba242..0000000 --- a/eventlet/tests/__init__.py +++ /dev/null @@ -1,301 +0,0 @@ -# package is named tests, not test, so it won't be confused with test in stdlib -from __future__ import print_function - -import errno -import gc -import os -try: - import resource -except ImportError: - resource = None -import signal -import subprocess -import sys -import unittest -import warnings - -import eventlet -from eventlet import tpool - - -# convenience for importers -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.""" - try: - from nose.plugins.skip import SkipTest - - 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 skip_if(condition): - """ Decorator that skips a test if the *condition* evaluates True. - *condition* can be a boolean or a callable that accepts one argument. - The callable will be called with the function to be decorated, and - should return True to skip the test. - """ - def skipped_wrapper(func): - def wrapped(*a, **kw): - if isinstance(condition, bool): - result = condition - else: - result = condition(func) - if result: - return skipped(func)(*a, **kw) - else: - return func(*a, **kw) - wrapped.__name__ = func.__name__ - return wrapped - return skipped_wrapper - - -def skip_unless(condition): - """ Decorator that skips a test if the *condition* does not return True. - *condition* can be a boolean or a callable that accepts one argument. - The callable will be called with the function to be decorated, and - should return True if the condition is satisfied. - """ - def skipped_wrapper(func): - def wrapped(*a, **kw): - if isinstance(condition, bool): - result = condition - else: - result = condition(func) - if not result: - return skipped(func)(*a, **kw) - else: - return func(*a, **kw) - wrapped.__name__ = func.__name__ - return wrapped - return skipped_wrapper - - -def using_pyevent(_f): - from eventlet.hubs import get_hub - return 'pyevent' in type(get_hub()).__module__ - - -def skip_with_pyevent(func): - """ Decorator that skips a test if we're using the pyevent hub.""" - return skip_if(using_pyevent)(func) - - -def skip_on_windows(func): - """ Decorator that skips a test on Windows.""" - return skip_if(sys.platform.startswith('win'))(func) - - -def skip_if_no_itimer(func): - """ Decorator that skips a test if the `itimer` module isn't found """ - has_itimer = False - try: - import itimer - has_itimer = True - except ImportError: - pass - return skip_unless(has_itimer)(func) - - -def skip_if_no_ssl(func): - """ Decorator that skips a test if SSL is not available.""" - try: - import eventlet.green.ssl - return func - except ImportError: - try: - import eventlet.green.OpenSSL - return func - except ImportError: - return skipped(func) - - -class TestIsTakingTooLong(Exception): - """ Custom exception class to be raised when a test's runtime exceeds a limit. """ - pass - - -class LimitedTestCase(unittest.TestCase): - """ Unittest subclass that adds a timeout to all tests. Subclasses must - be sure to call the LimitedTestCase setUp and tearDown methods. The default - timeout is 1 second, change it by setting TEST_TIMEOUT to the desired - quantity.""" - - TEST_TIMEOUT = 1 - - def setUp(self): - self.previous_alarm = None - self.timer = eventlet.Timeout(self.TEST_TIMEOUT, - TestIsTakingTooLong(self.TEST_TIMEOUT)) - - def reset_timeout(self, new_timeout): - """Changes the timeout duration; only has effect during one test. - `new_timeout` can be int or float. - """ - self.timer.cancel() - self.timer = eventlet.Timeout(new_timeout, - TestIsTakingTooLong(new_timeout)) - - def set_alarm(self, new_timeout): - """Call this in the beginning of your test if you expect busy loops. - Only has effect during one test. - `new_timeout` must be int. - """ - def sig_alarm_handler(sig, frame): - # Could arm previous alarm but test is failed anyway - # seems to be no point in restoring previous state. - raise TestIsTakingTooLong(new_timeout) - - self.previous_alarm = ( - signal.signal(signal.SIGALRM, sig_alarm_handler), - signal.alarm(new_timeout), - ) - - def tearDown(self): - self.timer.cancel() - if self.previous_alarm: - signal.signal(signal.SIGALRM, self.previous_alarm[0]) - signal.alarm(self.previous_alarm[1]) - - tpool.killall() - gc.collect() - eventlet.sleep(0) - verify_hub_empty() - - def assert_less_than(self, a, b, msg=None): - msg = msg or "%s not less than %s" % (a, b) - assert a < b, msg - - assertLessThan = assert_less_than - - def assert_less_than_equal(self, a, b, msg=None): - msg = msg or "%s not less than or equal to %s" % (a, b) - assert a <= b, msg - - assertLessThanEqual = assert_less_than_equal - - -def check_idle_cpu_usage(duration, allowed_part): - if resource is None: - # TODO: use https://code.google.com/p/psutil/ - from nose.plugins.skip import SkipTest - raise SkipTest('CPU usage testing not supported (`import resource` failed)') - - r1 = resource.getrusage(resource.RUSAGE_SELF) - eventlet.sleep(duration) - r2 = resource.getrusage(resource.RUSAGE_SELF) - utime = r2.ru_utime - r1.ru_utime - stime = r2.ru_stime - r1.ru_stime - assert utime + stime < duration * allowed_part, \ - "CPU usage over limit: user %.0f%% sys %.0f%% allowed %.0f%%" % ( - utime / duration * 100, stime / duration * 100, - allowed_part * 100) - - -def verify_hub_empty(): - from eventlet import hubs - hub = hubs.get_hub() - num_readers = len(hub.get_readers()) - num_writers = len(hub.get_writers()) - num_timers = hub.get_timers_count() - assert num_readers == 0 and num_writers == 0, "Readers: %s Writers: %s" % ( - num_readers, num_writers) - - -def find_command(command): - for dir in os.getenv('PATH', '/usr/bin:/usr/sbin').split(os.pathsep): - p = os.path.join(dir, command) - if os.access(p, os.X_OK): - return p - raise IOError(errno.ENOENT, 'Command not found: %r' % command) - - -def silence_warnings(func): - def wrapper(*args, **kw): - warnings.simplefilter('ignore', DeprecationWarning) - try: - return func(*args, **kw) - finally: - warnings.simplefilter('default', DeprecationWarning) - wrapper.__name__ = func.__name__ - return wrapper - - -def get_database_auth(): - """Retrieves a dict of connection parameters for connecting to test databases. - - Authentication parameters are highly-machine specific, so - get_database_auth gets its information from either environment - variables or a config file. The environment variable is - "EVENTLET_DB_TEST_AUTH" and it should contain a json object. If - this environment variable is present, it's used and config files - are ignored. If it's not present, it looks in the local directory - (tests) and in the user's home directory for a file named - ".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')) - - files = [os.path.join(os.path.dirname(__file__), '.test_dbauth'), - os.path.join(os.path.expanduser('~'), '.test_dbauth')] - for f in files: - 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 - # because we know that the structure is a two-level dict. - return dict( - [(str(modname), dict( - [(str(k), str(v)) for k, v in connectargs.items()])) - for modname, connectargs in auth_utf8.items()]) - except IOError: - pass - return retval - - -def run_python(path): - if not path.endswith('.py'): - path += '.py' - path = os.path.abspath(path) - dir_ = os.path.dirname(path) - new_env = os.environ.copy() - new_env['PYTHONPATH'] = os.pathsep.join(sys.path + [dir_]) - p = subprocess.Popen( - [sys.executable, path], - env=new_env, - stderr=subprocess.STDOUT, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - ) - output, _ = p.communicate() - return 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')