import time import unittest from slixmpp.test import SlixTest class TestStreamPresence(SlixTest): """ Test handling roster updates. """ def setUp(self): self.stream_start(jid='tester@localhost', plugins=[]) def tearDown(self): self.stream_close() def testInitialUnavailablePresences(self): """ Test receiving unavailable presences from JIDs that are not online. """ events = set() def got_offline(presence): # The got_offline event should not be triggered. events.add('got_offline') def unavailable(presence): # The presence_unavailable event should be triggered. events.add('unavailable') self.xmpp.add_event_handler('got_offline', got_offline) self.xmpp.add_event_handler('presence_unavailable', unavailable) self.recv(""" <presence type="unavailable" from="otheruser@localhost" to="tester@localhost"/> """) self.assertEqual(events, {'unavailable'}, "Got offline incorrectly triggered: %s." % events) def testGotOffline(self): """Test that got_offline is triggered properly.""" events = [] def got_offline(presence): events.append('got_offline') self.xmpp.add_event_handler('got_offline', got_offline) # Setup roster. Use a 'set' instead of 'result' so we # don't have to handle get_roster() blocking. # # We use the stream to initialize the roster to make # the test independent of the roster implementation. self.recv(""" <iq type="set"> <query xmlns="jabber:iq:roster"> <item jid="otheruser@localhost" name="Other User" subscription="both"> <group>Testers</group> </item> </query> </iq> """) # Contact comes online. self.recv(""" <presence from="otheruser@localhost/foobar" to="tester@localhost" /> """) # Contact goes offline, should trigger got_offline. self.recv(""" <presence from="otheruser@localhost/foobar" to="tester@localhost" type="unavailable" /> """) self.assertEqual(events, ['got_offline'], "Got offline incorrectly triggered: %s" % events) def testGotOnline(self): """Test that got_online is triggered properly.""" events = set() def presence_available(p): events.add('presence_available') def got_online(p): events.add('got_online') self.xmpp.add_event_handler('presence_available', presence_available) self.xmpp.add_event_handler('got_online', got_online) self.recv(""" <presence from="user@localhost" to="tester@localhost" /> """) expected = {'presence_available', 'got_online'} self.assertEqual(events, expected, "Incorrect events triggered: %s" % events) def testAutoAuthorizeAndSubscribe(self): """ Test auto authorizing and auto subscribing to subscription requests. """ events = set() def presence_subscribe(p): events.add('presence_subscribe') def changed_subscription(p): events.add('changed_subscription') self.xmpp.add_event_handler('changed_subscription', changed_subscription) self.xmpp.add_event_handler('presence_subscribe', presence_subscribe) # With these settings we should accept a subscription # and request a subscription in return. self.xmpp.auto_authorize = True self.xmpp.auto_subscribe = True 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" /> """) self.send(""" <presence to="user@localhost" type="subscribe" /> """) expected = {'presence_subscribe', 'changed_subscription'} self.assertEqual(events, expected, "Incorrect events triggered: %s" % events) def testNoAutoAuthorize(self): """Test auto rejecting subscription requests.""" events = set() def presence_subscribe(p): events.add('presence_subscribe') def changed_subscription(p): events.add('changed_subscription') self.xmpp.add_event_handler('changed_subscription', changed_subscription) self.xmpp.add_event_handler('presence_subscribe', presence_subscribe) # With this setting we should reject all subscriptions. self.xmpp.roster['tester@localhost'].auto_authorize = False self.recv(""" <presence from="user@localhost" to="tester@localhost" type="subscribe" /> """) self.send(""" <presence to="user@localhost" type="unsubscribed" /> """) expected = {'presence_subscribe', 'changed_subscription'} self.assertEqual(events, expected, "Incorrect events triggered: %s" % events) def test_presence_events(self): """Test that presence events are raised.""" events = [] ptypes = ['available', 'away', 'dnd', 'xa', 'chat', 'unavailable', 'subscribe', 'subscribed', 'unsubscribe', 'unsubscribed'] for ptype in ptypes: handler = lambda p: events.append(p['type']) self.xmpp.add_event_handler('presence_%s' % ptype, handler) self.recv(""" <presence /> """) self.recv(""" <presence><show>away</show></presence> """) self.recv(""" <presence><show>dnd</show></presence> """) self.recv(""" <presence><show>xa</show></presence> """) self.recv(""" <presence><show>chat</show></presence> """) self.recv(""" <presence type="unavailable" /> """) self.recv(""" <presence type="subscribe" /> """) self.recv(""" <presence type="subscribed" /> """) self.recv(""" <presence type="unsubscribe" /> """) self.recv(""" <presence type="unsubscribed" /> """) self.assertEqual(events, ptypes, "Not all events raised: %s" % events) def test_changed_status(self): """Test that the changed_status event is handled properly.""" events = [] def changed_status(presence): events.append(presence['type']) self.xmpp.add_event_handler('changed_status', changed_status) self.recv(""" <presence from="user@example.com" to="tester@localhost" /> """) self.recv(""" <presence from="user@example.com" to="tester@localhost" /> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>away</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>away</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>dnd</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>dnd</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>chat</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>chat</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>xa</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>xa</show> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost" type="unavailable" /> """) self.recv(""" <presence from="user@example.com" to="tester@localhost" type="unavailable" /> """) self.recv(""" <presence from="user@example.com" to="tester@localhost" /> """) self.recv(""" <presence from="user@example.com" to="tester@localhost" /> """) self.recv(""" <presence from="user@example.com" to="tester@localhost" /> """) # Changed status text, so fire new event self.recv(""" <presence from="user@example.com" to="tester@localhost"> <status>Testing!</status> </presence> """) # No change in show/status values, no event self.recv(""" <presence from="user@example.com" to="tester@localhost"> <status>Testing!</status> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>dnd</show> <status>Testing!</status> </presence> """) self.recv(""" <presence from="user@example.com" to="tester@localhost"> <show>dnd</show> <status>Testing!</status> </presence> """) self.assertEqual(events, ['available', 'away', 'dnd', 'chat', 'xa', 'unavailable', 'available', 'available', 'dnd'], "Changed status events incorrect: %s" % events) suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamPresence)