tests/test_stream_xep_0047.py

   1 import threading
   2 import time
   3
   4 import unittest
   5 from slixmpp.test import SlixTest
   6
   7
   8 class TestInBandByteStreams(SlixTest):
   9
  10     def setUp(self):
  11         self.stream_start(plugins=['xep_0047', 'xep_0030'])
  12
  13     def tearDown(self):
  14         self.stream_close()
  15
  16     def testOpenStream(self):
  17         """Test requesting a stream, successfully"""
  18
  19         events = []
  20
  21         def on_stream_start(stream):
  22             events.append('ibb_stream_start')
  23
  24
  25         self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
  26
  27         t = threading.Thread(name='open_stream',
  28                              target=self.xmpp['xep_0047'].open_stream,
  29                              args=('tester@localhost/receiver',),
  30                              kwargs={'sid': 'testing'})
  31         t.start()
  32
  33         self.send("""
  34           <iq type="set" to="tester@localhost/receiver" id="1">
  35             <open xmlns="http://jabber.org/protocol/ibb"
  36                   sid="testing"
  37                   block-size="4096"
  38                   stanza="iq" />
  39           </iq>
  40         """)
  41
  42         self.recv("""
  43           <iq type="result" id="1"
  44               to="tester@localhost"
  45               from="tester@localhost/receiver" />
  46         """)
  47
  48         t.join()
  49
  50         time.sleep(0.2)
  51
  52         self.assertEqual(events, ['ibb_stream_start'])
  53
  54     def testAysncOpenStream(self):
  55         """Test requesting a stream, aysnc"""
  56
  57         events = set()
  58
  59         def on_stream_start(stream):
  60             events.add('ibb_stream_start')
  61
  62         def stream_callback(iq):
  63             events.add('callback')
  64
  65         self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
  66
  67         t = threading.Thread(name='open_stream',
  68                              target=self.xmpp['xep_0047'].open_stream,
  69                              args=('tester@localhost/receiver',),
  70                              kwargs={'sid': 'testing',
  71                                      'block': False,
  72                                      'callback': stream_callback})
  73         t.start()
  74
  75         self.send("""
  76           <iq type="set" to="tester@localhost/receiver" id="1">
  77             <open xmlns="http://jabber.org/protocol/ibb"
  78                   sid="testing"
  79                   block-size="4096"
  80                   stanza="iq" />
  81           </iq>
  82         """)
  83
  84         self.recv("""
  85           <iq type="result" id="1"
  86               to="tester@localhost"
  87               from="tester@localhost/receiver" />
  88         """)
  89
  90         t.join()
  91
  92         time.sleep(0.2)
  93
  94         self.assertEqual(events, set(['ibb_stream_start', 'callback']))
  95
  96     def testSendData(self):
  97         """Test sending data over an in-band bytestream."""
  98
  99         streams = []
 100         data = []
 101
 102         def on_stream_start(stream):
 103             streams.append(stream)
 104
 105         def on_stream_data(d):
 106             data.append(d['data'])
 107
 108         self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
 109         self.xmpp.add_event_handler('ibb_stream_data', on_stream_data)
 110
 111         t = threading.Thread(name='open_stream',
 112                              target=self.xmpp['xep_0047'].open_stream,
 113                              args=('tester@localhost/receiver',),
 114                              kwargs={'sid': 'testing'})
 115         t.start()
 116
 117         self.send("""
 118           <iq type="set" to="tester@localhost/receiver" id="1">
 119             <open xmlns="http://jabber.org/protocol/ibb"
 120                   sid="testing"
 121                   block-size="4096"
 122                   stanza="iq" />
 123           </iq>
 124         """)
 125
 126         self.recv("""
 127           <iq type="result" id="1"
 128               to="tester@localhost"
 129               from="tester@localhost/receiver" />
 130         """)
 131
 132         t.join()
 133
 134         time.sleep(0.2)
 135
 136         stream = streams[0]
 137
 138
 139         # Test sending data out
 140         stream.send("Testing")
 141
 142         self.send("""
 143           <iq type="set" id="2"
 144               from="tester@localhost"
 145               to="tester@localhost/receiver">
 146             <data xmlns="http://jabber.org/protocol/ibb"
 147                   seq="0"
 148                   sid="testing">
 149               VGVzdGluZw==
 150             </data>
 151           </iq>
 152         """)
 153
 154         self.recv("""
 155           <iq type="result" id="2"
 156               to="tester@localhost"
 157               from="tester@localhost/receiver" />
 158         """)
 159
 160         # Test receiving data
 161         self.recv("""
 162           <iq type="set" id="A"
 163               to="tester@localhost"
 164               from="tester@localhost/receiver">
 165             <data xmlns="http://jabber.org/protocol/ibb"
 166                   seq="0"
 167                   sid="testing">
 168               aXQgd29ya3Mh
 169             </data>
 170           </iq>
 171         """)
 172
 173         self.send("""
 174           <iq type="result" id="A"
 175               to="tester@localhost/receiver" />
 176         """)
 177
 178         self.assertEqual(data, [b'it works!'])
 179
 180
 181 suite = unittest.TestLoader().loadTestsFromTestCase(TestInBandByteStreams)