tests/test_stream_filters.py

   1 import time
   2
   3 from slixmpp import Message
   4 import unittest
   5 from slixmpp.test import SlixTest
   6
   7
   8 class TestFilters(SlixTest):
   9
  10     """
  11     Test using incoming and outgoing filters.
  12     """
  13
  14     def setUp(self):
  15         self.stream_start()
  16
  17     def tearDown(self):
  18         self.stream_close()
  19
  20     def testIncoming(self):
  21
  22         data = []
  23
  24         def in_filter(stanza):
  25             if isinstance(stanza, Message):
  26                 if stanza['body'] == 'testing':
  27                     stanza['subject'] = stanza['body'] + ' filter'
  28                     print('>>> %s' % stanza['subject'])
  29             return stanza
  30
  31         def on_message(msg):
  32             print('<<< %s' % msg['subject'])
  33             data.append(msg['subject'])
  34
  35         self.xmpp.add_filter('in', in_filter)
  36         self.xmpp.add_event_handler('message', on_message)
  37
  38         self.recv("""
  39           <message>
  40             <body>no filter</body>
  41           </message>
  42         """)
  43
  44         self.recv("""
  45           <message>
  46             <body>testing</body>
  47           </message>
  48         """)
  49
  50         time.sleep(0.5)
  51
  52         self.assertEqual(data, ['', 'testing filter'],
  53                 'Incoming filter did not apply %s' % data)
  54
  55     def testOutgoing(self):
  56
  57         def out_filter(stanza):
  58             if isinstance(stanza, Message):
  59                 if stanza['body'] == 'testing':
  60                     stanza['body'] = 'changed!'
  61             return stanza
  62
  63         self.xmpp.add_filter('out', out_filter)
  64
  65         m1 = self.Message()
  66         m1['body'] = 'testing'
  67         m1.send()
  68
  69         m2 = self.Message()
  70         m2['body'] = 'blah'
  71         m2.send()
  72
  73         self.send("""
  74           <message>
  75             <body>changed!</body>
  76           </message>
  77         """)
  78
  79         self.send("""
  80           <message>
  81             <body>blah</body>
  82           </message>
  83         """)
  84
  85
  86
  87 suite = unittest.TestLoader().loadTestsFromTestCase(TestFilters)