+++ /dev/null
-import eventlet
-from eventlet import event
-from tests import LimitedTestCase
-
-
-class TestEvent(LimitedTestCase):
- def test_waiting_for_event(self):
- evt = event.Event()
- value = 'some stuff'
-
- def send_to_event():
- evt.send(value)
- eventlet.spawn_n(send_to_event)
- self.assertEqual(evt.wait(), value)
-
- def test_multiple_waiters(self):
- self._test_multiple_waiters(False)
-
- def test_multiple_waiters_with_exception(self):
- self._test_multiple_waiters(True)
-
- def _test_multiple_waiters(self, exception):
- evt = event.Event()
- results = []
-
- def wait_on_event(i_am_done):
- evt.wait()
- results.append(True)
- i_am_done.send()
- if exception:
- raise Exception()
-
- waiters = []
- count = 5
- for i in range(count):
- waiters.append(event.Event())
- eventlet.spawn_n(wait_on_event, waiters[-1])
- eventlet.sleep() # allow spawns to start executing
- evt.send()
-
- for w in waiters:
- w.wait()
-
- self.assertEqual(len(results), count)
-
- def test_reset(self):
- evt = event.Event()
-
- # calling reset before send should throw
- self.assertRaises(AssertionError, evt.reset)
-
- value = 'some stuff'
-
- def send_to_event():
- evt.send(value)
- eventlet.spawn_n(send_to_event)
- self.assertEqual(evt.wait(), value)
-
- # now try it again, and we should get the same exact value,
- # and we shouldn't be allowed to resend without resetting
- value2 = 'second stuff'
- self.assertRaises(AssertionError, evt.send, value2)
- self.assertEqual(evt.wait(), value)
-
- # reset and everything should be happy
- evt.reset()
-
- def send_to_event2():
- evt.send(value2)
- eventlet.spawn_n(send_to_event2)
- self.assertEqual(evt.wait(), value2)
-
- def test_double_exception(self):
- evt = event.Event()
- # send an exception through the event
- evt.send(exc=RuntimeError('from test_double_exception'))
- self.assertRaises(RuntimeError, evt.wait)
- evt.reset()
- # shouldn't see the RuntimeError again
- eventlet.Timeout(0.001)
- self.assertRaises(eventlet.Timeout, evt.wait)