source: wokkel/test/test_pubsub.py @ 18:4812cc70807d

Last change on this file since 18:4812cc70807d was 18:4812cc70807d, checked in by Ralph Meijer <ralphm@…>, 14 years ago

Add maxItems argument to PubSubClient?.items(), lot of docs and tests.

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