import threading
import time
import unittest
from sleekxmpp.test import SleekTest
class TestInBandByteStreams(SleekTest):
def setUp(self):
self.stream_start(plugins=['xep_0047', 'xep_0030'])
def tearDown(self):
self.stream_close()
def testOpenStream(self):
"""Test requesting a stream, successfully"""
events = []
def on_stream_start(stream):
events.append('ibb_stream_start')
self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
t = threading.Thread(name='open_stream',
target=self.xmpp['xep_0047'].open_stream,
args=('tester@localhost/receiver',),
kwargs={'sid': 'testing'})
t.start()
self.send("""
""")
self.recv("""
""")
t.join()
time.sleep(0.2)
self.assertEqual(events, ['ibb_stream_start'])
def testAysncOpenStream(self):
"""Test requesting a stream, aysnc"""
events = set()
def on_stream_start(stream):
events.add('ibb_stream_start')
def stream_callback(iq):
events.add('callback')
self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
t = threading.Thread(name='open_stream',
target=self.xmpp['xep_0047'].open_stream,
args=('tester@localhost/receiver',),
kwargs={'sid': 'testing',
'block': False,
'callback': stream_callback})
t.start()
self.send("""
""")
self.recv("""
""")
t.join()
time.sleep(0.2)
self.assertEqual(events, set(['ibb_stream_start', 'callback']))
def testSendData(self):
"""Test sending data over an in-band bytestream."""
streams = []
data = []
def on_stream_start(stream):
streams.append(stream)
def on_stream_data(d):
data.append(d['data'])
self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
self.xmpp.add_event_handler('ibb_stream_data', on_stream_data)
t = threading.Thread(name='open_stream',
target=self.xmpp['xep_0047'].open_stream,
args=('tester@localhost/receiver',),
kwargs={'sid': 'testing'})
t.start()
self.send("""
""")
self.recv("""
""")
t.join()
time.sleep(0.2)
stream = streams[0]
# Test sending data out
stream.send("Testing")
self.send("""
VGVzdGluZw==
""")
self.recv("""
""")
# Test receiving data
self.recv("""
aXQgd29ya3Mh
""")
self.send("""
""")
self.assertEqual(data, [b'it works!'])
suite = unittest.TestLoader().loadTestsFromTestCase(TestInBandByteStreams)