eventlet/tests/test__event.py

   1 import unittest
   2 from eventlet import spawn, sleep, with_timeout
   3 from eventlet.event import Event
   4 import eventlet
   5 from tests import LimitedTestCase
   6
   7 DELAY = 0.01
   8
   9
  10 class TestEvent(LimitedTestCase):
  11
  12     def test_send_exc(self):
  13         log = []
  14         e = Event()
  15
  16         def waiter():
  17             try:
  18                 result = e.wait()
  19                 log.append(('received', result))
  20             except Exception as ex:
  21                 log.append(('catched', ex))
  22         spawn(waiter)
  23         sleep(0)  # let waiter to block on e.wait()
  24         obj = Exception()
  25         e.send(exc=obj)
  26         sleep(0)
  27         sleep(0)
  28         assert log == [('catched', obj)], log
  29
  30     def test_send(self):
  31         event1 = Event()
  32         event2 = Event()
  33
  34         spawn(event1.send, 'hello event1')
  35         eventlet.Timeout(0, ValueError('interrupted'))
  36         try:
  37             result = event1.wait()
  38         except ValueError:
  39             X = object()
  40             result = with_timeout(DELAY, event2.wait, timeout_value=X)
  41             assert result is X, 'Nobody sent anything to event2 yet it received %r' % (result, )
  42
  43
  44 if __name__ == '__main__':
  45     unittest.main()