Mercurial > libervia-pubsub

comparison idavoll/test/test_backend.py @ 171:bc269696ef42

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Reply with the correct error condition on subscription when not subscribed. Author: ralphm. Fixes #7.
author Ralph Meijer <ralphm@ik.nu>
date Thu, 03 Jan 2008 13:10:25 +0000
parents e2c2c2baf483
children 9bfb00edd0cc
comparison
equal deleted inserted replaced
170:958e69630e52 171:bc269696ef42
1 # Copyright (c) 2003-2007 Ralph Meijer 1 # Copyright (c) 2003-2008 Ralph Meijer
2 # See LICENSE for details. 2 # See LICENSE for details.
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.internet import defer
11 from twisted.trial import unittest 9 from twisted.trial import unittest
12 from twisted.words.protocols.jabber import jid 10 from twisted.words.protocols.jabber import jid
11 from twisted.words.protocols.jabber.error import StanzaError
13 12
14 from wokkel import pubsub 13 from wokkel import pubsub
15 14
16 from idavoll import backend, error, iidavoll 15 from idavoll import backend, error
17 16
18 OWNER = jid.JID('owner@example.com') 17 OWNER = jid.JID('owner@example.com')
19 18
20 class BackendTest(unittest.TestCase): 19 class BackendTest(unittest.TestCase):
21 def test_delete_node(self): 20 def test_delete_node(self):
104 self.backend.register_notifier(checkID) 103 self.backend.register_notifier(checkID)
105 104
106 items = [pubsub.Item()] 105 items = [pubsub.Item()]
107 d = self.backend.publish('node', items, OWNER) 106 d = self.backend.publish('node', items, OWNER)
108 return d 107 return d
108
109
110 class PubSubServiceFromBackendTest(unittest.TestCase):
111
112 def test_unsubscribeNotSubscribed(self):
113 """
114 Test unsubscription request when not subscribed.
115 """
116
117 class TestBackend(object):
118 def supports_publisher_affiliation(self):
119 return True
120
121 def supports_outcast_affiliation(self):
122 return True
123
124 def supports_persistent_items(self):
125 return True
126
127 def supports_instant_nodes(self):
128 return True
129
130 def register_notifier(self, observerfn, *args, **kwargs):
131 return
132
133 def unsubscribe(self, nodeIdentifier, subscriber, requestor):
134 return defer.fail(error.NotSubscribed())
135
136 def cb(e):
137 self.assertEquals('unexpected-request', e.condition)
138
139 s = backend.PubSubServiceFromBackend(TestBackend())
140 d = s.unsubscribe(OWNER, 'test.example.org', 'test', OWNER)
141 self.assertFailure(d, StanzaError)
142 d.addCallback(cb)
143 return d