tests/live_multiple_streams.py

   1 import logging
   2
   3 from sleekxmpp.test import *
   4
   5
   6 class TestMultipleStreams(SleekTest):
   7     """
   8     Test that we can test a live stanza stream.
   9     """
  10
  11     def setUp(self):
  12         self.client1 = SleekTest()
  13         self.client2 = SleekTest()
  14
  15     def tearDown(self):
  16         self.client1.stream_close()
  17         self.client2.stream_close()
  18
  19     def testMultipleStreams(self):
  20         """Test that we can interact with multiple live ClientXMPP instance."""
  21
  22         client1 = self.client1
  23         client2 = self.client2
  24
  25         client1.stream_start(mode='client',
  26                              socket='live',
  27                              skip=True,
  28                              jid='user@localhost/test1',
  29                              password='user')
  30         client2.stream_start(mode='client',
  31                              socket='live',
  32                              skip=True,
  33                              jid='user@localhost/test2',
  34                              password='user')
  35
  36         client1.xmpp.send_message(mto='user@localhost/test2',
  37                                   mbody='test')
  38
  39         client1.send('message@body=test', method='stanzapath')
  40         client2.recv('message@body=test', method='stanzapath')
  41
  42
  43 suite = unittest.TestLoader().loadTestsFromTestCase(TestMultipleStreams)
  44
  45 if __name__ == '__main__':
  46     logging.basicConfig(level=logging.DEBUG,
  47                         format='%(levelname)-8s %(message)s')
  48
  49     tests = unittest.TestSuite([suite])
  50     result = unittest.TextTestRunner(verbosity=2).run(tests)
  51     test_ns = 'http://andyet.net/protocol/tests'
  52     print("<tests xmlns='%s' %s %s %s %s />" % (
  53         test_ns,
  54         'ran="%s"' % result.testsRun,
  55         'errors="%s"' % len(result.errors),
  56         'fails="%s"' % len(result.failures),
  57         'success="%s"' % result.wasSuccessful()))