import collections import dbus import dbus.service import dbus.mainloop.glib import gobject import logging import threading """ MockFlimflam provides a select few methods from the flimflam DBus API so that we can track "dbus-send" invocations sent by the shill init scripts. It could be used as a kernel for a test of other facilities that use the shill/flimflam DBus API and at that point it should be moved out of this specific test. """ MethodCall = collections.namedtuple("MethodCall", ["method", "argument"]) class FlimflamManager(dbus.service.Object): """ The flimflam DBus Manager object instance. Methods in this object are called whenever a DBus RPC method is invoked. """ def __init__(self, bus, object_path): dbus.service.Object.__init__(self, bus, object_path) self.method_calls = [] @dbus.service.method('org.chromium.flimflam.Manager', in_signature='s', out_signature='o') def CreateProfile(self, profile): """ Creates a profile. @param profile string name of profile to create. """ self.add_method_call('CreateProfile', profile) return '/' @dbus.service.method('org.chromium.flimflam.Manager', in_signature='s', out_signature='') def RemoveProfile(self, profile): """ Removes a profile. @param profile string name of profile to remove. """ self.add_method_call('RemoveProfile', profile) @dbus.service.method('org.chromium.flimflam.Manager', in_signature='s', out_signature='o') def PushProfile(self, profile): """ Pushes a profile. @param profile string name of profile to push. """ self.add_method_call('PushProfile', profile) return '/' @dbus.service.method('org.chromium.flimflam.Manager', in_signature='ss', out_signature='o') def InsertUserProfile(self, profile, user_hash): """ Inserts a profile. @param profile string name of profile to insert. @param user_hash string user hash associated with this profile. """ self.add_method_call('InsertUserProfile', (profile, user_hash)) return '/' @dbus.service.method('org.chromium.flimflam.Manager', in_signature='s', out_signature='') def PopProfile(self, profile): """ Pops a profile. @param profile string name of profile to pop. """ self.add_method_call('PopProfile', profile) @dbus.service.method('org.chromium.flimflam.Manager', in_signature='', out_signature='') def PopAllUserProfiles(self): """ Pops all user profiles from the profile stack.. """ self.add_method_call('PopAllUserProfiles', '') def add_method_call(self, method, arg): """ Note that a method call was made. @param method string the method that was called. @param arg tuple list of arguments that were called on |method|. """ print "Called method %s" % method logging.info("Mock Flimflam method %s called with argument %s", method, arg) self.method_calls.append(MethodCall(method, arg)) def get_method_calls(self): """ Provide the method call list, clears this list internally. @return list of MethodCall objects """ method_calls = self.method_calls self.method_calls = [] return method_calls class MockFlimflam(threading.Thread): """ This thread object instantiates a mock flimflam manager and runs a mainloop that receives DBus API messages. """ FLIMFLAM = "org.chromium.flimflam" def __init__(self): threading.Thread.__init__(self) gobject.threads_init() def run(self): """ Runs the main loop. """ dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) self.bus = dbus.SystemBus() name = dbus.service.BusName(self.FLIMFLAM, self.bus) self.manager = FlimflamManager(self.bus, '/') self.mainloop = gobject.MainLoop() self.mainloop.run() def quit(self): """ Quits the main loop. """ self.mainloop.quit() def get_method_calls(self): """ Returns the method calls that were called on the mock object. @return list of MethodCall objects representing the methods called. """ return self.manager.get_method_calls() if __name__ == '__main__': MockFlimflam().run()