← Back to branch summary

~ubuntuone-control-tower/ubuntuone-dev-tools/stable-0-2

« back to all changes in this revision

Viewing changes to ubuntuone/devtools/testcase.py

  • Committer: Tarmac
  • Author(s): Rodney Dawes
  • Date: 2011-06-01 19:31:18 UTC
  • mfrom: (33.1.1 dbus-strict-priv)
  • Revision ID: tarmac-20110601193118-akr9508qdp35idbw
Be more strict about connection to the dbus service, and cleanup of connections
Just enforce disconnection from the bus, rather than trying to remove signals
Don't do inlineCallbacks for now, as it seems to cause many problems

Show diffs side-by-side

added added

removed removed

Lines of Context:
ubuntuone/devtools/testcase.py
27
27
 
28
28
from functools import wraps
29
29
 
30
 
from twisted.internet import defer
31
 
from twisted.python import failure
32
30
from twisted.trial.unittest import TestCase, SkipTest
33
31
 
34
32
# DBusRunner for DBusTestCase using tests
213
211
        services.extend([DBusRunner])
214
212
        return services
215
213
 
216
 
    @defer.inlineCallbacks
217
214
    def setUp(self):
218
215
        """Setup the infrastructure fo the test (dbus service)."""
219
216
        # Class 'BaseTestCase' has no 'setUp' member
224
221
 
225
222
        # We need to ensure DBUS_SESSION_BUS_ADDRESS is private here
226
223
        from urllib import unquote
227
 
        bus_address = os.environ.get('DBUS_SESSION_BUS_ADDRESS', None)
 
224
        bus_address = yield os.environ.get('DBUS_SESSION_BUS_ADDRESS', None)
228
225
        if os.path.dirname(unquote(bus_address.split(',')[0].split('=')[1])) \
229
226
                != os.path.dirname(os.getcwd()):
230
227
            raise InvalidSessionBus('DBUS_SESSION_BUS_ADDRES is wrong.')
231
228
 
232
229
        # Set up the main loop and bus connection
233
 
        self.loop = DBusGMainLoop(set_as_default=True)
234
 
        self.bus = dbus.bus.BusConnection(address_or_type=bus_address,
235
 
                                          mainloop=self.loop)
 
230
        self.loop = yield DBusGMainLoop(set_as_default=True)
 
231
        self.bus = yield dbus.bus.BusConnection(address_or_type=bus_address,
 
232
                                                mainloop=self.loop)
 
233
 
 
234
        # Check that we are on the correct bus for real
 
235
        if len(self.bus.list_names()) > 2:
 
236
            raise InvalidSessionBus('Too many bus connections: %s' %
 
237
                                    len(self.bus.list_names()))
236
238
 
237
239
        # monkeypatch busName.__del__ to avoid errors on gc
238
240
        # we take care of releasing the name in shutdown
239
241
        service.BusName.__del__ = lambda _: None
240
 
        self.bus.set_exit_on_disconnect(False)
 
242
        yield self.bus.set_exit_on_disconnect(False)
241
243
        self.signal_receivers = set()
242
244
 
243
245
    def tearDown(self):
244
246
        """Cleanup the test."""
245
 
        # Class 'BaseTestCase' has no 'tearDown' member
246
 
        # pylint: disable=E1101
247
 
        d = self.cleanup_signal_receivers(self.signal_receivers)
248
 
        d.addBoth(self._tear_down)
249
 
        d.addBoth(lambda _: super(DBusTestCase, self).tearDown())
250
 
        return d
251
 
 
252
 
    def _tear_down(self):
253
 
        """Shutdown."""
254
 
        self.bus.flush()
255
 
        self.bus.close()
256
 
 
257
 
    def error_handler(self, error):
258
 
        """Default error handler for DBus calls."""
259
 
        if isinstance(error, failure.Failure):
260
 
            self.fail(error.getErrorMessage())
261
 
 
262
 
    def cleanup_signal_receivers(self, signal_receivers):
263
 
        """Cleanup self.signal_receivers and returns a deferred."""
264
 
        # dbus modules will be imported by the decorator
265
 
        # pylint: disable=E0602
266
 
        deferreds = []
267
 
        for match in signal_receivers:
268
 
            d = defer.Deferred()
269
 
 
270
 
            def callback(*args):
271
 
                """Callback that accepts *args."""
272
 
                if not d.called:
273
 
                    d.callback(args)
274
 
            self.bus.call_async(dbus.bus.BUS_DAEMON_NAME,
275
 
                                dbus.bus.BUS_DAEMON_PATH,
276
 
                                dbus.bus.BUS_DAEMON_IFACE, 'RemoveMatch', 's',
277
 
                                (str(match),), callback, self.error_handler)
278
 
            deferreds.append(d)
279
 
        if deferreds:
280
 
            return defer.DeferredList(deferreds)
281
 
        else:
282
 
            return defer.succeed(True)
 
247
        yield self.bus.flush()
 
248
        yield self.bus.close()
 
249
        yield super(DBusTestCase, self).tearDown()