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

312 lines
9.1 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
refactor: no import * in tests
2013-07-26 11:02:26 +00:00
import unittest
Rename to slixmpp
2014-07-17 12:19:04 +00:00
from slixmpp.exceptions import IqTimeout
from slixmpp.test import SlixTest
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
import time
import threading
Rename to slixmpp
2014-07-17 12:19:04 +00:00
class TestStreamRoster(SlixTest):
Added stream tests for rosters.
2010-10-24 16:53:14 +00:00
"""
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
Update the test suite. - monkey-patch our own monkey-patched idle_call to run events immediatly rather than adding them to the event queue, and add a fake transport with a fake socket. - remove the test file related to xep_0059 as it relies on blocking behavior, and comment out one xep_0030 test uses xep_0059 - remove many instances of threading and sleep()s because they do nothing except waste time and introduce race conditions. - keep exactly two sleep() in IoT xeps because they rely on timeouts
2015-02-12 11:23:47 +00:00
self.xmpp.get_roster()
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.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>
""")
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
fixing deprecation warnings for pytest
2018-10-08 21:25:23 +00:00
self.assertTrue(len(roster_updates) == 1,
Remove `roster_received` event
2013-03-28 17:05:37 +00:00
"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
fixing deprecation warnings for pytest
2018-10-08 21:25:23 +00:00
self.assertTrue('roster_update' in events,
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
"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 testRosterCallback(self):
"""Test handling a roster request callback."""
self.stream_start()
events = []
def roster_callback(iq):
events.append('roster_callback')
Update the test suite. - monkey-patch our own monkey-patched idle_call to run events immediatly rather than adding them to the event queue, and add a fake transport with a fake socket. - remove the test file related to xep_0059 as it relies on blocking behavior, and comment out one xep_0030 test uses xep_0059 - remove many instances of threading and sleep()s because they do nothing except waste time and introduce race conditions. - keep exactly two sleep() in IoT xeps because they rely on timeouts
2015-02-12 11:23:47 +00:00
self.xmpp.get_roster(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
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>
""")
fixing deprecation warnings for pytest
2018-10-08 21:25:23 +00:00
self.assertTrue(events == ['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
"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>
""")
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())
fixing deprecation warnings for pytest
2018-10-08 21:25:23 +00:00
self.assertTrue(jids == ['andré@foo'],
Merge branch 'develop' into roster Conflicts: tests/test_stream_roster.py
2011-05-21 01:12:53 +00:00
"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
""")
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}}
fixing deprecation warnings for pytest
2018-10-08 21:25:23 +00:00
self.assertTrue(result == expected,
Merge branch 'develop' into roster Conflicts: tests/test_stream_roster.py
2011-05-21 01:12:53 +00:00
"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)
Update the test suite. - monkey-patch our own monkey-patched idle_call to run events immediatly rather than adding them to the event queue, and add a fake transport with a fake socket. - remove the test file related to xep_0059 as it relies on blocking behavior, and comment out one xep_0030 test uses xep_0059 - remove many instances of threading and sleep()s because they do nothing except waste time and introduce race conditions. - keep exactly two sleep() in IoT xeps because they rely on timeouts
2015-02-12 11:23:47 +00:00
self.xmpp.get_roster()
Add tests for roster versioning.
2012-03-08 00:11:59 +00:00
self.send("""
<iq type="get" id="1">
<query xmlns="jabber:iq:roster" />
</iq>
""")
self.recv("""
<iq to="tester@localhost" type="result" id="1" />
""")
def testBootstrapRosterVer(self):
"""Test bootstrapping with roster versioning."""
self.stream_start()
self.xmpp.features.add('rosterver')
self.xmpp.client_roster.version = ''
Update the test suite. - monkey-patch our own monkey-patched idle_call to run events immediatly rather than adding them to the event queue, and add a fake transport with a fake socket. - remove the test file related to xep_0059 as it relies on blocking behavior, and comment out one xep_0030 test uses xep_0059 - remove many instances of threading and sleep()s because they do nothing except waste time and introduce race conditions. - keep exactly two sleep() in IoT xeps because they rely on timeouts
2015-02-12 11:23:47 +00:00
self.xmpp.get_roster()
Add tests for roster versioning.
2012-03-08 00:11:59 +00:00
self.send("""
<iq type="get" id="1">
<query xmlns="jabber:iq:roster" ver="" />
</iq>
""")
self.recv("""
<iq to="tester@localhost" type="result" id="1" />
""")
def testExistingRosterVer(self):
"""Test using a stored roster version."""
self.stream_start()
self.xmpp.features.add('rosterver')
self.xmpp.client_roster.version = '42'
Update the test suite. - monkey-patch our own monkey-patched idle_call to run events immediatly rather than adding them to the event queue, and add a fake transport with a fake socket. - remove the test file related to xep_0059 as it relies on blocking behavior, and comment out one xep_0030 test uses xep_0059 - remove many instances of threading and sleep()s because they do nothing except waste time and introduce race conditions. - keep exactly two sleep() in IoT xeps because they rely on timeouts
2015-02-12 11:23:47 +00:00
self.xmpp.get_roster()
Add tests for roster versioning.
2012-03-08 00:11:59 +00:00
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" />
""")
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