Mercurial > libervia-pubsub
diff idavoll/test/test_gateway.py @ 209:7f3ffb7a1a9e
Add support for node deletion with redirect.
author | Ralph Meijer <ralphm@ik.nu> |
---|---|
date | Fri, 30 Jan 2009 14:35:05 +0000 |
parents | 274a45d2a5ab |
children | bfc198af5d27 |
line wrap: on
line diff
--- a/idavoll/test/test_gateway.py Tue Sep 09 14:54:33 2008 +0000 +++ b/idavoll/test/test_gateway.py Fri Jan 30 14:35:05 2009 +0000 @@ -97,6 +97,86 @@ d.addCallback(cb) return d + def test_delete(self): + def cb(response): + xmppURI = response['uri'] + d = self.client.delete(xmppURI) + return d + + d = self.client.create() + d.addCallback(cb) + return d + + def test_deleteWithRedirect(self): + def cb(response): + xmppURI = response['uri'] + redirectURI = 'xmpp:%s?node=test' % componentJID + d = self.client.delete(xmppURI, redirectURI) + return d + + d = self.client.create() + d.addCallback(cb) + return d + + def test_deleteNotification(self): + def onNotification(data, headers): + try: + self.assertTrue(headers.hasHeader('Event')) + self.assertEquals(['DELETED'], headers.getRawHeaders('Event')) + self.assertFalse(headers.hasHeader('Link')) + except: + self.client.deferred.errback() + else: + self.client.deferred.callback(None) + + def cb(response): + xmppURI = response['uri'] + d = self.client.subscribe(xmppURI) + d.addCallback(lambda _: xmppURI) + return d + + def cb2(xmppURI): + d = self.client.delete(xmppURI) + return d + + self.client.callback = onNotification + self.client.deferred = defer.Deferred() + d = self.client.create() + d.addCallback(cb) + d.addCallback(cb2) + return defer.gatherResults([d, self.client.deferred]) + + def test_deleteNotificationWithRedirect(self): + redirectURI = 'xmpp:%s?node=test' % componentJID + + def onNotification(data, headers): + try: + self.assertTrue(headers.hasHeader('Event')) + self.assertEquals(['DELETED'], headers.getRawHeaders('Event')) + self.assertEquals(['<%s>; rel=alternate' % redirectURI], + headers.getRawHeaders('Link')) + except: + self.client.deferred.errback() + else: + self.client.deferred.callback(None) + + def cb(response): + xmppURI = response['uri'] + d = self.client.subscribe(xmppURI) + d.addCallback(lambda _: xmppURI) + return d + + def cb2(xmppURI): + d = self.client.delete(xmppURI, redirectURI) + return d + + self.client.callback = onNotification + self.client.deferred = defer.Deferred() + d = self.client.create() + d.addCallback(cb) + d.addCallback(cb2) + return defer.gatherResults([d, self.client.deferred]) + def test_list(self): d = self.client.listNodes() return d