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

219 lines
6.7 KiB
Python
Raw Normal View History

Fix double roster entry issue with Unicode. JIDs with Unicode values were being encoded by the JID class instead of leaving them as just Unicode strings. It may still be a good idea to use from __future__ import unicode_literals pretty much everywhere though. Fixes issue #88.
2011-05-20 20:48:13 +00:00
# -*- encoding:utf-8 -*-
from __future__ import unicode_literals
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."""
self.stream_start(mode='client')
self.failUnless(self.xmpp.roster == {}, "Initial roster not empty.")
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("""
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
<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()
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)
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
roster = {'user@localhost': {'name': 'User',
'subscription': 'both',
'groups': ['Friends', 'Examples'],
'presence': {},
'in_roster': True}}
self.failUnless(self.xmpp.roster == roster,
"Unexpected roster values: %s" % self.xmpp.roster)
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('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')
self.failUnless(self.xmpp.roster == {}, "Initial roster not empty.")
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("""
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
<iq type="set" id="1">
<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>
""")
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)
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
roster = {'user@localhost': {'name': 'User',
'subscription': 'both',
'groups': ['Friends', 'Examples'],
'presence': {},
'in_roster': True}}
self.failUnless(self.xmpp.roster == roster,
"Unexpected roster values: %s" % self.xmpp.roster)
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('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,
Fix test for get_roster(). Python2.6 has issues passing a Unicode string as a keyword name.
2011-05-25 22:52:42 +00:00
kwargs={str('callback'): roster_callback})
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
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
Fix double roster entry issue with Unicode. JIDs with Unicode values were being encoded by the JID class instead of leaving them as just Unicode strings. It may still be a good idea to use from __future__ import unicode_literals pretty much everywhere though. Fixes issue #88.
2011-05-20 20:48:13 +00:00
def testRosterUnicode(self):
"""Test that JIDs with Unicode values are handled properly."""
self.stream_start()
self.recv("""
<iq to="tester@localhost" type="set" id="1">
<query xmlns="jabber:iq:roster">
<item jid="andré@foo" subscription="both">
<group>Unicode</group>
</item>
</query>
</iq>
""")
# Give the event queue time to process.
time.sleep(.1)
roster = {'andré@foo': {
'name': '',
'subscription': 'both',
'groups': ['Unicode'],
'presence': {},
'in_roster': True}}
self.failUnless(self.xmpp.roster == roster,
"Unexpected roster values: %s" % self.xmpp.roster)
self.recv("""
Make roster test a little more robust.
2011-05-21 01:15:57 +00:00
<presence to="tester@localhost" from="andré@foo/bar">
<show>away</show>
<status>Testing</status>
</presence>
Fix double roster entry issue with Unicode. JIDs with Unicode values were being encoded by the JID class instead of leaving them as just Unicode strings. It may still be a good idea to use from __future__ import unicode_literals pretty much everywhere though. Fixes issue #88.
2011-05-20 20:48:13 +00:00
""")
# Give the event queue time to process.
time.sleep(.1)
roster = {'andré@foo': {
'name': '',
'subscription': 'both',
'groups': ['Unicode'],
'presence': {
'bar':{'priority':0,
Make roster test a little more robust.
2011-05-21 01:15:57 +00:00
'status':'Testing',
'show':'away'}},
Fix double roster entry issue with Unicode. JIDs with Unicode values were being encoded by the JID class instead of leaving them as just Unicode strings. It may still be a good idea to use from __future__ import unicode_literals pretty much everywhere though. Fixes issue #88.
2011-05-20 20:48:13 +00:00
'in_roster': True}}
self.failUnless(self.xmpp.roster == roster,
"Unexpected roster values: %s" % self.xmpp.roster)
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