Mercurial > libervia-pubsub

comparison idavoll/test/test_backend.py @ 172:9bfb00edd0cc

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Fix Backend's getNodeInfo so that it properly adds its errback. Author: ralphm. Fixes #9.
author Ralph Meijer <ralphm@ik.nu>
date Tue, 15 Jan 2008 14:00:07 +0000
parents bc269696ef42
children bd86f0c3fd39
comparison
equal deleted inserted replaced
171:bc269696ef42 172:9bfb00edd0cc
105 items = [pubsub.Item()] 105 items = [pubsub.Item()]
106 d = self.backend.publish('node', items, OWNER) 106 d = self.backend.publish('node', items, OWNER)
107 return d 107 return d
108 108
109 109
110 class BaseTestBackend(object):
111 """
112 Base class for backend stubs.
113 """
114
115 def supports_publisher_affiliation(self):
116 return True
117
118 def supports_outcast_affiliation(self):
119 return True
120
121 def supports_persistent_items(self):
122 return True
123
124 def supports_instant_nodes(self):
125 return True
126
127 def register_notifier(self, observerfn, *args, **kwargs):
128 return
129
110 class PubSubServiceFromBackendTest(unittest.TestCase): 130 class PubSubServiceFromBackendTest(unittest.TestCase):
111 131
112 def test_unsubscribeNotSubscribed(self): 132 def test_unsubscribeNotSubscribed(self):
113 """ 133 """
114 Test unsubscription request when not subscribed. 134 Test unsubscription request when not subscribed.
115 """ 135 """
116 136
117 class TestBackend(object): 137 class TestBackend(BaseTestBackend):
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): 138 def unsubscribe(self, nodeIdentifier, subscriber, requestor):
134 return defer.fail(error.NotSubscribed()) 139 return defer.fail(error.NotSubscribed())
135 140
136 def cb(e): 141 def cb(e):
137 self.assertEquals('unexpected-request', e.condition) 142 self.assertEquals('unexpected-request', e.condition)
139 s = backend.PubSubServiceFromBackend(TestBackend()) 144 s = backend.PubSubServiceFromBackend(TestBackend())
140 d = s.unsubscribe(OWNER, 'test.example.org', 'test', OWNER) 145 d = s.unsubscribe(OWNER, 'test.example.org', 'test', OWNER)
141 self.assertFailure(d, StanzaError) 146 self.assertFailure(d, StanzaError)
142 d.addCallback(cb) 147 d.addCallback(cb)
143 return d 148 return d
149
150 def test_getNodeInfo(self):
151 """
152 Test retrieving node information.
153 """
154
155 class TestBackend(BaseTestBackend):
156 def get_node_type(self, nodeIdentifier):
157 return defer.succeed('leaf')
158
159 def get_node_meta_data(self, nodeIdentifier):
160 return defer.succeed({'pubsub#persist_items': True})
161
162 def cb(info):
163 self.assertIn('type', info)
164 self.assertEquals('leaf', info['type'])
165 self.assertIn('meta-data', info)
166 self.assertEquals({'pubsub#persist_items': True}, info['meta-data'])
167
168 s = backend.PubSubServiceFromBackend(TestBackend())
169 d = s.getNodeInfo(OWNER, 'test.example.org', 'test')
170 d.addCallback(cb)
171 return d