source: wokkel/test/test_xmppim.py @ 101:d88462a49f86

Last change on this file since 101:d88462a49f86 was 96:8e6130587088, checked in by Ralph Meijer <ralphm@…>, 11 years ago

Remove copyright dates from individual source files, only update LICENSE.
File size: 16.2 KB
Line 
1# Copyright (c) Ralph Meijer.
2# See LICENSE for details
3
4"""
5Tests for L{wokkel.xmppim}.
6"""
7
8from twisted.internet import defer
9from twisted.trial import unittest
10from twisted.words.protocols.jabber.jid import JID
11from twisted.words.protocols.jabber.xmlstream import toResponse
12from twisted.words.xish import domish, utility
13
14from wokkel import xmppim
15from wokkel.generic import ErrorStanza, parseXml
16from wokkel.test.helpers import XmlStreamStub
17
18NS_XML = 'http://www.w3.org/XML/1998/namespace'
19NS_ROSTER = 'jabber:iq:roster'
20
21class PresenceClientProtocolTest(unittest.TestCase):
22    def setUp(self):
23        self.output = []
24        self.protocol = xmppim.PresenceClientProtocol()
25        self.protocol.parent = self
26
27    def send(self, obj):
28        self.output.append(obj)
29
30    def test_unavailableDirected(self):
31        """
32        Test sending of directed unavailable presence broadcast.
33        """
34
35        self.protocol.unavailable(JID('user@example.com'))
36        presence = self.output[-1]
37        self.assertEquals("presence", presence.name)
38        self.assertEquals(None, presence.uri)
39        self.assertEquals("user@example.com", presence.getAttribute('to'))
40        self.assertEquals("unavailable", presence.getAttribute('type'))
41
42    def test_unavailableWithStatus(self):
43        """
44        Test sending of directed unavailable presence broadcast with status.
45        """
46
47        self.protocol.unavailable(JID('user@example.com'),
48                                  {None: 'Disconnected'})
49        presence = self.output[-1]
50        self.assertEquals("presence", presence.name)
51        self.assertEquals(None, presence.uri)
52        self.assertEquals("user@example.com", presence.getAttribute('to'))
53        self.assertEquals("unavailable", presence.getAttribute('type'))
54        self.assertEquals("Disconnected", unicode(presence.status))
55
56    def test_unavailableBroadcast(self):
57        """
58        Test sending of unavailable presence broadcast.
59        """
60
61        self.protocol.unavailable(None)
62        presence = self.output[-1]
63        self.assertEquals("presence", presence.name)
64        self.assertEquals(None, presence.uri)
65        self.assertEquals(None, presence.getAttribute('to'))
66        self.assertEquals("unavailable", presence.getAttribute('type'))
67
68    def test_unavailableBroadcastNoEntityParameter(self):
69        """
70        Test sending of unavailable presence broadcast by not passing entity.
71        """
72
73        self.protocol.unavailable()
74        presence = self.output[-1]
75        self.assertEquals("presence", presence.name)
76        self.assertEquals(None, presence.uri)
77        self.assertEquals(None, presence.getAttribute('to'))
78        self.assertEquals("unavailable", presence.getAttribute('type'))
79
80
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
452class RosterClientProtocolTest(unittest.TestCase):
453    """
454    Tests for L{xmppim.RosterClientProtocol}.
455    """
456
457    def setUp(self):
458        self.stub = XmlStreamStub()
459        self.protocol = xmppim.RosterClientProtocol()
460        self.protocol.xmlstream = self.stub.xmlstream
461        self.protocol.connectionInitialized()
462
463
464    def test_removeItem(self):
465        """
466        Removing a roster item is setting an item with subscription C{remove}.
467        """
468        d = self.protocol.removeItem(JID('test@example.org'))
469
470        # Inspect outgoing iq request
471
472        iq = self.stub.output[-1]
473        self.assertEquals('set', iq.getAttribute('type'))
474        self.assertNotIdentical(None, iq.query)
475        self.assertEquals(NS_ROSTER, iq.query.uri)
476
477        children = list(domish.generateElementsQNamed(iq.query.children,
478                                                      'item', NS_ROSTER))
479        self.assertEquals(1, len(children))
480        child = children[0]
481        self.assertEquals('test@example.org', child['jid'])
482        self.assertEquals('remove', child['subscription'])
483
484        # Fake successful response
485
486        response = toResponse(iq, 'result')
487        self.stub.send(response)
488        return d
Note: See TracBrowser for help on using the repository browser.