Mercurial > libervia-pubsub

comparison idavoll/test/test_backend.py @ 168:e2c2c2baf483

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Fix use of uuid module now shipping with Python 2.5.
author Ralph Meijer <ralphm@ik.nu>
date Wed, 03 Oct 2007 19:07:50 +0000
parents ef22e4150caa
children bc269696ef42
comparison
equal deleted inserted replaced
167:ef22e4150caa 168:e2c2c2baf483
3 3
4 """ 4 """
5 Tests for L{idavoll.backend}. 5 Tests for L{idavoll.backend}.
6 """ 6 """
7 7
8 from zope.interface import implements
9
10 from twisted.internet import defer
8 from twisted.trial import unittest 11 from twisted.trial import unittest
9 from zope.interface import implements
10 from twisted.internet import defer
11 from twisted.words.protocols.jabber import jid 12 from twisted.words.protocols.jabber import jid
13
14 from wokkel import pubsub
12 15
13 from idavoll import backend, error, iidavoll 16 from idavoll import backend, error, iidavoll
14 17
15 OWNER = jid.JID('owner@example.com') 18 OWNER = jid.JID('owner@example.com')
16 19
17 class testNode: 20 class BackendTest(unittest.TestCase):
18 id = 'to-be-deleted' 21 def test_delete_node(self):
19 def get_affiliation(self, entity): 22 class testNode:
20 if entity is OWNER: 23 id = 'to-be-deleted'
21 return defer.succeed('owner') 24 def get_affiliation(self, entity):
25 if entity is OWNER:
26 return defer.succeed('owner')
22 27
23 class testStorage: 28 class testStorage:
29 def get_node(self, node_id):
30 return defer.succeed(testNode())
24 31
25 implements(iidavoll.IStorage) 32 def delete_node(self, node_id):
33 if node_id in ['to-be-deleted']:
34 self.delete_called = True
35 return defer.succeed(None)
36 else:
37 return defer.fail(error.NodeNotFound())
26 38
27 def get_node(self, node_id):
28 return defer.succeed(testNode())
29
30 def delete_node(self, node_id):
31 if node_id in ['to-be-deleted']:
32 self.delete_called = True
33 return defer.succeed(None)
34 else:
35 return defer.fail(error.NodeNotFound())
36
37 class BackendTest(unittest.TestCase):
38 def setUp(self):
39 self.storage = testStorage()
40 self.backend = backend.BackendService(self.storage)
41 self.storage.backend = self.backend
42
43 self.pre_delete_called = False
44 self.delete_called = False
45
46 def testDeleteNode(self):
47 def pre_delete(node_id): 39 def pre_delete(node_id):
48 self.pre_delete_called = True 40 self.pre_delete_called = True
49 return defer.succeed(None) 41 return defer.succeed(None)
50 42
51 def cb(result): 43 def cb(result):
52 self.assertTrue(self.pre_delete_called) 44 self.assertTrue(self.pre_delete_called)
53 self.assertTrue(self.storage.delete_called) 45 self.assertTrue(self.storage.delete_called)
54 46
47 self.storage = testStorage()
48 self.backend = backend.BackendService(self.storage)
49 self.storage.backend = self.backend
50
51 self.pre_delete_called = False
52 self.delete_called = False
53
55 self.backend.register_pre_delete(pre_delete) 54 self.backend.register_pre_delete(pre_delete)
56 d = self.backend.delete_node('to-be-deleted', OWNER) 55 d = self.backend.delete_node('to-be-deleted', OWNER)
57 d.addCallback(cb) 56 d.addCallback(cb)
58 return d 57 return d
58
59 def test_create_nodeNoID(self):
60 """
61 Test creation of a node without a given node identifier.
62 """
63 class testStorage:
64 def create_node(self, node_id, requestor):
65 self.node_id = node_id
66 return defer.succeed(None)
67
68 self.storage = testStorage()
69 self.backend = backend.BackendService(self.storage)
70 self.storage.backend = self.backend
71
72 def checkID(nodeIdentifier):
73 self.assertNotIdentical(None, nodeIdentifier)
74 self.assertIdentical(self.storage.node_id, nodeIdentifier)
75
76 d = self.backend.create_node(None, OWNER)
77 d.addCallback(checkID)
78 return d
79
80 def test_publishNoID(self):
81 """
82 Test publish request with an item without a node identifier.
83 """
84 class testNode:
85 id = 'node'
86 def get_affiliation(self, entity):
87 if entity is OWNER:
88 return defer.succeed('owner')
89 def get_configuration(self):
90 return {'pubsub#deliver_payloads': True,
91 'pubsub#persist_items': False}
92
93 class testStorage:
94 def get_node(self, node_id):
95 return defer.succeed(testNode())
96
97 def checkID(notification):
98 self.assertNotIdentical(None, notification['items'][0]['id'])
99
100 self.storage = testStorage()
101 self.backend = backend.BackendService(self.storage)
102 self.storage.backend = self.backend
103
104 self.backend.register_notifier(checkID)
105
106 items = [pubsub.Item()]
107 d = self.backend.publish('node', items, OWNER)
108 return d