Ignore:
Timestamp:
Jul 17, 2009, 3:19:40 PM (13 years ago)
Author:
Ralph Meijer <ralphm@…>
Branch:
default
Convert:
svn:b33ecbfc-034c-dc11-8662-000475d9059e/trunk@186
Message:

Implement a better presence protocol handler PresenceProtocol?.

This handler is not client-only, like PresenceClientProtocol?, and accepts
sender arguments to all methods for outgoing presence. The *Received
methods now expect a single argument that is a parsed representation of the
presence stanza.

Also included is a simple XMPP server example that connects via the
server-to-server protocol, accepting presence requests and echoing incoming
messages for all potential addresses at a domain.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • wokkel/test/test_xmppim.py

    r28 r68  
    1 # Copyright (c) 2003-2008 Ralph Meijer
     1# Copyright (c) 2003-2009 Ralph Meijer
    22# See LICENSE for details
    33
     
    66"""
    77
     8from twisted.internet import defer
    89from twisted.trial import unittest
    910from twisted.words.protocols.jabber.jid import JID
    1011from twisted.words.protocols.jabber.xmlstream import toResponse
    11 from twisted.words.xish import domish
     12from twisted.words.xish import domish, utility
    1213
    1314from wokkel import xmppim
     15from wokkel.generic import ErrorStanza, parseXml
    1416from wokkel.test.helpers import XmlStreamStub
    1517
     18NS_XML = 'http://www.w3.org/XML/1998/namespace'
    1619NS_ROSTER = 'jabber:iq:roster'
    1720
     
    7679
    7780
     81
     82class AvailabilityPresenceTest(unittest.TestCase):
     83
     84    def test_fromElement(self):
     85        xml = """<presence from='user@example.org' to='user@example.com'>
     86                   <show>chat</show>
     87                   <status>Let's chat!</status>
     88                   <priority>50</priority>
     89                 </presence>
     90              """
     91
     92        presence = xmppim.AvailabilityPresence.fromElement(parseXml(xml))
     93        self.assertEquals(JID('user@example.org'), presence.sender)
     94        self.assertEquals(JID('user@example.com'), presence.recipient)
     95        self.assertTrue(presence.available)
     96        self.assertEquals('chat', presence.show)
     97        self.assertEquals({None: "Let's chat!"}, presence.statuses)
     98        self.assertEquals(50, presence.priority)
     99
     100
     101class PresenceProtocolTest(unittest.TestCase):
     102    """
     103    Tests for L{xmppim.PresenceProtocol}
     104    """
     105
     106    def setUp(self):
     107        self.output = []
     108        self.protocol = xmppim.PresenceProtocol()
     109        self.protocol.parent = self
     110        self.protocol.xmlstream = utility.EventDispatcher()
     111        self.protocol.connectionInitialized()
     112
     113
     114    def send(self, obj):
     115        self.output.append(obj)
     116
     117
     118    def test_errorReceived(self):
     119        """
     120        Incoming presence stanzas are parsed and dispatched.
     121        """
     122        xml = """<presence type="error"/>"""
     123
     124        def errorReceived(error):
     125            xmppim.PresenceProtocol.errorReceived(self.protocol, error)
     126            try:
     127                self.assertIsInstance(error, ErrorStanza)
     128            except:
     129                d.errback()
     130            else:
     131                d.callback(None)
     132
     133        d = defer.Deferred()
     134        self.protocol.errorReceived = errorReceived
     135        self.protocol.xmlstream.dispatch(parseXml(xml))
     136        return d
     137
     138
     139    def test_availableReceived(self):
     140        """
     141        Incoming presence stanzas are parsed and dispatched.
     142        """
     143        xml = """<presence/>"""
     144
     145        def availableReceived(presence):
     146            xmppim.PresenceProtocol.availableReceived(self.protocol, presence)
     147            try:
     148                self.assertIsInstance(presence, xmppim.AvailabilityPresence)
     149            except:
     150                d.errback()
     151            else:
     152                d.callback(None)
     153
     154        d = defer.Deferred()
     155        self.protocol.availableReceived = availableReceived
     156        self.protocol.xmlstream.dispatch(parseXml(xml))
     157        return d
     158
     159
     160    def test_unavailableReceived(self):
     161        """
     162        Incoming presence stanzas are parsed and dispatched.
     163        """
     164        xml = """<presence type='unavailable'/>"""
     165
     166        def unavailableReceived(presence):
     167            xmppim.PresenceProtocol.unavailableReceived(self.protocol, presence)
     168            try:
     169                self.assertIsInstance(presence, xmppim.AvailabilityPresence)
     170            except:
     171                d.errback()
     172            else:
     173                d.callback(None)
     174
     175        d = defer.Deferred()
     176        self.protocol.unavailableReceived = unavailableReceived
     177        self.protocol.xmlstream.dispatch(parseXml(xml))
     178        return d
     179
     180
     181    def test_subscribeReceived(self):
     182        """
     183        Incoming presence stanzas are parsed and dispatched.
     184        """
     185        xml = """<presence type='subscribe'/>"""
     186
     187        def subscribeReceived(presence):
     188            xmppim.PresenceProtocol.subscribeReceived(self.protocol, presence)
     189            try:
     190                self.assertIsInstance(presence, xmppim.SubscriptionPresence)
     191            except:
     192                d.errback()
     193            else:
     194                d.callback(None)
     195
     196        d = defer.Deferred()
     197        self.protocol.subscribeReceived = subscribeReceived
     198        self.protocol.xmlstream.dispatch(parseXml(xml))
     199        return d
     200
     201
     202    def test_unsubscribeReceived(self):
     203        """
     204        Incoming presence stanzas are parsed and dispatched.
     205        """
     206        xml = """<presence type='unsubscribe'/>"""
     207
     208        def unsubscribeReceived(presence):
     209            xmppim.PresenceProtocol.unsubscribeReceived(self.protocol, presence)
     210            try:
     211                self.assertIsInstance(presence, xmppim.SubscriptionPresence)
     212            except:
     213                d.errback()
     214            else:
     215                d.callback(None)
     216
     217        d = defer.Deferred()
     218        self.protocol.unsubscribeReceived = unsubscribeReceived
     219        self.protocol.xmlstream.dispatch(parseXml(xml))
     220        return d
     221
     222
     223    def test_subscribedReceived(self):
     224        """
     225        Incoming presence stanzas are parsed and dispatched.
     226        """
     227        xml = """<presence type='subscribed'/>"""
     228
     229        def subscribedReceived(presence):
     230            xmppim.PresenceProtocol.subscribedReceived(self.protocol, presence)
     231            try:
     232                self.assertIsInstance(presence, xmppim.SubscriptionPresence)
     233            except:
     234                d.errback()
     235            else:
     236                d.callback(None)
     237
     238        d = defer.Deferred()
     239        self.protocol.subscribedReceived = subscribedReceived
     240        self.protocol.xmlstream.dispatch(parseXml(xml))
     241        return d
     242
     243
     244    def test_unsubscribedReceived(self):
     245        """
     246        Incoming presence stanzas are parsed and dispatched.
     247        """
     248        xml = """<presence type='unsubscribed'/>"""
     249
     250        def unsubscribedReceived(presence):
     251            xmppim.PresenceProtocol.unsubscribedReceived(self.protocol,
     252                                                         presence)
     253            try:
     254                self.assertIsInstance(presence, xmppim.SubscriptionPresence)
     255            except:
     256                d.errback()
     257            else:
     258                d.callback(None)
     259
     260        d = defer.Deferred()
     261        self.protocol.unsubscribedReceived = unsubscribedReceived
     262        self.protocol.xmlstream.dispatch(parseXml(xml))
     263        return d
     264
     265
     266    def test_probeReceived(self):
     267        """
     268        Incoming presence stanzas are parsed and dispatched.
     269        """
     270        xml = """<presence type='probe'/>"""
     271
     272        def probeReceived(presence):
     273            xmppim.PresenceProtocol.probeReceived(self.protocol, presence)
     274            try:
     275                self.assertIsInstance(presence, xmppim.ProbePresence)
     276            except:
     277                d.errback()
     278            else:
     279                d.callback(None)
     280
     281        d = defer.Deferred()
     282        self.protocol.probeReceived = probeReceived
     283        self.protocol.xmlstream.dispatch(parseXml(xml))
     284        return d
     285
     286    def test_available(self):
     287        """
     288        It should be possible to pass a sender address.
     289        """
     290        self.protocol.available(JID('user@example.com'),
     291                                show=u'chat',
     292                                status=u'Talk to me!',
     293                                priority=50)
     294        element = self.output[-1]
     295        self.assertEquals("user@example.com", element.getAttribute('to'))
     296        self.assertIdentical(None, element.getAttribute('type'))
     297        self.assertEquals(u'chat', unicode(element.show))
     298        self.assertEquals(u'Talk to me!', unicode(element.status))
     299        self.assertEquals(u'50', unicode(element.priority))
     300
     301    def test_availableLanguages(self):
     302        """
     303        It should be possible to pass a sender address.
     304        """
     305        self.protocol.available(JID('user@example.com'),
     306                                show=u'chat',
     307                                statuses={None: u'Talk to me!',
     308                                          'nl': u'Praat met me!'},
     309                                priority=50)
     310        element = self.output[-1]
     311        self.assertEquals("user@example.com", element.getAttribute('to'))
     312        self.assertIdentical(None, element.getAttribute('type'))
     313        self.assertEquals(u'chat', unicode(element.show))
     314
     315        statuses = {}
     316        for status in element.elements():
     317            if status.name == 'status':
     318                lang = status.getAttribute((NS_XML, 'lang'))
     319                statuses[lang] = unicode(status)
     320
     321        self.assertIn(None, statuses)
     322        self.assertEquals(u'Talk to me!', statuses[None])
     323        self.assertIn('nl', statuses)
     324        self.assertEquals(u'Praat met me!', statuses['nl'])
     325        self.assertEquals(u'50', unicode(element.priority))
     326
     327
     328    def test_availableSender(self):
     329        """
     330        It should be possible to pass a sender address.
     331        """
     332        self.protocol.available(JID('user@example.com'),
     333                                sender=JID('user@example.org'))
     334        element = self.output[-1]
     335        self.assertEquals("user@example.org", element.getAttribute('from'))
     336
     337
     338    def test_unavailableDirected(self):
     339        """
     340        Test sending of directed unavailable presence broadcast.
     341        """
     342
     343        self.protocol.unavailable(JID('user@example.com'))
     344        element = self.output[-1]
     345        self.assertEquals("presence", element.name)
     346        self.assertEquals(None, element.uri)
     347        self.assertEquals("user@example.com", element.getAttribute('to'))
     348        self.assertEquals("unavailable", element.getAttribute('type'))
     349
     350    def test_unavailableWithStatus(self):
     351        """
     352        Test sending of directed unavailable presence broadcast with status.
     353        """
     354
     355        self.protocol.unavailable(JID('user@example.com'),
     356                                  {None: 'Disconnected'})
     357        element = self.output[-1]
     358        self.assertEquals("presence", element.name)
     359        self.assertEquals(None, element.uri)
     360        self.assertEquals("user@example.com", element.getAttribute('to'))
     361        self.assertEquals("unavailable", element.getAttribute('type'))
     362        self.assertEquals("Disconnected", unicode(element.status))
     363
     364
     365    def test_unavailableBroadcast(self):
     366        """
     367        Test sending of unavailable presence broadcast.
     368        """
     369
     370        self.protocol.unavailable(None)
     371        element = self.output[-1]
     372        self.assertEquals("presence", element.name)
     373        self.assertEquals(None, element.uri)
     374        self.assertEquals(None, element.getAttribute('to'))
     375        self.assertEquals("unavailable", element.getAttribute('type'))
     376
     377
     378    def test_unavailableBroadcastNoRecipientParameter(self):
     379        """
     380        Test sending of unavailable presence broadcast by not passing entity.
     381        """
     382
     383        self.protocol.unavailable()
     384        element = self.output[-1]
     385        self.assertEquals("presence", element.name)
     386        self.assertEquals(None, element.uri)
     387        self.assertEquals(None, element.getAttribute('to'))
     388        self.assertEquals("unavailable", element.getAttribute('type'))
     389
     390
     391    def test_unavailableSender(self):
     392        """
     393        It should be possible to pass a sender address.
     394        """
     395        self.protocol.unavailable(JID('user@example.com'),
     396                                  sender=JID('user@example.org'))
     397        element = self.output[-1]
     398        self.assertEquals("user@example.org", element.getAttribute('from'))
     399
     400
     401    def test_subscribeSender(self):
     402        """
     403        It should be possible to pass a sender address.
     404        """
     405        self.protocol.subscribe(JID('user@example.com'),
     406                                sender=JID('user@example.org'))
     407        element = self.output[-1]
     408        self.assertEquals("user@example.org", element.getAttribute('from'))
     409
     410
     411    def test_unsubscribeSender(self):
     412        """
     413        It should be possible to pass a sender address.
     414        """
     415        self.protocol.unsubscribe(JID('user@example.com'),
     416                                  sender=JID('user@example.org'))
     417        element = self.output[-1]
     418        self.assertEquals("user@example.org", element.getAttribute('from'))
     419
     420
     421    def test_subscribedSender(self):
     422        """
     423        It should be possible to pass a sender address.
     424        """
     425        self.protocol.subscribed(JID('user@example.com'),
     426                                 sender=JID('user@example.org'))
     427        element = self.output[-1]
     428        self.assertEquals("user@example.org", element.getAttribute('from'))
     429
     430
     431    def test_unsubscribedSender(self):
     432        """
     433        It should be possible to pass a sender address.
     434        """
     435        self.protocol.unsubscribed(JID('user@example.com'),
     436                                   sender=JID('user@example.org'))
     437        element = self.output[-1]
     438        self.assertEquals("user@example.org", element.getAttribute('from'))
     439
     440
     441    def test_probeSender(self):
     442        """
     443        It should be possible to pass a sender address.
     444        """
     445        self.protocol.probe(JID('user@example.com'),
     446                            sender=JID('user@example.org'))
     447        element = self.output[-1]
     448        self.assertEquals("user@example.org", element.getAttribute('from'))
     449
     450
     451
    78452class RosterClientProtocolTest(unittest.TestCase):
    79453    """
Note: See TracChangeset for help on using the changeset viewer.