1 # package is named tests, not test, so it won't be confused with test in stdlib
2 from __future__ import print_function
18 from eventlet import tpool
21 # convenience for importers
26 """ Decorator that marks a function as skipped. Uses nose's SkipTest exception
27 if installed. Without nose, this will count skipped tests as passing tests."""
29 from nose.plugins.skip import SkipTest
33 skipme.__name__ = func.__name__
36 # no nose, we'll just skip the test ourselves
38 print(("Skipping {0}".format(func.__name__)))
39 skipme.__name__ = func.__name__
43 def skip_if(condition):
44 """ Decorator that skips a test if the *condition* evaluates True.
45 *condition* can be a boolean or a callable that accepts one argument.
46 The callable will be called with the function to be decorated, and
47 should return True to skip the test.
49 def skipped_wrapper(func):
50 def wrapped(*a, **kw):
51 if isinstance(condition, bool):
54 result = condition(func)
56 return skipped(func)(*a, **kw)
59 wrapped.__name__ = func.__name__
61 return skipped_wrapper
64 def skip_unless(condition):
65 """ Decorator that skips a test if the *condition* does not return True.
66 *condition* can be a boolean or a callable that accepts one argument.
67 The callable will be called with the function to be decorated, and
68 should return True if the condition is satisfied.
70 def skipped_wrapper(func):
71 def wrapped(*a, **kw):
72 if isinstance(condition, bool):
75 result = condition(func)
77 return skipped(func)(*a, **kw)
80 wrapped.__name__ = func.__name__
82 return skipped_wrapper
85 def using_pyevent(_f):
86 from eventlet.hubs import get_hub
87 return 'pyevent' in type(get_hub()).__module__
90 def skip_with_pyevent(func):
91 """ Decorator that skips a test if we're using the pyevent hub."""
92 return skip_if(using_pyevent)(func)
95 def skip_on_windows(func):
96 """ Decorator that skips a test on Windows."""
97 return skip_if(sys.platform.startswith('win'))(func)
100 def skip_if_no_itimer(func):
101 """ Decorator that skips a test if the `itimer` module isn't found """
108 return skip_unless(has_itimer)(func)
111 def skip_if_no_ssl(func):
112 """ Decorator that skips a test if SSL is not available."""
114 import eventlet.green.ssl
118 import eventlet.green.OpenSSL
124 class TestIsTakingTooLong(Exception):
125 """ Custom exception class to be raised when a test's runtime exceeds a limit. """
129 class LimitedTestCase(unittest.TestCase):
130 """ Unittest subclass that adds a timeout to all tests. Subclasses must
131 be sure to call the LimitedTestCase setUp and tearDown methods. The default
132 timeout is 1 second, change it by setting TEST_TIMEOUT to the desired
138 self.previous_alarm = None
139 self.timer = eventlet.Timeout(self.TEST_TIMEOUT,
140 TestIsTakingTooLong(self.TEST_TIMEOUT))
142 def reset_timeout(self, new_timeout):
143 """Changes the timeout duration; only has effect during one test.
144 `new_timeout` can be int or float.
147 self.timer = eventlet.Timeout(new_timeout,
148 TestIsTakingTooLong(new_timeout))
150 def set_alarm(self, new_timeout):
151 """Call this in the beginning of your test if you expect busy loops.
152 Only has effect during one test.
153 `new_timeout` must be int.
155 def sig_alarm_handler(sig, frame):
156 # Could arm previous alarm but test is failed anyway
157 # seems to be no point in restoring previous state.
158 raise TestIsTakingTooLong(new_timeout)
160 self.previous_alarm = (
161 signal.signal(signal.SIGALRM, sig_alarm_handler),
162 signal.alarm(new_timeout),
167 if self.previous_alarm:
168 signal.signal(signal.SIGALRM, self.previous_alarm[0])
169 signal.alarm(self.previous_alarm[1])
176 def assert_less_than(self, a, b, msg=None):
177 msg = msg or "%s not less than %s" % (a, b)
180 assertLessThan = assert_less_than
182 def assert_less_than_equal(self, a, b, msg=None):
183 msg = msg or "%s not less than or equal to %s" % (a, b)
186 assertLessThanEqual = assert_less_than_equal
189 def check_idle_cpu_usage(duration, allowed_part):
191 # TODO: use https://code.google.com/p/psutil/
192 from nose.plugins.skip import SkipTest
193 raise SkipTest('CPU usage testing not supported (`import resource` failed)')
195 r1 = resource.getrusage(resource.RUSAGE_SELF)
196 eventlet.sleep(duration)
197 r2 = resource.getrusage(resource.RUSAGE_SELF)
198 utime = r2.ru_utime - r1.ru_utime
199 stime = r2.ru_stime - r1.ru_stime
200 assert utime + stime < duration * allowed_part, \
201 "CPU usage over limit: user %.0f%% sys %.0f%% allowed %.0f%%" % (
202 utime / duration * 100, stime / duration * 100,
206 def verify_hub_empty():
207 from eventlet import hubs
209 num_readers = len(hub.get_readers())
210 num_writers = len(hub.get_writers())
211 num_timers = hub.get_timers_count()
212 assert num_readers == 0 and num_writers == 0, "Readers: %s Writers: %s" % (
213 num_readers, num_writers)
216 def find_command(command):
217 for dir in os.getenv('PATH', '/usr/bin:/usr/sbin').split(os.pathsep):
218 p = os.path.join(dir, command)
219 if os.access(p, os.X_OK):
221 raise IOError(errno.ENOENT, 'Command not found: %r' % command)
224 def silence_warnings(func):
225 def wrapper(*args, **kw):
226 warnings.simplefilter('ignore', DeprecationWarning)
228 return func(*args, **kw)
230 warnings.simplefilter('default', DeprecationWarning)
231 wrapper.__name__ = func.__name__
235 def get_database_auth():
236 """Retrieves a dict of connection parameters for connecting to test databases.
238 Authentication parameters are highly-machine specific, so
239 get_database_auth gets its information from either environment
240 variables or a config file. The environment variable is
241 "EVENTLET_DB_TEST_AUTH" and it should contain a json object. If
242 this environment variable is present, it's used and config files
243 are ignored. If it's not present, it looks in the local directory
244 (tests) and in the user's home directory for a file named
245 ".test_dbauth", which contains a json map of parameters to the
250 'MySQLdb': {'host': 'localhost', 'user': 'root', 'passwd': ''},
251 'psycopg2': {'user': 'test'},
257 import simplejson as json
259 print("No json implementation, using baked-in db credentials.")
262 if 'EVENTLET_DB_TEST_AUTH' in os.environ:
263 return json.loads(os.environ.get('EVENTLET_DB_TEST_AUTH'))
265 files = [os.path.join(os.path.dirname(__file__), '.test_dbauth'),
266 os.path.join(os.path.expanduser('~'), '.test_dbauth')]
269 auth_utf8 = json.load(open(f))
270 # Have to convert unicode objects to str objects because
271 # mysqldb is dum. Using a doubly-nested list comprehension
272 # because we know that the structure is a two-level dict.
274 [(str(modname), dict(
275 [(str(k), str(v)) for k, v in connectargs.items()]))
276 for modname, connectargs in auth_utf8.items()])
282 def run_python(path):
283 if not path.endswith('.py'):
285 path = os.path.abspath(path)
286 dir_ = os.path.dirname(path)
287 new_env = os.environ.copy()
288 new_env['PYTHONPATH'] = os.pathsep.join(sys.path + [dir_])
289 p = subprocess.Popen(
290 [sys.executable, path],
292 stderr=subprocess.STDOUT,
293 stdin=subprocess.PIPE,
294 stdout=subprocess.PIPE,
296 output, _ = p.communicate()
300 certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt')
301 private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key')