github/SleekXMPP
1
0
Fork 0
mirror of https://github.com/correl/SleekXMPP.git synced 2024-11-30 19:19:55 +00:00
Code Issues Projects Releases Packages Wiki Activity
SleekXMPP/tests/test_stream_handlers.py

151 lines
3.9 KiB
Python
Raw Permalink Normal View History

Added option for iq.send to accept a callhandler. The callback will be a stream level handler, and will not execute in its own thread. If you must have a thread, have the callback function raise a custom event, which can be processed by another event handler, which may run in an individual thread, like so: def handle_reply(self, iq): self.event('custom_event', iq) def do_long_operation_in_thread(self, iq): ... self.add_event_handler('custom_event', self.do_long_operation_in_thread) ...take out already prepared iq stanza... iq.send(callback=self.handle_reply)
2010-12-07 22:19:39 +00:00
import time
Unit test reorganization. Moved SleekTest to sleekxmpp.test. Organized test suites by their focus. - Suites focused on testing stanza objects are named test_stanza_X.py - Suites focused on testing stream behavior are name test_stream_X.py
2010-10-07 14:58:13 +00:00
from sleekxmpp.test import *
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
from sleekxmpp.xmlstream.handler import *
from sleekxmpp.xmlstream.matcher import *
Unit test reorganization. Moved SleekTest to sleekxmpp.test. Organized test suites by their focus. - Suites focused on testing stanza objects are named test_stanza_X.py - Suites focused on testing stream behavior are name test_stream_X.py
2010-10-07 14:58:13 +00:00
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
class TestHandlers(SleekTest):
"""
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
Test using handlers and waiters.
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
"""
def setUp(self):
Changed SleekTest to use underscored names.
2010-10-07 13:22:27 +00:00
self.stream_start()
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
def tearDown(self):
Changed SleekTest to use underscored names.
2010-10-07 13:22:27 +00:00
self.stream_close()
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
def testCallback(self):
"""Test using stream callback handlers."""
def callback_handler(stanza):
self.xmpp.sendRaw("""
<message>
<body>Success!</body>
</message>
""")
callback = Callback('Test Callback',
MatchXPath('{test}tester'),
callback_handler)
self.xmpp.registerHandler(callback)
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("""<tester xmlns="test" />""")
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
msg = self.Message()
msg['body'] = 'Success!'
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(msg)
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
def testWaiter(self):
"""Test using stream waiter handler."""
def waiter_handler(stanza):
iq = self.xmpp.Iq()
iq['id'] = 'test'
iq['type'] = 'set'
iq['query'] = 'test'
reply = iq.send(block=True)
if reply:
self.xmpp.sendRaw("""
<message>
<body>Successful: %s</body>
</message>
""" % reply['query'])
self.xmpp.add_event_handler('message', waiter_handler, threaded=True)
# Send message to trigger waiter_handler
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 the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
<message>
<body>Testing</body>
</message>
""")
# Check that Iq was sent by waiter_handler
iq = self.Iq()
iq['id'] = 'test'
iq['type'] = 'set'
iq['query'] = 'test'
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(iq)
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
# Send the reply Iq
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 the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
<iq id="test" type="result">
<query xmlns="test" />
</iq>
""")
# Check that waiter_handler received the reply
msg = self.Message()
msg['body'] = 'Successful: test'
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(msg)
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
def testWaiterTimeout(self):
"""Test that waiter handler is removed after timeout."""
def waiter_handler(stanza):
iq = self.xmpp.Iq()
iq['id'] = 'test2'
iq['type'] = 'set'
iq['query'] = 'test2'
reply = iq.send(block=True, timeout=0)
self.xmpp.add_event_handler('message', waiter_handler, threaded=True)
# Start test by triggerig waiter_handler
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("""<message><body>Start Test</body></message>""")
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
# Check that Iq was sent to trigger start of timeout period
iq = self.Iq()
iq['id'] = 'test2'
iq['type'] = 'set'
iq['query'] = 'test2'
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(iq)
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
# Check that the waiter is no longer registered
waiter_exists = self.xmpp.removeHandler('IqWait_test2')
self.failUnless(waiter_exists == False,
"Waiter handler was not removed.")
Added option for iq.send to accept a callhandler. The callback will be a stream level handler, and will not execute in its own thread. If you must have a thread, have the callback function raise a custom event, which can be processed by another event handler, which may run in an individual thread, like so: def handle_reply(self, iq): self.event('custom_event', iq) def do_long_operation_in_thread(self, iq): ... self.add_event_handler('custom_event', self.do_long_operation_in_thread) ...take out already prepared iq stanza... iq.send(callback=self.handle_reply)
2010-12-07 22:19:39 +00:00
def testIqCallback(self):
"""Test that iq.send(callback=handle_foo) works."""
events = []
def handle_foo(iq):
events.append('foo')
iq = self.Iq()
iq['type'] = 'get'
iq['id'] = 'test-foo'
iq['to'] = 'user@localhost'
iq['query'] = 'foo'
iq.send(callback=handle_foo)
self.send("""
<iq type="get" id="test-foo" to="user@localhost">
<query xmlns="foo" />
</iq>
""")
self.recv("""
<iq type="result" id="test-foo"
to="test@localhost"
from="user@localhost">
<query xmlns="foo">
<data />
</query>
</iq>
""")
# Give event queue time to process
time.sleep(0.1)
self.failUnless(events == ['foo'],
"Iq callback was not executed: %s" % events)
Updated the suite of handler classes with documentation. Updated XMLStream to return True or False from removeHandler to indicate if the handler existed and was removed. Waiter handlers now unregister themselves after timing out.
2010-08-27 20:42:26 +00:00
suite = unittest.TestLoader().loadTestsFromTestCase(TestHandlers)
Reference in a new issue Copy permalink