X-Git-Url: https://review.fuel-infra.org/gitweb?a=blobdiff_plain;ds=sidebyside;f=eventlet%2Ftests%2Fevent_test.py;fp=eventlet%2Ftests%2Fevent_test.py;h=0000000000000000000000000000000000000000;hb=358bd9258c2b6d2ee74de4dfd07a5123107abad4;hp=c4608d67d2f7167bb0ee9d3ba5417aac771287c4;hpb=376ff3bfe7071cc0793184a378c4e74508fb0d97;p=packages%2Ftrusty%2Fpython-eventlet.git diff --git a/eventlet/tests/event_test.py b/eventlet/tests/event_test.py deleted file mode 100644 index c4608d6..0000000 --- a/eventlet/tests/event_test.py +++ /dev/null @@ -1,81 +0,0 @@ -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)