Mercurial > libervia-pubsub

comparison sat_pubsub/test/test_storage.py @ 232:923281d4c5bc

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
renamed idavoll directory to sat_pubsub
author Goffi <goffi@goffi.org>
date Thu, 17 May 2012 12:48:14 +0200
parents idavoll/test/test_storage.py@8540825f85e0
children 564ae55219e1
comparison
equal deleted inserted replaced
231:d99047cd90f9 232:923281d4c5bc
1 # Copyright (c) 2003-2010 Ralph Meijer
2 # See LICENSE for details.
3
4 """
5 Tests for L{idavoll.memory_storage} and L{idavoll.pgsql_storage}.
6 """
7
8 from zope.interface.verify import verifyObject
9 from twisted.trial import unittest
10 from twisted.words.protocols.jabber import jid
11 from twisted.internet import defer
12 from twisted.words.xish import domish
13
14 from idavoll import error, iidavoll
15
16 OWNER = jid.JID('owner@example.com/Work')
17 SUBSCRIBER = jid.JID('subscriber@example.com/Home')
18 SUBSCRIBER_NEW = jid.JID('new@example.com/Home')
19 SUBSCRIBER_TO_BE_DELETED = jid.JID('to_be_deleted@example.com/Home')
20 SUBSCRIBER_PENDING = jid.JID('pending@example.com/Home')
21 PUBLISHER = jid.JID('publisher@example.com')
22 ITEM = domish.Element((None, 'item'))
23 ITEM['id'] = 'current'
24 ITEM.addElement(('testns', 'test'), content=u'Test \u2083 item')
25 ITEM_NEW = domish.Element((None, 'item'))
26 ITEM_NEW['id'] = 'new'
27 ITEM_NEW.addElement(('testns', 'test'), content=u'Test \u2083 item')
28 ITEM_UPDATED = domish.Element((None, 'item'))
29 ITEM_UPDATED['id'] = 'current'
30 ITEM_UPDATED.addElement(('testns', 'test'), content=u'Test \u2084 item')
31 ITEM_TO_BE_DELETED = domish.Element((None, 'item'))
32 ITEM_TO_BE_DELETED['id'] = 'to-be-deleted'
33 ITEM_TO_BE_DELETED.addElement(('testns', 'test'), content=u'Test \u2083 item')
34
35 def decode(object):
36 if isinstance(object, str):
37 object = object.decode('utf-8')
38 return object
39
40
41
42 class StorageTests:
43
44 def _assignTestNode(self, node):
45 self.node = node
46
47
48 def setUp(self):
49 d = self.s.getNode('pre-existing')
50 d.addCallback(self._assignTestNode)
51 return d
52
53
54 def test_interfaceIStorage(self):
55 self.assertTrue(verifyObject(iidavoll.IStorage, self.s))
56
57
58 def test_interfaceINode(self):
59 self.assertTrue(verifyObject(iidavoll.INode, self.node))
60
61
62 def test_interfaceILeafNode(self):
63 self.assertTrue(verifyObject(iidavoll.ILeafNode, self.node))
64
65
66 def test_getNode(self):
67 return self.s.getNode('pre-existing')
68
69
70 def test_getNonExistingNode(self):
71 d = self.s.getNode('non-existing')
72 self.assertFailure(d, error.NodeNotFound)
73 return d
74
75
76 def test_getNodeIDs(self):
77 def cb(nodeIdentifiers):
78 self.assertIn('pre-existing', nodeIdentifiers)
79 self.assertNotIn('non-existing', nodeIdentifiers)
80
81 return self.s.getNodeIds().addCallback(cb)
82
83
84 def test_createExistingNode(self):
85 config = self.s.getDefaultConfiguration('leaf')
86 config['pubsub#node_type'] = 'leaf'
87 d = self.s.createNode('pre-existing', OWNER, config)
88 self.assertFailure(d, error.NodeExists)
89 return d
90
91
92 def test_createNode(self):
93 def cb(void):
94 d = self.s.getNode('new 1')
95 return d
96
97 config = self.s.getDefaultConfiguration('leaf')
98 config['pubsub#node_type'] = 'leaf'
99 d = self.s.createNode('new 1', OWNER, config)
100 d.addCallback(cb)
101 return d
102
103
104 def test_createNodeChangingConfig(self):
105 """
106 The configuration passed to createNode must be free to be changed.
107 """
108 def cb(result):
109 node1, node2 = result
110 self.assertTrue(node1.getConfiguration()['pubsub#persist_items'])
111
112 config = {
113 "pubsub#persist_items": True,
114 "pubsub#deliver_payloads": True,
115 "pubsub#send_last_published_item": 'on_sub',
116 "pubsub#node_type": 'leaf',
117 }
118
119 def unsetPersistItems(_):
120 config["pubsub#persist_items"] = False
121
122 d = defer.succeed(None)
123 d.addCallback(lambda _: self.s.createNode('new 1', OWNER, config))
124 d.addCallback(unsetPersistItems)
125 d.addCallback(lambda _: self.s.createNode('new 2', OWNER, config))
126 d.addCallback(lambda _: defer.gatherResults([
127 self.s.getNode('new 1'),
128 self.s.getNode('new 2')]))
129 d.addCallback(cb)
130 return d
131
132
133 def test_deleteNonExistingNode(self):
134 d = self.s.deleteNode('non-existing')
135 self.assertFailure(d, error.NodeNotFound)
136 return d
137
138
139 def test_deleteNode(self):
140 def cb(void):
141 d = self.s.getNode('to-be-deleted')
142 self.assertFailure(d, error.NodeNotFound)
143 return d
144
145 d = self.s.deleteNode('to-be-deleted')
146 d.addCallback(cb)
147 return d
148
149
150 def test_getAffiliations(self):
151 def cb(affiliations):
152 self.assertIn(('pre-existing', 'owner'), affiliations)
153
154 d = self.s.getAffiliations(OWNER)
155 d.addCallback(cb)
156 return d
157
158
159 def test_getSubscriptions(self):
160 def cb(subscriptions):
161 found = False
162 for subscription in subscriptions:
163 if (subscription.nodeIdentifier == 'pre-existing' and
164 subscription.subscriber == SUBSCRIBER and
165 subscription.state == 'subscribed'):
166 found = True
167 self.assertTrue(found)
168
169 d = self.s.getSubscriptions(SUBSCRIBER)
170 d.addCallback(cb)
171 return d
172
173
174 # Node tests
175
176 def test_getType(self):
177 self.assertEqual(self.node.getType(), 'leaf')
178
179
180 def test_getConfiguration(self):
181 config = self.node.getConfiguration()
182 self.assertIn('pubsub#persist_items', config.iterkeys())
183 self.assertIn('pubsub#deliver_payloads', config.iterkeys())
184 self.assertEqual(config['pubsub#persist_items'], True)
185 self.assertEqual(config['pubsub#deliver_payloads'], True)
186
187
188 def test_setConfiguration(self):
189 def getConfig(node):
190 d = node.setConfiguration({'pubsub#persist_items': False})
191 d.addCallback(lambda _: node)
192 return d
193
194 def checkObjectConfig(node):
195 config = node.getConfiguration()
196 self.assertEqual(config['pubsub#persist_items'], False)
197
198 def getNode(void):
199 return self.s.getNode('to-be-reconfigured')
200
201 def checkStorageConfig(node):
202 config = node.getConfiguration()
203 self.assertEqual(config['pubsub#persist_items'], False)
204
205 d = self.s.getNode('to-be-reconfigured')
206 d.addCallback(getConfig)
207 d.addCallback(checkObjectConfig)
208 d.addCallback(getNode)
209 d.addCallback(checkStorageConfig)
210 return d
211
212
213 def test_getMetaData(self):
214 metaData = self.node.getMetaData()
215 for key, value in self.node.getConfiguration().iteritems():
216 self.assertIn(key, metaData.iterkeys())
217 self.assertEqual(value, metaData[key])
218 self.assertIn('pubsub#node_type', metaData.iterkeys())
219 self.assertEqual(metaData['pubsub#node_type'], 'leaf')
220
221
222 def test_getAffiliation(self):
223 def cb(affiliation):
224 self.assertEqual(affiliation, 'owner')
225
226 d = self.node.getAffiliation(OWNER)
227 d.addCallback(cb)
228 return d
229
230
231 def test_getNonExistingAffiliation(self):
232 def cb(affiliation):
233 self.assertEqual(affiliation, None)
234
235 d = self.node.getAffiliation(SUBSCRIBER)
236 d.addCallback(cb)
237 return d
238
239
240 def test_addSubscription(self):
241 def cb1(void):
242 return self.node.getSubscription(SUBSCRIBER_NEW)
243
244 def cb2(subscription):
245 self.assertEqual(subscription.state, 'pending')
246
247 d = self.node.addSubscription(SUBSCRIBER_NEW, 'pending', {})
248 d.addCallback(cb1)
249 d.addCallback(cb2)
250 return d
251
252
253 def test_addExistingSubscription(self):
254 d = self.node.addSubscription(SUBSCRIBER, 'pending', {})
255 self.assertFailure(d, error.SubscriptionExists)
256 return d
257
258
259 def test_getSubscription(self):
260 def cb(subscriptions):
261 self.assertEquals(subscriptions[0].state, 'subscribed')
262 self.assertEquals(subscriptions[1].state, 'pending')
263 self.assertEquals(subscriptions[2], None)
264
265 d = defer.gatherResults([self.node.getSubscription(SUBSCRIBER),
266 self.node.getSubscription(SUBSCRIBER_PENDING),
267 self.node.getSubscription(OWNER)])
268 d.addCallback(cb)
269 return d
270
271
272 def test_removeSubscription(self):
273 return self.node.removeSubscription(SUBSCRIBER_TO_BE_DELETED)
274
275
276 def test_removeNonExistingSubscription(self):
277 d = self.node.removeSubscription(OWNER)
278 self.assertFailure(d, error.NotSubscribed)
279 return d
280
281
282 def test_getNodeSubscriptions(self):
283 def extractSubscribers(subscriptions):
284 return [subscription.subscriber for subscription in subscriptions]
285
286 def cb(subscribers):
287 self.assertIn(SUBSCRIBER, subscribers)
288 self.assertNotIn(SUBSCRIBER_PENDING, subscribers)
289 self.assertNotIn(OWNER, subscribers)
290
291 d = self.node.getSubscriptions('subscribed')
292 d.addCallback(extractSubscribers)
293 d.addCallback(cb)
294 return d
295
296
297 def test_isSubscriber(self):
298 def cb(subscribed):
299 self.assertEquals(subscribed[0][1], True)
300 self.assertEquals(subscribed[1][1], True)
301 self.assertEquals(subscribed[2][1], False)
302 self.assertEquals(subscribed[3][1], False)
303
304 d = defer.DeferredList([self.node.isSubscribed(SUBSCRIBER),
305 self.node.isSubscribed(SUBSCRIBER.userhostJID()),
306 self.node.isSubscribed(SUBSCRIBER_PENDING),
307 self.node.isSubscribed(OWNER)])
308 d.addCallback(cb)
309 return d
310
311
312 def test_storeItems(self):
313 def cb1(void):
314 return self.node.getItemsById(['new'])
315
316 def cb2(result):
317 self.assertEqual(ITEM_NEW.toXml(), result[0].toXml())
318
319 d = self.node.storeItems([ITEM_NEW], PUBLISHER)
320 d.addCallback(cb1)
321 d.addCallback(cb2)
322 return d
323
324
325 def test_storeUpdatedItems(self):
326 def cb1(void):
327 return self.node.getItemsById(['current'])
328
329 def cb2(result):
330 self.assertEqual(ITEM_UPDATED.toXml(), result[0].toXml())
331
332 d = self.node.storeItems([ITEM_UPDATED], PUBLISHER)
333 d.addCallback(cb1)
334 d.addCallback(cb2)
335 return d
336
337
338 def test_removeItems(self):
339 def cb1(result):
340 self.assertEqual(['to-be-deleted'], result)
341 return self.node.getItemsById(['to-be-deleted'])
342
343 def cb2(result):
344 self.assertEqual(0, len(result))
345
346 d = self.node.removeItems(['to-be-deleted'])
347 d.addCallback(cb1)
348 d.addCallback(cb2)
349 return d
350
351
352 def test_removeNonExistingItems(self):
353 def cb(result):
354 self.assertEqual([], result)
355
356 d = self.node.removeItems(['non-existing'])
357 d.addCallback(cb)
358 return d
359
360
361 def test_getItems(self):
362 def cb(result):
363 items = [item.toXml() for item in result]
364 self.assertIn(ITEM.toXml(), items)
365
366 d = self.node.getItems()
367 d.addCallback(cb)
368 return d
369
370
371 def test_lastItem(self):
372 def cb(result):
373 self.assertEqual(1, len(result))
374 self.assertEqual(ITEM.toXml(), result[0].toXml())
375
376 d = self.node.getItems(1)
377 d.addCallback(cb)
378 return d
379
380
381 def test_getItemsById(self):
382 def cb(result):
383 self.assertEqual(1, len(result))
384
385 d = self.node.getItemsById(['current'])
386 d.addCallback(cb)
387 return d
388
389
390 def test_getNonExistingItemsById(self):
391 def cb(result):
392 self.assertEqual(0, len(result))
393
394 d = self.node.getItemsById(['non-existing'])
395 d.addCallback(cb)
396 return d
397
398
399 def test_purge(self):
400 def cb1(node):
401 d = node.purge()
402 d.addCallback(lambda _: node)
403 return d
404
405 def cb2(node):
406 return node.getItems()
407
408 def cb3(result):
409 self.assertEqual([], result)
410
411 d = self.s.getNode('to-be-purged')
412 d.addCallback(cb1)
413 d.addCallback(cb2)
414 d.addCallback(cb3)
415 return d
416
417
418 def test_getNodeAffilatiations(self):
419 def cb1(node):
420 return node.getAffiliations()
421
422 def cb2(affiliations):
423 affiliations = dict(((a[0].full(), a[1]) for a in affiliations))
424 self.assertEquals(affiliations[OWNER.userhost()], 'owner')
425
426 d = self.s.getNode('pre-existing')
427 d.addCallback(cb1)
428 d.addCallback(cb2)
429 return d
430
431
432
433 class MemoryStorageStorageTestCase(unittest.TestCase, StorageTests):
434
435 def setUp(self):
436 from idavoll.memory_storage import Storage, PublishedItem, LeafNode
437 from idavoll.memory_storage import Subscription
438
439 defaultConfig = Storage.defaultConfig['leaf']
440
441 self.s = Storage()
442 self.s._nodes['pre-existing'] = \
443 LeafNode('pre-existing', OWNER, defaultConfig)
444 self.s._nodes['to-be-deleted'] = \
445 LeafNode('to-be-deleted', OWNER, None)
446 self.s._nodes['to-be-reconfigured'] = \
447 LeafNode('to-be-reconfigured', OWNER, defaultConfig)
448 self.s._nodes['to-be-purged'] = \
449 LeafNode('to-be-purged', OWNER, None)
450
451 subscriptions = self.s._nodes['pre-existing']._subscriptions
452 subscriptions[SUBSCRIBER.full()] = Subscription('pre-existing',
453 SUBSCRIBER,
454 'subscribed')
455 subscriptions[SUBSCRIBER_TO_BE_DELETED.full()] = \
456 Subscription('pre-existing', SUBSCRIBER_TO_BE_DELETED,
457 'subscribed')
458 subscriptions[SUBSCRIBER_PENDING.full()] = \
459 Subscription('pre-existing', SUBSCRIBER_PENDING,
460 'pending')
461
462 item = PublishedItem(ITEM_TO_BE_DELETED, PUBLISHER)
463 self.s._nodes['pre-existing']._items['to-be-deleted'] = item
464 self.s._nodes['pre-existing']._itemlist.append(item)
465 self.s._nodes['to-be-purged']._items['to-be-deleted'] = item
466 self.s._nodes['to-be-purged']._itemlist.append(item)
467 item = PublishedItem(ITEM, PUBLISHER)
468 self.s._nodes['pre-existing']._items['current'] = item
469 self.s._nodes['pre-existing']._itemlist.append(item)
470
471 return StorageTests.setUp(self)
472
473
474
475 class PgsqlStorageStorageTestCase(unittest.TestCase, StorageTests):
476
477 dbpool = None
478
479 def setUp(self):
480 from idavoll.pgsql_storage import Storage
481 from twisted.enterprise import adbapi
482 if self.dbpool is None:
483 self.__class__.dbpool = adbapi.ConnectionPool('psycopg2',
484 database='pubsub_test',
485 cp_reconnect=True,
486 client_encoding='utf-8',
487 )
488 self.s = Storage(self.dbpool)
489 self.dbpool.start()
490 d = self.dbpool.runInteraction(self.init)
491 d.addCallback(lambda _: StorageTests.setUp(self))
492 return d
493
494
495 def tearDown(self):
496 return self.dbpool.runInteraction(self.cleandb)
497
498
499 def init(self, cursor):
500 self.cleandb(cursor)
501 cursor.execute("""INSERT INTO nodes
502 (node, node_type, persist_items)
503 VALUES ('pre-existing', 'leaf', TRUE)""")
504 cursor.execute("""INSERT INTO nodes (node) VALUES ('to-be-deleted')""")
505 cursor.execute("""INSERT INTO nodes (node) VALUES ('to-be-reconfigured')""")
506 cursor.execute("""INSERT INTO nodes (node) VALUES ('to-be-purged')""")
507 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""",
508 (OWNER.userhost(),))
509 cursor.execute("""INSERT INTO affiliations
510 (node_id, entity_id, affiliation)
511 SELECT node_id, entity_id, 'owner'
512 FROM nodes, entities
513 WHERE node='pre-existing' AND jid=%s""",
514 (OWNER.userhost(),))
515 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""",
516 (SUBSCRIBER.userhost(),))
517 cursor.execute("""INSERT INTO subscriptions
518 (node_id, entity_id, resource, state)
519 SELECT node_id, entity_id, %s, 'subscribed'
520 FROM nodes, entities
521 WHERE node='pre-existing' AND jid=%s""",
522 (SUBSCRIBER.resource,
523 SUBSCRIBER.userhost()))
524 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""",
525 (SUBSCRIBER_TO_BE_DELETED.userhost(),))
526 cursor.execute("""INSERT INTO subscriptions
527 (node_id, entity_id, resource, state)
528 SELECT node_id, entity_id, %s, 'subscribed'
529 FROM nodes, entities
530 WHERE node='pre-existing' AND jid=%s""",
531 (SUBSCRIBER_TO_BE_DELETED.resource,
532 SUBSCRIBER_TO_BE_DELETED.userhost()))
533 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""",
534 (SUBSCRIBER_PENDING.userhost(),))
535 cursor.execute("""INSERT INTO subscriptions
536 (node_id, entity_id, resource, state)
537 SELECT node_id, entity_id, %s, 'pending'
538 FROM nodes, entities
539 WHERE node='pre-existing' AND jid=%s""",
540 (SUBSCRIBER_PENDING.resource,
541 SUBSCRIBER_PENDING.userhost()))
542 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""",
543 (PUBLISHER.userhost(),))
544 cursor.execute("""INSERT INTO items
545 (node_id, publisher, item, data, date)
546 SELECT node_id, %s, 'to-be-deleted', %s,
547 now() - interval '1 day'
548 FROM nodes
549 WHERE node='pre-existing'""",
550 (PUBLISHER.userhost(),
551 ITEM_TO_BE_DELETED.toXml()))
552 cursor.execute("""INSERT INTO items (node_id, publisher, item, data)
553 SELECT node_id, %s, 'to-be-deleted', %s
554 FROM nodes
555 WHERE node='to-be-purged'""",
556 (PUBLISHER.userhost(),
557 ITEM_TO_BE_DELETED.toXml()))
558 cursor.execute("""INSERT INTO items (node_id, publisher, item, data)
559 SELECT node_id, %s, 'current', %s
560 FROM nodes
561 WHERE node='pre-existing'""",
562 (PUBLISHER.userhost(),
563 ITEM.toXml()))
564
565
566 def cleandb(self, cursor):
567 cursor.execute("""DELETE FROM nodes WHERE node in
568 ('non-existing', 'pre-existing', 'to-be-deleted',
569 'new 1', 'new 2', 'new 3', 'to-be-reconfigured',
570 'to-be-purged')""")
571 cursor.execute("""DELETE FROM entities WHERE jid=%s""",
572 (OWNER.userhost(),))
573 cursor.execute("""DELETE FROM entities WHERE jid=%s""",
574 (SUBSCRIBER.userhost(),))
575 cursor.execute("""DELETE FROM entities WHERE jid=%s""",
576 (SUBSCRIBER_TO_BE_DELETED.userhost(),))
577 cursor.execute("""DELETE FROM entities WHERE jid=%s""",
578 (SUBSCRIBER_PENDING.userhost(),))
579 cursor.execute("""DELETE FROM entities WHERE jid=%s""",
580 (PUBLISHER.userhost(),))
581
582 try:
583 import psycopg2
584 except ImportError:
585 PgsqlStorageStorageTestCase.skip = "Psycopg2 not available"