6
from dbus.service import method
8
from dbus import DBusException, Array, Byte
10
from twisted.internet.defer import Deferred
12
from landscape.lib.dbus_util import (get_bus, get_object,
13
method as async_method,
14
byte_array, array_to_string,
15
Object, ServiceUnknownError,
17
from landscape.tests.helpers import (
18
LandscapeIsolatedTest, DBusHelper, LandscapeTest)
19
from landscape.tests.mocker import ARGS, KWARGS
22
class BoringService(Object):
24
bus_name = "sample.service"
25
object_path = "/com/example/BoringService"
26
iface_name = "com.example.BoringService"
37
def return_none(self):
41
class AsynchronousWrapperTests(LandscapeIsolatedTest):
43
helpers = [DBusHelper]
46
super(AsynchronousWrapperTests, self).setUp()
47
self.service = BoringService(self.bus)
48
self.remote_service = get_object(self.bus,
49
self.service.bus_name,
50
self.service.object_path,
51
self.service.iface_name)
53
def test_get_bus(self):
54
self.assertEquals(type(get_bus("session")), dbus.SessionBus)
55
self.assertEquals(type(get_bus("system")), dbus.SystemBus)
56
self.assertRaises(ValueError, get_bus, "nope")
58
def test_get_object_returns_deferred(self):
60
There is a L{dbus.Bus.get_object} replacement, L{get_object}, which
61
returns an object which returns Deferreds on method calls
63
result = self.remote_service.return1()
64
self.assertTrue(isinstance(result, Deferred))
65
result.addCallback(self.assertEquals, 1)
68
def test_get_object_returns_failing_deferred(self):
70
The asynchronous method wrapper deals with errors appropriately, by
71
converting them to errbacks on a Deferred.
73
result = self.remote_service.error()
74
self.assertTrue(isinstance(result, Deferred))
75
self.assertFailure(result, DBusException)
78
def test_return_none(self):
80
L{get_object} has no problems with methods that don't return values.
82
result = self.remote_service.return_none()
83
result.addCallback(self.assertEquals, ())
86
def test_helper_methods(self):
87
"""The wrapper shouldn't get in the way of standard methods."""
88
self.remote_service.connect_to_signal("nononono", lambda: None)
90
def test_default_interface_name(self):
92
When an interface isn't provided to get_object(), one is automatically
93
generated from the object_path. This allows us to work with older
94
versions of Python dbus without too much pain.
96
class MyService(Object):
97
bus_name = "my.bus.name"
98
object_path = "/my/Service"
102
service = MyService(self.bus)
103
remote_service = get_object(self.bus, MyService.bus_name,
104
MyService.object_path)
106
result = remote_service.return2()
107
result.addCallback(self.assertEquals, 2)
111
class HalfSynchronousService(Object):
113
bus_name = "sample.service"
114
object_path = "/com/example/UnitedStatesOfWhatever"
115
iface_name = "com.example.UnitedStatesOfWhateverIface"
117
def __init__(self, bus):
118
super(HalfSynchronousService, self).__init__(bus)
119
self.deferred = Deferred()
121
@async_method(iface_name)
125
@async_method(iface_name)
126
def return_none(self):
129
@async_method(iface_name)
130
def sync_error(self):
133
@async_method(iface_name)
138
class AsynchronousMethodTests(LandscapeIsolatedTest):
140
helpers = [DBusHelper]
143
super(AsynchronousMethodTests, self).setUp()
144
self.service = HalfSynchronousService(self.bus)
145
self.remote_service = get_object(self.bus,
146
self.service.bus_name,
147
self.service.object_path,
148
self.service.iface_name)
150
def test_synchronous_method(self):
152
Using the L{landscape.lib.dbus_util.method} decorator to declare a
153
method basically works the same as L{dbus.service.method}, if you
154
return a value synchronously.
156
return self.remote_service.add1(3).addCallback(self.assertEquals, 4)
158
def test_return_None(self):
160
Methods should be able to return None, and this will be translated to a
163
d = self.remote_service.return_none()
164
return d.addCallback(self.assertEquals, ())
166
def test_asynchronous_method(self):
168
However, when a Deferred is returned, it will be handled so that the
169
return value of the method is the ultimate result of the Deferred.
171
d = self.remote_service.async()
172
d.addCallback(self.assertEquals, "hi")
173
self.service.deferred.callback("hi")
176
def test_synchronous_error(self):
178
Synchronous exceptions are propagated as normal.
180
self.log_helper.ignore_errors(ZeroDivisionError)
181
d = self.remote_service.sync_error()
182
def got_error(dbus_exception):
183
# This is pretty much the best we can do, afaict; it doesn't
184
# include any more information but the type.
185
self.assertTrue("ZeroDivisionError" in str(dbus_exception))
186
# assertFailure to make sure it *actually* fails
187
# assertFailure causes the next callback to get the *exception* object
188
# (not a Failure). That means we should add a callback to check stuff
189
# about the exception, not an errback.
190
self.assertFailure(d, DBusException)
191
d.addCallback(got_error)
194
def test_asynchronous_error(self):
196
Returning a Deferred which fails is propagated in the same way as a
197
synchronous exception is.
199
self.log_helper.ignore_errors(ZeroDivisionError)
200
# ignore the result of the method and convert it to an exception
201
self.service.deferred.addCallback(lambda ignored: 1/0)
202
d = self.remote_service.async()
204
def got_error(dbus_exception):
205
# This is pretty much the best we can do, afaict; it doesn't
206
# include any more information but the type.
207
self.assertTrue("ZeroDivisionError" in str(dbus_exception),
210
# fire off the result of the async method call
211
self.service.deferred.callback("ignored")
213
# assertFailure to make sure it *actually* fails
214
# assertFailure causes the next callback to get the *exception* object
215
# (not a Failure). That means we should add a callback to check stuff
216
# about the exception, not an errback.
217
self.assertFailure(d, DBusException)
218
d.addCallback(got_error)
221
def test_errors_get_logged(self):
223
An exception raised during processing of a method call should be
226
self.log_helper.ignore_errors(ZeroDivisionError)
227
d = self.remote_service.sync_error()
228
def got_error(failure):
229
log = self.logfile.getvalue()
230
self.assertTrue("Traceback" in log)
231
self.assertTrue("ZeroDivisionError" in log)
232
return d.addErrback(got_error)
235
class ErrorHandlingTests(LandscapeIsolatedTest):
237
helpers = [DBusHelper]
239
def test_service_unknown(self):
240
remote_service = get_object(self.bus, "com.foo", "/com/foo/Bar",
241
"com.foo", retry_timeout=0)
242
d = remote_service.foo()
243
self.assertFailure(d, ServiceUnknownError)
248
class SecurityErrorTests(LandscapeTest):
249
"""Tests for cases that SecurityError is raised."""
252
super(SecurityErrorTests, self).setUp()
253
self.bus = self.mocker.mock()
254
self.service = self.mocker.mock()
255
self.bus.get_object("com.foo", "/com/foo", introspect=False)
256
self.mocker.result(self.service)
257
self.mocker.count(0, None)
258
self.remote = get_object(self.bus, "com.foo", "/com/foo")
260
def _test_security_error(self, error_message):
261
def raise_dbus_error(*args, **kw):
262
kw["error_handler"](DBusException(error_message))
264
self.service.send_message(ARGS, KWARGS)
265
self.mocker.call(raise_dbus_error)
268
d = self.remote.send_message({"type": "text-message",
270
self.assertFailure(d, SecurityError)
273
def test_feisty_security_error(self):
275
When an exception that looks like a security error from DBUS
276
0.80.x is raised, this should be translated to a
279
return self._test_security_error(
280
"A security policy in place prevents this sender from sending "
281
"this message to this recipient, see message bus configuration "
282
"file (rejected message had interface "
283
'"com.canonical.landscape" member "send_message" error name '
284
'"(unset)" destination ":1.107")')
286
def test_gutsy_security_error(self):
288
When an exception that looks like a security error on DBUS 0.82.x
289
is raised, this should be translated to a L{SecurityError}.
291
return self._test_security_error(
292
"org.freedesktop.DBus.Error.AccessDenied: A security "
293
"policy in place prevents this sender from sending this "
294
"message to this recipient, see message bus configuration "
295
"file (rejected message had interface "
296
'"com.canonical.landscape" member "send_message" error '
297
'name "(unset)" destination ":1.15")')
302
class RetryTests(LandscapeIsolatedTest):
304
helpers = [DBusHelper]
307
super(RetryTests, self).setUp()
309
self.remote_service = get_object(self.bus,
310
HalfSynchronousService.bus_name,
311
HalfSynchronousService.object_path,
312
HalfSynchronousService.iface_name)
315
super(RetryTests, self).tearDown()
316
for pid in self.pids:
318
os.kill(pid, signal.SIGKILL)
323
def test_retry_on_first_call(self):
325
If an object is unavailable when a method is called,
326
AsynchronousProxyMethod will continue trying to get it for a while.
328
d = self.remote_service.add1(0)
329
HalfSynchronousService(self.bus)
330
return d.addCallback(self.assertEquals, 1)
332
def _start_service_in_subprocess(self):
333
executable = self.make_path("""\
335
from twisted.internet.glib2reactor import install
337
from twisted.internet import reactor
339
from dbus import SessionBus
344
from landscape.lib.tests.test_dbus_util import HalfSynchronousService
347
HalfSynchronousService(bus)
349
""" % (sys.executable, sys.path))
350
os.chmod(executable, 0755)
353
os.execlp(executable, executable)
354
self.pids.append(pid)
356
def _stop_service_in_subprocess(self):
357
os.kill(self.pids[-1], signal.SIGKILL)
358
os.waitpid(self.pids[-1], 0)
360
def test_retry_on_second_call(self):
362
Either get_object or an actual method call may raise an exception about
363
an object not being available. This test ensures that the exception
364
raised from the method call is handled to retry, by causing the object
365
to be cached with an initial successful call.
367
self._start_service_in_subprocess()
368
result = self.remote_service.add1(0)
369
def got_first_result(result):
370
self.assertEquals(result, 1)
371
self._stop_service_in_subprocess()
372
second_result = self.remote_service.add1(1)
373
self._start_service_in_subprocess()
375
result.addCallback(got_first_result)
376
result.addCallback(self.assertEquals, 2)
379
def test_timeout_time(self):
381
It's possible to specify the retry timout to use, and when the timeout
382
is reached, the underlying dbus error will be raised.
384
remote_service = get_object(self.bus,
385
HalfSynchronousService.bus_name,
386
HalfSynchronousService.object_path,
389
start_time = time.time()
390
result = remote_service.add1(0)
391
self.assertFailure(result, ServiceUnknownError)
393
def got_error(exception):
394
self.assertTrue(time.time() - start_time > 1)
395
return result.addCallback(got_error)
398
def test_catch_synchronous_errors_from_method(self):
400
DBus sometimes raises synchronous errors from calling the method, for
401
example, when the value passed does not match the signature. The
402
asynchronous call wrapper should handle this case and fail the
405
self.log_helper.ignore_errors(".*Unable to set arguments.*")
406
HalfSynchronousService(self.bus)
407
d = self.remote_service.add1(None)
408
self.assertFailure(d, TypeError)
412
class UtilityTests(LandscapeTest):
413
def test_array_to_string(self):
414
self.assertEquals(array_to_string([102, 111, 111]), "foo")
416
array_to_string(Array([Byte(102), Byte(111), Byte(111)])),
419
def test_byte_array(self):
420
self.assertEquals(byte_array("foo"), [102, 111, 111])
421
self.assertEquals(byte_array("foo"),
422
Array([Byte(102), Byte(111), Byte(111)]))
424
def test_array_to_string_with_horrible_dapper_signing_bug(self):
426
In older versions of dbus, bytes would be deserialized incorrectly as
427
signed. array_to_string compensates for this.
429
self.assertEquals(array_to_string([-56, -127]), chr(200) + chr(129))
added
removed
landscape/lib/tests/test_dbus_util.py
