mirror of
https://github.com/correl/SleekXMPP.git
synced 2024-11-24 03:00:15 +00:00
7152d93dd0
A better method than using time.sleep is needed. Maybe use queue.task_done to detect when event processing has ended? Research time!
153 lines
4 KiB
Python
153 lines
4 KiB
Python
import time
|
|
|
|
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)
|
|
|
|
# Give the event queue time to process.
|
|
time.sleep(0.1)
|
|
|
|
# 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.")
|
|
|
|
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)
|
|
|
|
|
|
suite = unittest.TestLoader().loadTestsFromTestCase(TestHandlers)
|