# -*- encoding:utf-8 -*-
from __future__ import unicode_literals
import unittest
from slixmpp.exceptions import IqTimeout
from slixmpp.test import SlixTest
import time
import threading
class TestStreamRoster(SlixTest):
"""
Test handling roster updates.
"""
def tearDown(self):
self.stream_close()
def testGetRoster(self):
"""Test handling roster requests."""
self.stream_start(mode='client', jid='tester@localhost')
roster_updates = []
self.xmpp.add_event_handler('roster_update', roster_updates.append)
self.xmpp.get_roster()
self.send("""
""")
self.recv("""
-
Friends
Examples
""")
self.check_roster('tester@localhost', 'user@localhost',
name='User',
subscription='from',
afrom=True,
pending_out=True,
groups=['Friends', 'Examples'])
self.failUnless(len(roster_updates) == 1,
"Wrong number of roster_update events fired: %s (should be 1)" % len(roster_updates))
def testRosterSet(self):
"""Test handling pushed roster updates."""
self.stream_start(mode='client')
events = []
def roster_update(e):
events.append('roster_update')
self.xmpp.add_event_handler('roster_update', roster_update)
self.recv("""
-
Friends
Examples
""")
self.send("""
""")
self.check_roster('tester@localhost', 'user@localhost',
name='User',
subscription='both',
groups=['Friends', 'Examples'])
self.failUnless('roster_update' in events,
"Roster updated event not triggered: %s" % events)
def testRosterPushRemove(self):
"""Test handling roster item removal updates."""
self.stream_start(mode='client')
events = []
# Add roster item
self.recv("""
-
Friends
Examples
""")
self.send("""
""")
self.assertTrue('user@localhost' in self.xmpp.client_roster)
# Receive item remove push
self.recv("""
-
""")
self.send("""
""")
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("""
-
Friends
Examples
""")
self.send("""
""")
def testRosterCallback(self):
"""Test handling a roster request callback."""
self.stream_start()
events = []
def roster_callback(iq):
events.append('roster_callback')
self.xmpp.get_roster(callback=roster_callback)
self.send("""
""")
self.recv("""
-
Friends
Examples
""")
self.failUnless(events == ['roster_callback'],
"Roster timeout event not triggered: %s." % events)
def testRosterUnicode(self):
"""Test that JIDs with Unicode values are handled properly."""
self.stream_start(plugins=[])
self.recv("""
-
Unicode
""")
self.check_roster('tester@localhost', 'andré@foo',
subscription='both',
groups=['Unicode'])
jids = list(self.xmpp.client_roster.keys())
self.failUnless(jids == ['andré@foo'],
"Too many roster entries found: %s" % jids)
self.recv("""
away
Testing
""")
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)
def testSendLastPresence(self):
"""Test that sending the last presence works."""
self.stream_start(plugins=[])
self.xmpp.send_presence(pshow='dnd')
self.xmpp.auto_authorize = True
self.xmpp.auto_subscribe = True
self.send("""
dnd
""")
self.recv("""
""")
self.send("""
""")
self.send("""
dnd
""")
def testUnsupportedRosterVer(self):
"""Test working with a server without roster versioning."""
self.stream_start()
self.assertTrue('rosterver' not in self.xmpp.features)
self.xmpp.get_roster()
self.send("""
""")
self.recv("""
""")
def testBootstrapRosterVer(self):
"""Test bootstrapping with roster versioning."""
self.stream_start()
self.xmpp.features.add('rosterver')
self.xmpp.client_roster.version = ''
self.xmpp.get_roster()
self.send("""
""")
self.recv("""
""")
def testExistingRosterVer(self):
"""Test using a stored roster version."""
self.stream_start()
self.xmpp.features.add('rosterver')
self.xmpp.client_roster.version = '42'
self.xmpp.get_roster()
self.send("""
""")
self.recv("""
""")
suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamRoster)