import logging from sleekxmpp.test import * class TestLiveStream(SleekTest): """ Test that we can test a live stanza stream. """ def tearDown(self): self.stream_close() def testClientConnection(self): """Test that we can interact with a live ClientXMPP instance.""" self.stream_start(mode='client', socket='live', skip=False, jid='user@localhost/test', password='user') # Use sid=None to ignore any id sent by the server since # we can't know it in advance. self.recv_header(sfrom='localhost', sid=None) self.send_header(sto='localhost') self.recv_feature(""" DIGEST-MD5 PLAIN """) self.send_feature(""" """) self.recv_feature(""" """) self.send_header(sto='localhost') self.recv_header(sfrom='localhost', sid=None) self.recv_feature(""" DIGEST-MD5 PLAIN """) self.send_feature(""" AHVzZXIAdXNlcg== """) self.recv_feature(""" """) self.send_header(sto='localhost') self.recv_header(sfrom='localhost', sid=None) self.recv_feature(""" """) # Should really use send, but our Iq stanza objects # can't handle bind element payloads yet. self.send_feature(""" test """) self.recv_feature(""" user@localhost/test """) self.stream_close() suite = unittest.TestLoader().loadTestsFromTestCase(TestLiveStream) if __name__ == '__main__': logging.basicConfig(level=logging.DEBUG, format='%(levelname)-8s %(message)s') tests = unittest.TestSuite([suite]) result = unittest.TextTestRunner(verbosity=2).run(tests) test_ns = 'http://andyet.net/protocol/tests' print("" % ( test_ns, 'ran="%s"' % result.testsRun, 'errors="%s"' % len(result.errors), 'fails="%s"' % len(result.failures), 'success="%s"' % result.wasSuccessful()))