slixmpp/tests/test_stream.py

import time
import unittest
from sleekxmpp.test import SleekTest


class TestStreamTester(SleekTest):
    """
    Test that we can simulate and test a stanza stream.
    """

    def tearDown(self):
        self.stream_close()

    def testClientEcho(self):
        """Test that we can interact with a ClientXMPP instance."""
        self.stream_start(mode='client')

        def echo(msg):
            msg.reply('Thanks for sending: %(body)s' % msg).send()

        self.xmpp.add_event_handler('message', echo)

        self.recv("""
          <message to="tester@localhost" from="user@localhost">
            <body>Hi!</body>
          </message>
        """)

        self.send("""
          <message to="user@localhost">
            <body>Thanks for sending: Hi!</body>
          </message>
        """)

    def testComponentEcho(self):
        """Test that we can interact with a ComponentXMPP instance."""
        self.stream_start(mode='component')

        def echo(msg):
            msg.reply('Thanks for sending: %(body)s' % msg).send()

        self.xmpp.add_event_handler('message', echo)

        self.recv("""
          <message to="tester.localhost" from="user@localhost">
            <body>Hi!</body>
          </message>
        """)

        self.send("""
          <message to="user@localhost" from="tester.localhost">
            <body>Thanks for sending: Hi!</body>
          </message>
        """)

    def testSendStreamHeader(self):
        """Test that we can check a sent stream header."""
        self.stream_start(mode='client', skip=False)
        self.send_header(sto='localhost')

    def testStreamDisconnect(self):
        """Test that the test socket can simulate disconnections."""
        self.stream_start()
        events = set()

        def stream_error(event):
            events.add('socket_error')

        self.xmpp.add_event_handler('socket_error', stream_error)

        self.stream_disconnect()
        self.xmpp.send_raw('  ')

        time.sleep(.1)

        self.failUnless('socket_error' in events,
                "Stream error event not raised: %s" % events)


suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamTester)
Added support for testind disconnect errors. 2011-05-27 18:01:30 +00:00			`import time`
refactor: no import * in tests 2013-07-26 11:02:26 +00:00			`import unittest`
			`from sleekxmpp.test import SleekTest`
Updated SleekTest to be able to simulate and test interactions with an XML stream. 2010-07-14 19:24:37 +00:00

			`class TestStreamTester(SleekTest):`
			`"""`
			`Test that we can simulate and test a stanza stream.`
			`"""`

			`def tearDown(self):`
Changed SleekTest to use underscored names. 2010-10-07 13:22:27 +00:00			`self.stream_close()`
Updated SleekTest to be able to simulate and test interactions with an XML stream. 2010-07-14 19:24:37 +00:00
Updated SleekTest and related tests. May now use a component for stream testing. Methods provided for testing stream headers. 2010-10-06 22:10:04 +00:00			`def testClientEcho(self):`
			`"""Test that we can interact with a ClientXMPP instance."""`
Changed SleekTest to use underscored names. 2010-10-07 13:22:27 +00:00			`self.stream_start(mode='client')`
Updated SleekTest and related tests. May now use a component for stream testing. Methods provided for testing stream headers. 2010-10-06 22:10:04 +00:00
Updated SleekTest to be able to simulate and test interactions with an XML stream. 2010-07-14 19:24:37 +00:00			`def echo(msg):`
			`msg.reply('Thanks for sending: %(body)s' % msg).send()`
Corrected stream header tester. Added test for testing stream headers. 2010-10-06 22:46:23 +00:00
Updated SleekTest to be able to simulate and test interactions with an XML stream. 2010-07-14 19:24:37 +00:00			`self.xmpp.add_event_handler('message', echo)`
Corrected stream header tester. Added test for testing stream headers. 2010-10-06 22:46:23 +00:00
Simplified SleekTest. * check_stanza does not require stanza_class parameter. Introspection! * check_message, check_iq, and check_presence removed -- use check instead. * stream_send_stanza, stream_send_message, stream_send_iq, and stream_send_presence removed -- use send instead. * Use recv instead of recv_message, recv_presence, etc. * check_jid instead of check_JID * stream_start may accept multi=True to return a new SleekTest instance for testing multiple streams at once. 2010-11-05 18:45:58 +00:00			`self.recv("""`
Updated SleekTest to be able to simulate and test interactions with an XML stream. 2010-07-14 19:24:37 +00:00			`<message to="tester@localhost" from="user@localhost">`
			`<body>Hi!</body>`
			`</message>`
			`""")`
