Mercurial > libervia-pubsub
changeset 168:e2c2c2baf483
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 | 96afb248df5e |
files | INSTALL idavoll/backend.py idavoll/test/test_backend.py |
diffstat | 3 files changed, 82 insertions(+), 32 deletions(-) [+] |
line wrap: on
line diff
--- a/INSTALL Wed Oct 03 12:41:43 2007 +0000 +++ b/INSTALL Wed Oct 03 19:07:50 2007 +0000 @@ -3,7 +3,7 @@ - Twisted Core >= 2.0.0 - Twisted Words >= 0.4.0 -- uuid.py (http://ofxsuite.berlios.de/uuid.py) +- uuid.py (Python 2.5 std. lib. or http://pypi.python.org/pypi/uuid) - A jabber server that supports the component protocol (JEP-0114) - Wokkel (http://wokkel.ik.nu/)
--- a/idavoll/backend.py Wed Oct 03 12:41:43 2007 +0000 +++ b/idavoll/backend.py Wed Oct 03 19:07:50 2007 +0000 @@ -108,7 +108,7 @@ if persist_items or deliver_payloads: for item in items: if not item.getAttribute("id"): - item["id"] = uuid.generate() + item["id"] = str(uuid.uuid4()) if persist_items: d = node.store_items(items, requestor) @@ -186,7 +186,7 @@ def create_node(self, node_id, requestor): if not node_id: - node_id = 'generic/%s' % uuid.generate() + node_id = 'generic/%s' % uuid.uuid4() d = self.storage.create_node(node_id, requestor) d.addCallback(lambda _: node_id) return d
--- a/idavoll/test/test_backend.py Wed Oct 03 12:41:43 2007 +0000 +++ b/idavoll/test/test_backend.py Wed Oct 03 19:07:50 2007 +0000 @@ -5,45 +5,37 @@ Tests for L{idavoll.backend}. """ +from zope.interface import implements + +from twisted.internet import defer from twisted.trial import unittest -from zope.interface import implements -from twisted.internet import defer from twisted.words.protocols.jabber import jid +from wokkel import pubsub + from idavoll import backend, error, iidavoll OWNER = jid.JID('owner@example.com') -class testNode: - id = 'to-be-deleted' - def get_affiliation(self, entity): - if entity is OWNER: - return defer.succeed('owner') - -class testStorage: - - implements(iidavoll.IStorage) - - def get_node(self, node_id): - return defer.succeed(testNode()) +class BackendTest(unittest.TestCase): + def test_delete_node(self): + class testNode: + id = 'to-be-deleted' + def get_affiliation(self, entity): + if entity is OWNER: + return defer.succeed('owner') - def delete_node(self, node_id): - if node_id in ['to-be-deleted']: - self.delete_called = True - return defer.succeed(None) - else: - return defer.fail(error.NodeNotFound()) + class testStorage: + def get_node(self, node_id): + return defer.succeed(testNode()) -class BackendTest(unittest.TestCase): - def setUp(self): - self.storage = testStorage() - self.backend = backend.BackendService(self.storage) - self.storage.backend = self.backend + def delete_node(self, node_id): + if node_id in ['to-be-deleted']: + self.delete_called = True + return defer.succeed(None) + else: + return defer.fail(error.NodeNotFound()) - self.pre_delete_called = False - self.delete_called = False - - def testDeleteNode(self): def pre_delete(node_id): self.pre_delete_called = True return defer.succeed(None) @@ -52,7 +44,65 @@ self.assertTrue(self.pre_delete_called) self.assertTrue(self.storage.delete_called) + self.storage = testStorage() + self.backend = backend.BackendService(self.storage) + self.storage.backend = self.backend + + self.pre_delete_called = False + self.delete_called = False + self.backend.register_pre_delete(pre_delete) d = self.backend.delete_node('to-be-deleted', OWNER) d.addCallback(cb) return d + + def test_create_nodeNoID(self): + """ + Test creation of a node without a given node identifier. + """ + class testStorage: + def create_node(self, node_id, requestor): + self.node_id = node_id + return defer.succeed(None) + + self.storage = testStorage() + self.backend = backend.BackendService(self.storage) + self.storage.backend = self.backend + + def checkID(nodeIdentifier): + self.assertNotIdentical(None, nodeIdentifier) + self.assertIdentical(self.storage.node_id, nodeIdentifier) + + d = self.backend.create_node(None, OWNER) + d.addCallback(checkID) + return d + + def test_publishNoID(self): + """ + Test publish request with an item without a node identifier. + """ + class testNode: + id = 'node' + def get_affiliation(self, entity): + if entity is OWNER: + return defer.succeed('owner') + def get_configuration(self): + return {'pubsub#deliver_payloads': True, + 'pubsub#persist_items': False} + + class testStorage: + def get_node(self, node_id): + return defer.succeed(testNode()) + + def checkID(notification): + self.assertNotIdentical(None, notification['items'][0]['id']) + + self.storage = testStorage() + self.backend = backend.BackendService(self.storage) + self.storage.backend = self.backend + + self.backend.register_notifier(checkID) + + items = [pubsub.Item()] + d = self.backend.publish('node', items, OWNER) + return d