tests/test_events.py

   1 import time
   2 import unittest
   3 from slixmpp.test import SlixTest
   4
   5
   6 class TestEvents(SlixTest):
   7
   8     def setUp(self):
   9         self.stream_start()
  10
  11     def tearDown(self):
  12         self.stream_close()
  13
  14     def testEventHappening(self):
  15         """Test handler working"""
  16         happened = []
  17
  18         def handletestevent(event):
  19             happened.append(True)
  20
  21         self.xmpp.add_event_handler("test_event", handletestevent)
  22         self.xmpp.event("test_event")
  23         self.xmpp.event("test_event")
  24
  25         # Give the event queue time to process.
  26         time.sleep(0.1)
  27
  28         msg = "Event was not triggered the correct number of times: %s"
  29         self.failUnless(happened == [True, True], msg)
  30
  31     def testDelEvent(self):
  32         """Test handler working, then deleted and not triggered"""
  33         happened = []
  34
  35         def handletestevent(event):
  36             happened.append(True)
  37
  38         self.xmpp.add_event_handler("test_event", handletestevent)
  39         self.xmpp.event("test_event", {})
  40
  41         self.xmpp.del_event_handler("test_event", handletestevent)
  42
  43         # Should not trigger because it was deleted
  44         self.xmpp.event("test_event", {})
  45
  46         # Give the event queue time to process.
  47         time.sleep(0.1)
  48
  49         msg = "Event was not triggered the correct number of times: %s"
  50         self.failUnless(happened == [True], msg % happened)
  51
  52     def testAddDelAddEvent(self):
  53         """Test adding, then removing, then adding an event handler."""
  54         happened = []
  55
  56         def handletestevent(event):
  57             happened.append(True)
  58
  59         self.xmpp.add_event_handler("test_event", handletestevent)
  60         self.xmpp.event("test_event", {})
  61
  62         self.xmpp.del_event_handler("test_event", handletestevent)
  63         # Should not trigger because it was deleted
  64         self.xmpp.event("test_event", {})
  65
  66         self.xmpp.add_event_handler("test_event", handletestevent)
  67         self.xmpp.event("test_event", {})
  68
  69         # Give the event queue time to process.
  70         time.sleep(0.1)
  71
  72         msg = "Event was not triggered the correct number of times: %s"
  73         self.failUnless(happened == [True, True], msg % happened)
  74
  75     def testDisposableEvent(self):
  76         """Test disposable handler working, then not being triggered again."""
  77         happened = []
  78
  79         def handletestevent(event):
  80             happened.append(True)
  81
  82         self.xmpp.add_event_handler("test_event", handletestevent,
  83                                     disposable=True)
  84         self.xmpp.event("test_event", {})
  85
  86         # Should not trigger because it was deleted
  87         self.xmpp.event("test_event", {})
  88
  89         # Give the event queue time to process.
  90         time.sleep(0.1)
  91
  92         msg = "Event was not triggered the correct number of times: %s"
  93         self.failUnless(happened == [True], msg % happened)
  94
  95
  96 suite = unittest.TestLoader().loadTestsFromTestCase(TestEvents)