""" Tests with-statement behavior of Timeout class. Don't import when using Python 2.4. """ from __future__ import with_statement import sys import unittest import weakref import time from eventlet import sleep from eventlet.timeout import Timeout from tests import LimitedTestCase DELAY = 0.01 class Error(Exception): pass class Test(LimitedTestCase): def test_cancellation(self): # Nothing happens if with-block finishes before the timeout expires t = Timeout(DELAY*2) sleep(0) # make it pending assert t.pending, repr(t) with t: assert t.pending, repr(t) sleep(DELAY) # check if timer was actually cancelled assert not t.pending, repr(t) sleep(DELAY*2) def test_raising_self(self): # An exception will be raised if it's not try: with Timeout(DELAY) as t: sleep(DELAY*2) except Timeout, ex: assert ex is t, (ex, t) else: raise AssertionError('must raise Timeout') def test_raising_self_true(self): # specifying True as the exception raises self as well try: with Timeout(DELAY, True) as t: sleep(DELAY*2) except Timeout, ex: assert ex is t, (ex, t) else: raise AssertionError('must raise Timeout') def test_raising_custom_exception(self): # You can customize the exception raised: try: with Timeout(DELAY, IOError("Operation takes way too long")): sleep(DELAY*2) except IOError, ex: assert str(ex)=="Operation takes way too long", repr(ex) def test_raising_exception_class(self): # Providing classes instead of values should be possible too: try: with Timeout(DELAY, ValueError): sleep(DELAY*2) except ValueError: pass def test_raising_exc_tuple(self): try: 1//0 except: try: with Timeout(DELAY, sys.exc_info()[0]): sleep(DELAY*2) raise AssertionError('should not get there') raise AssertionError('should not get there') except ZeroDivisionError: pass else: raise AssertionError('should not get there') def test_cancel_timer_inside_block(self): # It's possible to cancel the timer inside the block: with Timeout(DELAY) as timer: timer.cancel() sleep(DELAY*2) def test_silent_block(self): # To silence the exception before exiting the block, pass # False as second parameter. XDELAY=0.1 start = time.time() with Timeout(XDELAY, False): sleep(XDELAY*2) delta = (time.time()-start) assert delta