source: wokkel/test/test_pubsub.py @ 19:8df7118d8b93

Last change on this file since 19:8df7118d8b93 was 19:8df7118d8b93, checked in by Ralph Meijer <ralphm@…>, 15 years ago

Clean up and release Wokkel 0.3.0.
File size: 16.6 KB
Line 
1# Copyright (c) 2003-2008 Ralph Meijer
2# See LICENSE for details.
3
4"""
5Tests for L{wokkel.pubsub}
6"""
7
8from twisted.trial import unittest
9from twisted.internet import defer
10from twisted.words.xish import domish
11from twisted.words.protocols.jabber import error
12from twisted.words.protocols.jabber.jid import JID
13
14from wokkel import pubsub
15from wokkel.test.helpers import XmlStreamStub
16
17try:
18    from twisted.words.protocols.jabber.xmlstream import toResponse
19except ImportError:
20    from wokkel.compat import toResponse
21
22NS_PUBSUB = 'http://jabber.org/protocol/pubsub'
23NS_PUBSUB_ERRORS = 'http://jabber.org/protocol/pubsub#errors'
24NS_PUBSUB_EVENT = 'http://jabber.org/protocol/pubsub#event'
25
26def calledAsync(fn):
27    """
28    Function wrapper that fires a deferred upon calling the given function.
29    """
30    d = defer.Deferred()
31
32    def func(*args, **kwargs):
33        try:
34            result = fn(*args, **kwargs)
35        except:
36            d.errback()
37        else:
38            d.callback(result)
39
40    return d, func
41
42
43class PubSubClientTest(unittest.TestCase):
44    timeout = 2
45
46    def setUp(self):
47        self.stub = XmlStreamStub()
48        self.protocol = pubsub.PubSubClient()
49        self.protocol.xmlstream = self.stub.xmlstream
50        self.protocol.connectionInitialized()
51
52
53    def test_event_items(self):
54        """
55        Test receiving an items event resulting in a call to itemsReceived.
56        """
57        message = domish.Element((None, 'message'))
58        message['from'] = 'pubsub.example.org'
59        message['to'] = 'user@example.org/home'
60        event = message.addElement((NS_PUBSUB_EVENT, 'event'))
61        items = event.addElement('items')
62        items['node'] = 'test'
63        item1 = items.addElement('item')
64        item1['id'] = 'item1'
65        item2 = items.addElement('retract')
66        item2['id'] = 'item2'
67        item3 = items.addElement('item')
68        item3['id'] = 'item3'
69
70        def itemsReceived(recipient, service, nodeIdentifier, items):
71            self.assertEquals(JID('user@example.org/home'), recipient)
72            self.assertEquals(JID('pubsub.example.org'), service)
73            self.assertEquals('test', nodeIdentifier)
74            self.assertEquals([item1, item2, item3], items)
75
76        d, self.protocol.itemsReceived = calledAsync(itemsReceived)
77        self.stub.send(message)
78        return d
79
80
81    def test_event_delete(self):
82        """
83        Test receiving a delete event resulting in a call to deleteReceived.
84        """
85        message = domish.Element((None, 'message'))
86        message['from'] = 'pubsub.example.org'
87        message['to'] = 'user@example.org/home'
88        event = message.addElement((NS_PUBSUB_EVENT, 'event'))
89        items = event.addElement('delete')
90        items['node'] = 'test'
91
92        def deleteReceived(recipient, service, nodeIdentifier):
93            self.assertEquals(JID('user@example.org/home'), recipient)
94            self.assertEquals(JID('pubsub.example.org'), service)
95            self.assertEquals('test', nodeIdentifier)
96
97        d, self.protocol.deleteReceived = calledAsync(deleteReceived)
98        self.stub.send(message)
99        return d
100
101
102    def test_event_purge(self):
103        """
104        Test receiving a purge event resulting in a call to purgeReceived.
105        """
106        message = domish.Element((None, 'message'))
107        message['from'] = 'pubsub.example.org'
108        message['to'] = 'user@example.org/home'
109        event = message.addElement((NS_PUBSUB_EVENT, 'event'))
110        items = event.addElement('purge')
111        items['node'] = 'test'
112
113        def purgeReceived(recipient, service, nodeIdentifier):
114            self.assertEquals(JID('user@example.org/home'), recipient)
115            self.assertEquals(JID('pubsub.example.org'), service)
116            self.assertEquals('test', nodeIdentifier)
117
118        d, self.protocol.purgeReceived = calledAsync(purgeReceived)
119        self.stub.send(message)
120        return d
121
122
123    def test_createNode(self):
124        """
125        Test sending create request.
126        """
127
128        def cb(nodeIdentifier):
129            self.assertEquals('test', nodeIdentifier)
130
131        d = self.protocol.createNode(JID('pubsub.example.org'), 'test')
132        d.addCallback(cb)
133
134        iq = self.stub.output[-1]
135        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
136        self.assertEquals('set', iq.getAttribute('type'))
137        self.assertEquals('pubsub', iq.pubsub.name)
138        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
139        children = list(domish.generateElementsQNamed(iq.pubsub.children,
140                                                      'create', NS_PUBSUB))
141        self.assertEquals(1, len(children))
142        child = children[0]
143        self.assertEquals('test', child['node'])
144
145        response = toResponse(iq, 'result')
146        self.stub.send(response)
147        return d
148
149
150    def test_createNodeInstant(self):
151        """
152        Test sending create request resulting in an instant node.
153        """
154
155        def cb(nodeIdentifier):
156            self.assertEquals('test', nodeIdentifier)
157
158        d = self.protocol.createNode(JID('pubsub.example.org'))
159        d.addCallback(cb)
160
161        iq = self.stub.output[-1]
162        children = list(domish.generateElementsQNamed(iq.pubsub.children,
163                                                      'create', NS_PUBSUB))
164        child = children[0]
165        self.assertFalse(child.hasAttribute('node'))
166
167        response = toResponse(iq, 'result')
168        command = response.addElement((NS_PUBSUB, 'pubsub'))
169        create = command.addElement('create')
170        create['node'] = 'test'
171        self.stub.send(response)
172        return d
173
174
175    def test_createNodeRenamed(self):
176        """
177        Test sending create request resulting in renamed node.
178        """
179
180        def cb(nodeIdentifier):
181            self.assertEquals('test2', nodeIdentifier)
182
183        d = self.protocol.createNode(JID('pubsub.example.org'), 'test')
184        d.addCallback(cb)
185
186        iq = self.stub.output[-1]
187        children = list(domish.generateElementsQNamed(iq.pubsub.children,
188                                                      'create', NS_PUBSUB))
189        child = children[0]
190        self.assertEquals('test', child['node'])
191
192        response = toResponse(iq, 'result')
193        command = response.addElement((NS_PUBSUB, 'pubsub'))
194        create = command.addElement('create')
195        create['node'] = 'test2'
196        self.stub.send(response)
197        return d
198
199
200    def test_deleteNode(self):
201        """
202        Test sending delete request.
203        """
204
205        d = self.protocol.deleteNode(JID('pubsub.example.org'), 'test')
206
207        iq = self.stub.output[-1]
208        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
209        self.assertEquals('set', iq.getAttribute('type'))
210        self.assertEquals('pubsub', iq.pubsub.name)
211        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
212        children = list(domish.generateElementsQNamed(iq.pubsub.children,
213                                                      'delete', NS_PUBSUB))
214        self.assertEquals(1, len(children))
215        child = children[0]
216        self.assertEquals('test', child['node'])
217
218        response = toResponse(iq, 'result')
219        self.stub.send(response)
220        return d
221
222
223    def test_publish(self):
224        """
225        Test sending publish request.
226        """
227
228        item = pubsub.Item()
229        d = self.protocol.publish(JID('pubsub.example.org'), 'test', [item])
230
231        iq = self.stub.output[-1]
232        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
233        self.assertEquals('set', iq.getAttribute('type'))
234        self.assertEquals('pubsub', iq.pubsub.name)
235        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
236        children = list(domish.generateElementsQNamed(iq.pubsub.children,
237                                                      'publish', NS_PUBSUB))
238        self.assertEquals(1, len(children))
239        child = children[0]
240        self.assertEquals('test', child['node'])
241        items = list(domish.generateElementsQNamed(child.children,
242                                                   'item', NS_PUBSUB))
243        self.assertEquals(1, len(items))
244        self.assertIdentical(item, items[0])
245
246        response = toResponse(iq, 'result')
247        self.stub.send(response)
248        return d
249
250
251    def test_publishNoItems(self):
252        """
253        Test sending publish request without items.
254        """
255
256        d = self.protocol.publish(JID('pubsub.example.org'), 'test')
257
258        iq = self.stub.output[-1]
259        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
260        self.assertEquals('set', iq.getAttribute('type'))
261        self.assertEquals('pubsub', iq.pubsub.name)
262        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
263        children = list(domish.generateElementsQNamed(iq.pubsub.children,
264                                                      'publish', NS_PUBSUB))
265        self.assertEquals(1, len(children))
266        child = children[0]
267        self.assertEquals('test', child['node'])
268
269        response = toResponse(iq, 'result')
270        self.stub.send(response)
271        return d
272
273
274    def test_subscribe(self):
275        """
276        Test sending subscription request.
277        """
278        d = self.protocol.subscribe(JID('pubsub.example.org'), 'test',
279                                      JID('user@example.org'))
280
281        iq = self.stub.output[-1]
282        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
283        self.assertEquals('set', iq.getAttribute('type'))
284        self.assertEquals('pubsub', iq.pubsub.name)
285        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
286        children = list(domish.generateElementsQNamed(iq.pubsub.children,
287                                                      'subscribe', NS_PUBSUB))
288        self.assertEquals(1, len(children))
289        child = children[0]
290        self.assertEquals('test', child['node'])
291        self.assertEquals('user@example.org', child['jid'])
292
293        response = toResponse(iq, 'result')
294        pubsub = response.addElement((NS_PUBSUB, 'pubsub'))
295        subscription = pubsub.addElement('subscription')
296        subscription['node'] = 'test'
297        subscription['jid'] = 'user@example.org'
298        subscription['subscription'] = 'subscribed'
299        self.stub.send(response)
300        return d
301
302
303    def test_subscribePending(self):
304        """
305        Test sending subscription request that results in a pending
306        subscription.
307        """
308        d = self.protocol.subscribe(JID('pubsub.example.org'), 'test',
309                                      JID('user@example.org'))
310
311        iq = self.stub.output[-1]
312        response = toResponse(iq, 'result')
313        command = response.addElement((NS_PUBSUB, 'pubsub'))
314        subscription = command.addElement('subscription')
315        subscription['node'] = 'test'
316        subscription['jid'] = 'user@example.org'
317        subscription['subscription'] = 'pending'
318        self.stub.send(response)
319        self.assertFailure(d, pubsub.SubscriptionPending)
320        return d
321
322
323    def test_subscribeUnconfigured(self):
324        """
325        Test sending subscription request that results in an unconfigured
326        subscription.
327        """
328        d = self.protocol.subscribe(JID('pubsub.example.org'), 'test',
329                                      JID('user@example.org'))
330
331        iq = self.stub.output[-1]
332        response = toResponse(iq, 'result')
333        command = response.addElement((NS_PUBSUB, 'pubsub'))
334        subscription = command.addElement('subscription')
335        subscription['node'] = 'test'
336        subscription['jid'] = 'user@example.org'
337        subscription['subscription'] = 'unconfigured'
338        self.stub.send(response)
339        self.assertFailure(d, pubsub.SubscriptionUnconfigured)
340        return d
341
342
343    def test_unsubscribe(self):
344        """
345        Test sending unsubscription request.
346        """
347        d = self.protocol.unsubscribe(JID('pubsub.example.org'), 'test',
348                                      JID('user@example.org'))
349
350        iq = self.stub.output[-1]
351        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
352        self.assertEquals('set', iq.getAttribute('type'))
353        self.assertEquals('pubsub', iq.pubsub.name)
354        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
355        children = list(domish.generateElementsQNamed(iq.pubsub.children,
356                                                      'unsubscribe', NS_PUBSUB))
357        self.assertEquals(1, len(children))
358        child = children[0]
359        self.assertEquals('test', child['node'])
360        self.assertEquals('user@example.org', child['jid'])
361
362        self.stub.send(toResponse(iq, 'result'))
363        return d
364
365
366    def test_items(self):
367        """
368        Test sending items request.
369        """
370        def cb(items):
371            self.assertEquals([], items)
372
373        d = self.protocol.items(JID('pubsub.example.org'), 'test')
374        d.addCallback(cb)
375
376        iq = self.stub.output[-1]
377        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
378        self.assertEquals('get', iq.getAttribute('type'))
379        self.assertEquals('pubsub', iq.pubsub.name)
380        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
381        children = list(domish.generateElementsQNamed(iq.pubsub.children,
382                                                      'items', NS_PUBSUB))
383        self.assertEquals(1, len(children))
384        child = children[0]
385        self.assertEquals('test', child['node'])
386
387        response = toResponse(iq, 'result')
388        items = response.addElement((NS_PUBSUB, 'pubsub')).addElement('items')
389        items['node'] = 'test'
390
391        self.stub.send(response)
392
393        return d
394
395
396    def test_itemsMaxItems(self):
397        """
398        Test sending items request, with limit on the number of items.
399        """
400        def cb(items):
401            self.assertEquals(2, len(items))
402            self.assertEquals([item1, item2], items)
403
404        d = self.protocol.items(JID('pubsub.example.org'), 'test', maxItems=2)
405        d.addCallback(cb)
406
407        iq = self.stub.output[-1]
408        self.assertEquals('pubsub.example.org', iq.getAttribute('to'))
409        self.assertEquals('get', iq.getAttribute('type'))
410        self.assertEquals('pubsub', iq.pubsub.name)
411        self.assertEquals(NS_PUBSUB, iq.pubsub.uri)
412        children = list(domish.generateElementsQNamed(iq.pubsub.children,
413                                                      'items', NS_PUBSUB))
414        self.assertEquals(1, len(children))
415        child = children[0]
416        self.assertEquals('test', child['node'])
417        self.assertEquals('2', child['max_items'])
418
419        response = toResponse(iq, 'result')
420        items = response.addElement((NS_PUBSUB, 'pubsub')).addElement('items')
421        items['node'] = 'test'
422        item1 = items.addElement('item')
423        item1['id'] = 'item1'
424        item2 = items.addElement('item')
425        item2['id'] = 'item2'
426
427        self.stub.send(response)
428
429        return d
430
431
432
433class PubSubServiceTest(unittest.TestCase):
434
435    def setUp(self):
436        self.output = []
437
438    def send(self, obj):
439        self.output.append(obj)
440
441    def test_onPublishNoNode(self):
442        handler = pubsub.PubSubService()
443        handler.parent = self
444        iq = domish.Element((None, 'iq'))
445        iq['from'] = 'user@example.org'
446        iq['to'] = 'pubsub.example.org'
447        iq['type'] = 'set'
448        iq.addElement((NS_PUBSUB, 'pubsub'))
449        iq.pubsub.addElement('publish')
450        handler.handleRequest(iq)
451
452        e = error.exceptionFromStanza(self.output[-1])
453        self.assertEquals('bad-request', e.condition)
454
455    def test_onPublish(self):
456        class Handler(pubsub.PubSubService):
457            def publish(self, *args, **kwargs):
458                self.args = args
459                self.kwargs = kwargs
460
461        handler = Handler()
462        handler.parent = self
463        iq = domish.Element((None, 'iq'))
464        iq['type'] = 'set'
465        iq['from'] = 'user@example.org'
466        iq['to'] = 'pubsub.example.org'
467        iq.addElement((NS_PUBSUB, 'pubsub'))
468        iq.pubsub.addElement('publish')
469        iq.pubsub.publish['node'] = 'test'
470        handler.handleRequest(iq)
471
472        self.assertEqual((JID('user@example.org'),
473                          JID('pubsub.example.org'), 'test', []), handler.args)
474
475    def test_onOptionsGet(self):
476        handler = pubsub.PubSubService()
477        handler.parent = self
478        iq = domish.Element((None, 'iq'))
479        iq['from'] = 'user@example.org'
480        iq['to'] = 'pubsub.example.org'
481        iq['type'] = 'get'
482        iq.addElement((NS_PUBSUB, 'pubsub'))
483        iq.pubsub.addElement('options')
484        handler.handleRequest(iq)
485
486        e = error.exceptionFromStanza(self.output[-1])
487        self.assertEquals('feature-not-implemented', e.condition)
488        self.assertEquals('unsupported', e.appCondition.name)
489        self.assertEquals(NS_PUBSUB_ERRORS, e.appCondition.uri)
Note: See TracBrowser for help on using the repository browser.