30
from dbus.mainloop.glib import DBusGMainLoop
28
from functools import wraps
32
30
from twisted.internet import defer
33
31
from twisted.python import failure
34
from twisted.trial.unittest import TestCase
32
from twisted.trial.unittest import TestCase, SkipTest
36
34
# DBusRunner for DBusTestCase using tests
37
35
from ubuntuone.devtools.services.dbus import DBusRunner
38
# pylint: disable=F0401,C0103
45
import dbus.service as service
50
from dbus.mainloop.glib import DBusGMainLoop
55
# pylint: enable=F0401,C0103
40
56
@contextlib.contextmanager
41
57
def environ(env_var, new_value):
42
58
"""context manager to replace/add an environ value"""
49
65
os.environ[env_var] = old_value
69
"""Return the obj calling the funct."""
73
# pylint: disable=C0103
75
"""Unconditionally skip a test."""
76
def decorator(test_item):
77
"""Decorate the test so that it is skipped."""
78
if not (isinstance(test_item, type) and\
79
issubclass(test_item, TestCase)):
82
def skip_wrapper(*args, **kwargs):
83
"""Skip a test method raising an exception."""
84
raise SkipTest(reason)
85
test_item = skip_wrapper
87
# tell twisted.trial.unittest to skip the test, pylint will complain
88
# since it thinks we are redefining a name out of the scope
89
# pylint: disable=W0621,W0612
90
test_item.skip = reason
91
# pylint: enable=W0621,W0612
92
# because the item was skipped, we will make sure that no
93
# services are started for it
94
if hasattr(test_item, "required_services"):
95
# pylint: disable=W0612
96
test_item.required_services = lambda *args, **kwargs: []
97
# pylint: enable=W0612
102
def skipIf(condition, reason):
103
"""Skip a test if the condition is true."""
105
return skipTest(reason)
109
def skipIfOS(current_os, reason):
110
"""Skip test for a particular os or lists of them."""
112
if sys.platform in current_os or sys.platform == current_os:
113
return skipTest(reason)
118
def skipIfNotOS(current_os, reason):
119
"""Skip test we are not in a particular os."""
121
if sys.platform not in current_os or\
122
sys.platform != current_os:
123
return skipTest(reason)
128
# pylint: enable=C0103
52
131
class FakeDBusInterface(object):
53
132
"""A fake DBusInterface..."""
118
197
os.makedirs(path)
200
@skipIf(dbus is None or service is None or DBusGMainLoop is None,
201
"The test requires dbus.")
121
202
class DBusTestCase(BaseTestCase):
122
203
"""Test the DBus event handling."""
131
212
"""Setup the infrastructure fo the test (dbus service)."""
132
213
# Class 'BaseTestCase' has no 'setUp' member
133
214
# pylint: disable=E1101
215
# dbus modules will be imported by the decorator
216
# pylint: disable=E0602
134
217
BaseTestCase.setUp(self)
135
218
self.loop = DBusGMainLoop(set_as_default=True)
136
219
self.bus = dbus.bus.BusConnection(mainloop=self.loop)
137
220
# monkeypatch busName.__del__ to avoid errors on gc
138
221
# we take care of releasing the name in shutdown
139
dbus.service.BusName.__del__ = lambda _: None
222
service.BusName.__del__ = lambda _: None
140
223
self.bus.set_exit_on_disconnect(False)
141
224
self.signal_receivers = set()
162
245
def cleanup_signal_receivers(self, signal_receivers):
163
246
"""Cleanup self.signal_receivers and returns a deferred."""
247
# dbus modules will be imported by the decorator
248
# pylint: disable=E0602
165
250
for match in signal_receivers:
166
251
d = defer.Deferred()