mirror of
https://github.com/correl/SleekXMPP.git
synced 2024-11-24 03:00:15 +00:00
d0c506f930
* 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.
112 lines
3 KiB
Python
112 lines
3 KiB
Python
from sleekxmpp.test import *
|
|
from sleekxmpp.xmlstream.handler import *
|
|
from sleekxmpp.xmlstream.matcher import *
|
|
|
|
|
|
class TestHandlers(SleekTest):
|
|
"""
|
|
Test using handlers and waiters.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.stream_start()
|
|
|
|
def tearDown(self):
|
|
self.stream_close()
|
|
|
|
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)
|
|
|
|
self.recv("""<tester xmlns="test" />""")
|
|
|
|
msg = self.Message()
|
|
msg['body'] = 'Success!'
|
|
self.send(msg)
|
|
|
|
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
|
|
self.recv("""
|
|
<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'
|
|
self.send(iq)
|
|
|
|
# Send the reply Iq
|
|
self.recv("""
|
|
<iq id="test" type="result">
|
|
<query xmlns="test" />
|
|
</iq>
|
|
""")
|
|
|
|
# Check that waiter_handler received the reply
|
|
msg = self.Message()
|
|
msg['body'] = 'Successful: test'
|
|
self.send(msg)
|
|
|
|
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
|
|
self.recv("""<message><body>Start Test</body></message>""")
|
|
|
|
# Check that Iq was sent to trigger start of timeout period
|
|
iq = self.Iq()
|
|
iq['id'] = 'test2'
|
|
iq['type'] = 'set'
|
|
iq['query'] = 'test2'
|
|
self.send(iq)
|
|
|
|
# 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.")
|
|
|
|
|
|
suite = unittest.TestLoader().loadTestsFromTestCase(TestHandlers)
|