eventlet/tests/timer_test.py

   1 from unittest import TestCase, main
   2
   3 import eventlet
   4 from eventlet import hubs
   5 from eventlet.hubs import timer
   6
   7
   8 class TestTimer(TestCase):
   9     def test_copy(self):
  10         t = timer.Timer(0, lambda: None)
  11         t2 = t.copy()
  12         assert t.seconds == t2.seconds
  13         assert t.tpl == t2.tpl
  14         assert t.called == t2.called
  15
  16     def test_schedule(self):
  17         hub = hubs.get_hub()
  18         # clean up the runloop, preventing side effects from previous tests
  19         # on this thread
  20         if hub.running:
  21             hub.abort()
  22             eventlet.sleep(0)
  23         called = []
  24         # t = timer.Timer(0, lambda: (called.append(True), hub.abort()))
  25         # t.schedule()
  26         # let's have a timer somewhere in the future; make sure abort() still works
  27         # (for pyevent, its dispatcher() does not exit if there is something scheduled)
  28         # XXX pyevent handles this, other hubs do not
  29         # hubs.get_hub().schedule_call_global(10000, lambda: (called.append(True), hub.abort()))
  30         hubs.get_hub().schedule_call_global(0, lambda: (called.append(True), hub.abort()))
  31         hub.default_sleep = lambda: 0.0
  32         hub.switch()
  33         assert called
  34         assert not hub.running
  35
  36
  37 if __name__ == '__main__':
  38     main()