pep/slixmpp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity
slixmpp/tests/test_stream_roster.py

163 lines
4.7 KiB
Python
Raw Normal View History

Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
from sleekxmpp.test import *
import time
import threading
class TestStreamRoster(SleekTest):
"""
Test handling roster updates.
"""
def tearDown(self):
self.stream_close()
def testGetRoster(self):
"""Test handling roster requests."""
First pass at integrating the new roster manager.
2010-10-27 03:47:17 +00:00
self.stream_start(mode='client', jid='tester@localhost')
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
Resolve timeout errors for get_roster. See issue #89 Using get_roster will now return the same types of values as Iq.send. If a timeout occurs, then the event 'roster_timeout' will be fired. A successful call to get_roster will also raise the 'roster_received' event. To ensure that the get_roster call was successful, here is a pattern to follow: def __init__(self, ...): ... self.add_event_handler('session_start', self.session_start) self.add_event_handler('roster_timeout', self.roster_timeout) self.add_event_handler('roster_received', self.roster_received) def session_start(self, e): self.send_presence() self.get_roster() def roster_timeout(self, e): # Optionally increase the timeout period self.get_roster(timeout=self.response_timeout * 2) def roster_received(self, iq): # Do stuff, roster has been initialized. ...
2011-05-20 16:56:00 +00:00
events = []
def roster_received(iq):
events.append('roster_received')
self.xmpp.add_event_handler('roster_received', roster_received)
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
# Since get_roster blocks, we need to run it in a thread.
t = threading.Thread(name='get_roster', target=self.xmpp.get_roster)
t.start()
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("""
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
<iq type="get" id="1">
<query xmlns="jabber:iq:roster" />
</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("""
First pass at integrating the new roster manager.
2010-10-27 03:47:17 +00:00
<iq to='tester@localhost' type="result" id="1">
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
<query xmlns="jabber:iq:roster">
<item jid="user@localhost"
name="User"
First pass at integrating the new roster manager.
2010-10-27 03:47:17 +00:00
subscription="from"
ask="subscribe">
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
<group>Friends</group>
<group>Examples</group>
</item>
</query>
</iq>
""")
# Wait for get_roster to return.
t.join()
First pass at integrating the new roster manager.
2010-10-27 03:47:17 +00:00
self.check_roster('tester@localhost', 'user@localhost',
name='User',
subscription='from',
afrom=True,
pending_out=True,
groups=['Friends', 'Examples'])
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
Resolve timeout errors for get_roster. See issue #89 Using get_roster will now return the same types of values as Iq.send. If a timeout occurs, then the event 'roster_timeout' will be fired. A successful call to get_roster will also raise the 'roster_received' event. To ensure that the get_roster call was successful, here is a pattern to follow: def __init__(self, ...): ... self.add_event_handler('session_start', self.session_start) self.add_event_handler('roster_timeout', self.roster_timeout) self.add_event_handler('roster_received', self.roster_received) def session_start(self, e): self.send_presence() self.get_roster() def roster_timeout(self, e): # Optionally increase the timeout period self.get_roster(timeout=self.response_timeout * 2) def roster_received(self, iq): # Do stuff, roster has been initialized. ...
2011-05-20 16:56:00 +00:00
# Give the event queue time to process.
time.sleep(.1)
self.failUnless('roster_received' in events,
"Roster received event not triggered: %s" % events)
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
def testRosterSet(self):
"""Test handling pushed roster updates."""
self.stream_start(mode='client')
Resolve timeout errors for get_roster. See issue #89 Using get_roster will now return the same types of values as Iq.send. If a timeout occurs, then the event 'roster_timeout' will be fired. A successful call to get_roster will also raise the 'roster_received' event. To ensure that the get_roster call was successful, here is a pattern to follow: def __init__(self, ...): ... self.add_event_handler('session_start', self.session_start) self.add_event_handler('roster_timeout', self.roster_timeout) self.add_event_handler('roster_received', self.roster_received) def session_start(self, e): self.send_presence() self.get_roster() def roster_timeout(self, e): # Optionally increase the timeout period self.get_roster(timeout=self.response_timeout * 2) def roster_received(self, iq): # Do stuff, roster has been initialized. ...
2011-05-20 16:56:00 +00:00
events = []
def roster_update(e):
events.append('roster_update')
self.xmpp.add_event_handler('roster_update', roster_update)
Added stream tests for rosters.
2010-10-24 16:53:14 +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("""
First pass at integrating the new roster manager.
2010-10-27 03:47:17 +00:00
<iq to='tester@localhost' type="set" id="1">
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
<query xmlns="jabber:iq:roster">
<item jid="user@localhost"
name="User"
subscription="both">
<group>Friends</group>
<group>Examples</group>
</item>
</query>
</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.send("""
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
<iq type="result" id="1">
<query xmlns="jabber:iq:roster" />
</iq>
""")
First pass at integrating the new roster manager.
2010-10-27 03:47:17 +00:00
self.check_roster('tester@localhost', 'user@localhost',
name='User',
subscription='both',
groups=['Friends', 'Examples'])
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
Resolve timeout errors for get_roster. See issue #89 Using get_roster will now return the same types of values as Iq.send. If a timeout occurs, then the event 'roster_timeout' will be fired. A successful call to get_roster will also raise the 'roster_received' event. To ensure that the get_roster call was successful, here is a pattern to follow: def __init__(self, ...): ... self.add_event_handler('session_start', self.session_start) self.add_event_handler('roster_timeout', self.roster_timeout) self.add_event_handler('roster_received', self.roster_received) def session_start(self, e): self.send_presence() self.get_roster() def roster_timeout(self, e): # Optionally increase the timeout period self.get_roster(timeout=self.response_timeout * 2) def roster_received(self, iq): # Do stuff, roster has been initialized. ...
2011-05-20 16:56:00 +00:00
# Give the event queue time to process.
time.sleep(.1)
self.failUnless('roster_update' in events,
"Roster updated event not triggered: %s" % events)
def testRosterTimeout(self):
"""Test handling a timed out roster request."""
self.stream_start()
events = []
def roster_timeout(event):
events.append('roster_timeout')
self.xmpp.add_event_handler('roster_timeout', roster_timeout)
self.xmpp.get_roster(timeout=0)
# Give the event queue time to process.
time.sleep(.1)
self.failUnless(events == ['roster_timeout'],
"Roster timeout event not triggered: %s." % events)
def testRosterCallback(self):
"""Test handling a roster request callback."""
self.stream_start()
events = []
def roster_callback(iq):
events.append('roster_callback')
# Since get_roster blocks, we need to run it in a thread.
t = threading.Thread(name='get_roster',
target=self.xmpp.get_roster,
kwargs={'callback': roster_callback})
t.start()
self.send("""
<iq type="get" id="1">
<query xmlns="jabber:iq:roster" />
</iq>
""")
self.recv("""
<iq type="result" id="1">
<query xmlns="jabber:iq:roster">
<item jid="user@localhost"
name="User"
subscription="both">
<group>Friends</group>
<group>Examples</group>
</item>
</query>
</iq>
""")
# Wait for get_roster to return.
t.join()
# Give the event queue time to process.
time.sleep(.1)
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
Resolve timeout errors for get_roster. See issue #89 Using get_roster will now return the same types of values as Iq.send. If a timeout occurs, then the event 'roster_timeout' will be fired. A successful call to get_roster will also raise the 'roster_received' event. To ensure that the get_roster call was successful, here is a pattern to follow: def __init__(self, ...): ... self.add_event_handler('session_start', self.session_start) self.add_event_handler('roster_timeout', self.roster_timeout) self.add_event_handler('roster_received', self.roster_received) def session_start(self, e): self.send_presence() self.get_roster() def roster_timeout(self, e): # Optionally increase the timeout period self.get_roster(timeout=self.response_timeout * 2) def roster_received(self, iq): # Do stuff, roster has been initialized. ...
2011-05-20 16:56:00 +00:00
self.failUnless(events == ['roster_callback'],
"Roster timeout event not triggered: %s." % events)
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamRoster)
Reference in a new issue Copy permalink