Corrected stream header tester. Added test for testing stream headers. 2010-10-06 22:46:23 +00:00
Simplified SleekTest. * check_stanza does not require stanza_class parameter. Introspection! * check_message, check_iq, and check_presence removed -- use check instead. * stream_send_stanza, stream_send_message, stream_send_iq, and stream_send_presence removed -- use send instead. * Use recv instead of recv_message, recv_presence, etc. * check_jid instead of check_JID * stream_start may accept multi=True to return a new SleekTest instance for testing multiple streams at once. 2010-11-05 18:45:58 +00:00			`self.send("""`
Update test to reflect change in reply() method that removes the from attribute. 2010-07-27 01:38:23 +00:00			`<message to="user@localhost">`
Updated SleekTest to be able to simulate and test interactions with an XML stream. 2010-07-14 19:24:37 +00:00			`<body>Thanks for sending: Hi!</body>`
			`</message>`
			`""")`

Updated SleekTest and related tests. May now use a component for stream testing. Methods provided for testing stream headers. 2010-10-06 22:10:04 +00:00			`def testComponentEcho(self):`
			`"""Test that we can interact with a ComponentXMPP instance."""`
Changed SleekTest to use underscored names. 2010-10-07 13:22:27 +00:00			`self.stream_start(mode='component')`
Updated SleekTest and related tests. May now use a component for stream testing. Methods provided for testing stream headers. 2010-10-06 22:10:04 +00:00
			`def echo(msg):`
			`msg.reply('Thanks for sending: %(body)s' % msg).send()`

			`self.xmpp.add_event_handler('message', echo)`

Simplified SleekTest. * check_stanza does not require stanza_class parameter. Introspection! * check_message, check_iq, and check_presence removed -- use check instead. * stream_send_stanza, stream_send_message, stream_send_iq, and stream_send_presence removed -- use send instead. * Use recv instead of recv_message, recv_presence, etc. * check_jid instead of check_JID * stream_start may accept multi=True to return a new SleekTest instance for testing multiple streams at once. 2010-11-05 18:45:58 +00:00			`self.recv("""`
Updated SleekTest and related tests. May now use a component for stream testing. Methods provided for testing stream headers. 2010-10-06 22:10:04 +00:00			`<message to="tester.localhost" from="user@localhost">`
			`<body>Hi!</body>`
			`</message>`
			`""")`

Simplified SleekTest. * check_stanza does not require stanza_class parameter. Introspection! * check_message, check_iq, and check_presence removed -- use check instead. * stream_send_stanza, stream_send_message, stream_send_iq, and stream_send_presence removed -- use send instead. * Use recv instead of recv_message, recv_presence, etc. * check_jid instead of check_JID * stream_start may accept multi=True to return a new SleekTest instance for testing multiple streams at once. 2010-11-05 18:45:58 +00:00			`self.send("""`
Updated SleekTest and related tests. May now use a component for stream testing. Methods provided for testing stream headers. 2010-10-06 22:10:04 +00:00			`<message to="user@localhost" from="tester.localhost">`
			`<body>Thanks for sending: Hi!</body>`
			`</message>`
			`""")`

Corrected stream header tester. Added test for testing stream headers. 2010-10-06 22:46:23 +00:00			`def testSendStreamHeader(self):`
			`"""Test that we can check a sent stream header."""`
Changed SleekTest to use underscored names. 2010-10-07 13:22:27 +00:00			`self.stream_start(mode='client', skip=False)`
Simplified SleekTest. * check_stanza does not require stanza_class parameter. Introspection! * check_message, check_iq, and check_presence removed -- use check instead. * stream_send_stanza, stream_send_message, stream_send_iq, and stream_send_presence removed -- use send instead. * Use recv instead of recv_message, recv_presence, etc. * check_jid instead of check_JID * stream_start may accept multi=True to return a new SleekTest instance for testing multiple streams at once. 2010-11-05 18:45:58 +00:00			`self.send_header(sto='localhost')`
Corrected stream header tester. Added test for testing stream headers. 2010-10-06 22:46:23 +00:00
Added support for testind disconnect errors. 2011-05-27 18:01:30 +00:00			`def testStreamDisconnect(self):`
			`"""Test that the test socket can simulate disconnections."""`
			`self.stream_start()`
			`events = set()`

			`def stream_error(event):`
			`events.add('socket_error')`

			`self.xmpp.add_event_handler('socket_error', stream_error)`

			`self.stream_disconnect()`
			`self.xmpp.send_raw(' ')`

			`time.sleep(.1)`

			`self.failUnless('socket_error' in events,`
			`"Stream error event not raised: %s" % events)`


Updated SleekTest to be able to simulate and test interactions with an XML stream. 2010-07-14 19:24:37 +00:00			`suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamTester)`