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

203 lines
5.4 KiB
Python
Raw 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
Add unit test for copying stanzas when passed to events.
2011-11-20 02:43:38 +00:00
from sleekxmpp import Message
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'
Begin experimental use of exceptions. Provides IqTimeout and IqError which are raised when an Iq response does not arrive in time, or it arrives with type='error'.
2011-05-31 19:48:43 +00:00
try:
reply = iq.send(block=True, timeout=0)
except IqTimeout:
pass
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
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
Fix test timeout issue. A better method than using time.sleep is needed. Maybe use queue.task_done to detect when event processing has ended? Research time!
2011-05-21 01:38:10 +00:00
# Give the event queue time to process.
time.sleep(0.1)
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)
Add unit test for copying stanzas when passed to events.
2011-11-20 02:43:38 +00:00
def testMultipleHandlersForStanza(self):
"""
Test that multiple handlers for a single stanza work
without clobbering each other.
"""
def handler_1(msg):
msg.reply("Handler 1: %s" % msg['body']).send()
def handler_2(msg):
msg.reply("Handler 2: %s" % msg['body']).send()
def handler_3(msg):
msg.reply("Handler 3: %s" % msg['body']).send()
self.stream_start()
self.xmpp.add_event_handler('message', handler_1)
self.xmpp.add_event_handler('message', handler_2)
self.xmpp.add_event_handler('message', handler_3)
self.recv("""
<message to="tester@localhost" from="user@example.com">
<body>Testing</body>
</message>
""")
# This test is brittle, depending on the fact that handlers
# will be checked in the order they are registered.
self.send("""
<message to="user@example.com">
<body>Handler 1: Testing</body>
</message>
""")
self.send("""
<message to="user@example.com">
<body>Handler 2: Testing</body>
</message>
""")
self.send("""
<message to="user@example.com">
<body>Handler 3: Testing</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
suite = unittest.TestLoader().loadTestsFromTestCase(TestHandlers)
Reference in a new issue Copy permalink