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

354 lines
10 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
Save progress
2011-05-20 21:42:40 +00:00
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
Remove `roster_received` event
2013-03-28 17:05:37 +00:00
roster_updates = []
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
Remove `roster_received` event
2013-03-28 17:05:37 +00:00
self.xmpp.add_event_handler('roster_update', roster_updates.append)
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
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)
Remove `roster_received` event
2013-03-28 17:05:37 +00:00
self.failUnless(len(roster_updates) == 1,
"Wrong number of roster_update events fired: %s (should be 1)" % len(roster_updates))
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)
Add tests for roster versioning.
2012-03-08 00:11:59 +00:00
def testRosterPushRemove(self):
"""Test handling roster item removal updates."""
self.stream_start(mode='client')
events = []
# Add roster item
self.recv("""
<iq to='tester@localhost' 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>
""")
self.send("""
<iq type="result" id="1">
<query xmlns="jabber:iq:roster" />
</iq>
""")
self.assertTrue('user@localhost' in self.xmpp.client_roster)
# Receive item remove push
self.recv("""
<iq to='tester@localhost' type="set" id="1">
<query xmlns="jabber:iq:roster">
<item jid="user@localhost"
subscription="remove">
</item>
</query>
</iq>
""")
self.send("""
<iq type="result" id="1">
<query xmlns="jabber:iq:roster" />
</iq>
""")
self.assertTrue('user@localhost' not in self.xmpp.client_roster)
def testUnauthorizedRosterPush(self):
"""Test rejecting a roster push from an unauthorized source."""
self.stream_start()
self.recv("""
Remove `roster_received` event
2013-03-28 17:05:37 +00:00
<iq to='tester@localhost' from="malicious_user@localhost"
Add tests for roster versioning.
2012-03-08 00:11:59 +00:00
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>
""")
self.send("""
<iq to="malicious_user@localhost" type="error" id="1">
<error type="cancel" code="503">
<service-unavailable xmlns="urn:ietf:params:xml:ns:xmpp-stanzas" />
</error>
</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
def testRosterTimeout(self):
"""Test handling a timed out roster request."""
self.stream_start()
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
def do_test():
self.xmpp.get_roster(timeout=0)
time.sleep(.1)
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
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
self.assertRaises(IqTimeout, do_test)
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
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,
update_caps() can now do presence broadcasting. As part of adding this feature: - fixed bug in update_caps() not assigning verstrings - fixed xep_0004 typo - can now use None as a roster key which will map to boundjid.bare - fixed using JID objects in disco node handlers - fixed failing test related to get_roster Several of these bugs I've fixed before, so I either didn't push them earlier, or I clobbered something when merging. *shrug*
2012-01-12 00:39:55 +00:00
kwargs={str('block'): False,
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
Save progress
2011-05-20 21:42:40 +00:00
def testRosterUnicode(self):
"""Test that JIDs with Unicode values are handled properly."""
Resolve plugin dependency chains with XEP-0115. The post_init() system can only reliably handle a single layer of dependencies between plugins, but PEP plugins with XEP-0115 exceed that limit and plugins can be post_init'ed out of order. To resolve this, we will special case XEP-0115 to be post_init'ed first until the new plugin system with dependency tracking is stable.
2012-03-10 20:48:35 +00:00
self.stream_start(plugins=[])
Save progress
2011-05-20 21:42:40 +00:00
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)
Merge branch 'develop' into roster Conflicts: tests/test_stream_roster.py
2011-05-21 01:12:53 +00:00
self.check_roster('tester@localhost', 'andré@foo',
subscription='both',
groups=['Unicode'])
Fix test for Python3. Issue of dict_keys vs list data types.
2011-05-21 01:36:09 +00:00
jids = list(self.xmpp.client_roster.keys())
Merge branch 'develop' into roster Conflicts: tests/test_stream_roster.py
2011-05-21 01:12:53 +00:00
self.failUnless(jids == ['andré@foo'],
"Too many roster entries found: %s" % jids)
Save progress
2011-05-20 21:42:40 +00:00
self.recv("""
Merge branch 'develop' into roster Conflicts: tests/test_stream_roster.py
2011-05-21 01:12:53 +00:00
<presence to="tester@localhost" from="andré@foo/bar">
<show>away</show>
<status>Testing</status>
</presence>
Save progress
2011-05-20 21:42:40 +00:00
""")
# Give the event queue time to process.
time.sleep(.1)
Merge branch 'develop' into roster Conflicts: tests/test_stream_roster.py
2011-05-21 01:12:53 +00:00
result = self.xmpp.client_roster['andré@foo'].resources
expected = {'bar': {'status':'Testing',
'show':'away',
'priority':0}}
self.failUnless(result == expected,
"Unexpected roster values: %s" % result)
Save progress
2011-05-20 21:42:40 +00:00
Integrate roster with BaseXMPP. Last sent stanzas are saved regardless of if the roster is used directly or self.send_presence
2011-06-16 23:03:31 +00:00
def testSendLastPresence(self):
"""Test that sending the last presence works."""
Resolve plugin dependency chains with XEP-0115. The post_init() system can only reliably handle a single layer of dependencies between plugins, but PEP plugins with XEP-0115 exceed that limit and plugins can be post_init'ed out of order. To resolve this, we will special case XEP-0115 to be post_init'ed first until the new plugin system with dependency tracking is stable.
2012-03-10 20:48:35 +00:00
self.stream_start(plugins=[])
Integrate roster with BaseXMPP. Last sent stanzas are saved regardless of if the roster is used directly or self.send_presence
2011-06-16 23:03:31 +00:00
self.xmpp.send_presence(pshow='dnd')
self.xmpp.auto_authorize = True
self.xmpp.auto_subscribe = True
self.send("""
<presence>
<show>dnd</show>
</presence>
""")
self.recv("""
<presence from="user@localhost"
to="tester@localhost"
type="subscribe" />
""")
self.send("""
<presence to="user@localhost"
type="subscribed" />
""")
self.send("""
<presence to="user@localhost">
<show>dnd</show>
</presence>
""")
Add tests for roster versioning.
2012-03-08 00:11:59 +00:00
def testUnsupportedRosterVer(self):
"""Test working with a server without roster versioning."""
self.stream_start()
self.assertTrue('rosterver' not in self.xmpp.features)
t = threading.Thread(name='get_roster', target=self.xmpp.get_roster)
t.start()
self.send("""
<iq type="get" id="1">
<query xmlns="jabber:iq:roster" />
</iq>
""")
self.recv("""
<iq to="tester@localhost" type="result" id="1" />
""")
t.join()
def testBootstrapRosterVer(self):
"""Test bootstrapping with roster versioning."""
self.stream_start()
self.xmpp.features.add('rosterver')
self.xmpp.client_roster.version = ''
t = threading.Thread(name='get_roster', target=self.xmpp.get_roster)
t.start()
self.send("""
<iq type="get" id="1">
<query xmlns="jabber:iq:roster" ver="" />
</iq>
""")
self.recv("""
<iq to="tester@localhost" type="result" id="1" />
""")
t.join()
def testExistingRosterVer(self):
"""Test using a stored roster version."""
self.stream_start()
self.xmpp.features.add('rosterver')
self.xmpp.client_roster.version = '42'
t = threading.Thread(name='get_roster', target=self.xmpp.get_roster)
t.start()
self.send("""
<iq type="get" id="1">
<query xmlns="jabber:iq:roster" ver="42" />
</iq>
""")
self.recv("""
<iq to="tester@localhost" type="result" id="1" />
""")
t.join()
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