summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/test_stanza_xep_0047.py13
-rw-r--r--tests/test_stream_xep_0047.py98
2 files changed, 111 insertions, 0 deletions
diff --git a/tests/test_stanza_xep_0047.py b/tests/test_stanza_xep_0047.py
index 1b212529..6aa2314b 100644
--- a/tests/test_stanza_xep_0047.py
+++ b/tests/test_stanza_xep_0047.py
@@ -73,5 +73,18 @@ class TestIBB(SleekTest):
self.assertTrue(errored, "ABCD?EFGH did not raise base64 error")
+ def testConvertData(self):
+ """Test that data is converted to base64"""
+ iq = Iq()
+ iq['type'] = 'set'
+ iq['ibb_data']['seq'] = 0
+ iq['ibb_data']['data'] = 'sleekxmpp'
+
+ self.check(iq, """
+ <iq type="set">
+ <data xmlns="http://jabber.org/protocol/ibb" seq="0">c2xlZWt4bXBw</data>
+ </iq>
+ """)
+
suite = unittest.TestLoader().loadTestsFromTestCase(TestIBB)
diff --git a/tests/test_stream_xep_0047.py b/tests/test_stream_xep_0047.py
new file mode 100644
index 00000000..485dafe5
--- /dev/null
+++ b/tests/test_stream_xep_0047.py
@@ -0,0 +1,98 @@
+import threading
+import time
+
+from sleekxmpp.test import *
+
+
+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("""
+ <iq type="set" to="tester@localhost/receiver" id="1">
+ <open xmlns="http://jabber.org/protocol/ibb"
+ sid="testing"
+ block-size="4096"
+ stanza="iq" />
+ </iq>
+ """)
+
+ self.recv("""
+ <iq type="result" id="1"
+ to="tester@localhost"
+ from="tester@localhost/receiver" />
+ """)
+
+ 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("""
+ <iq type="set" to="tester@localhost/receiver" id="1">
+ <open xmlns="http://jabber.org/protocol/ibb"
+ sid="testing"
+ block-size="4096"
+ stanza="iq" />
+ </iq>
+ """)
+
+ self.recv("""
+ <iq type="result" id="1"
+ to="tester@localhost"
+ from="tester@localhost/receiver" />
+ """)
+
+ t.join()
+
+ time.sleep(0.2)
+
+ self.assertEqual(events, set(['ibb_stream_start', 'callback']))
+
+
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestInBandByteStreams)