# Copyright (C) 2009, Tutorius.org # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA """ Probe Tests """ import unittest import pickle from dbus.mainloop.glib import DBusGMainLoop from dbus.mainloop import NULL_MAIN_LOOP import dbus from sugar.tutorius.TProbe import TProbe, ProbeProxy, ProbeManager from sugar.tutorius import addon from sugar.tutorius.actions import Action from sugar.tutorius.properties import TIntProperty, TStringProperty #Create a substitute addon create function old_addon_create = addon.create fake_addon_cache = {} def new_addon_create(name, *args, **kwargs): if name in fake_addon_cache: return fake_addon_cache[name](*args, **kwargs) else: return old_addon_create(name, *args, **kwargs) message_box = None event_box = None class MockAddon(Action): i = TIntProperty(0) s = TStringProperty("test") def do(self, **kwargs): global message_box message_box = (self.i, self.s) def undo(self): global message_box message_box = None def install_handlers(self, callback, **kwargs): global message_box message_box = callback def remove_handlers(self): global message_box message_box = None fake_addon_cache["MockAddon"] = MockAddon class MockActivity(object): def get_bundle_id(self): return "localhost.unittest.ProbeTest" def get_id(self): return "unique_id_1" class MockProbeProxy(object): _MockProxyCache = {} def __new__(cls, activityName, unique_id): #For testing, use only one instance per activityName return cls._MockProxyCache.setdefault(activityName, super(MockProbeProxy, cls).__new__(cls)) def __init__(self, activityName, unique_id): """ Constructor @param activityName unique activity id. Must be a valid dbus bus name. """ self.MockAction = None self.MockActionName = None self.MockActionUpdate = None self.MockEvent = None self.MockCB = None self.MockAlive = True self.MockEventAddr = None self.MockAddressCallback = None def isAlive(self): return self.MockAlive def install(self, action, action_installed_cb, error_cb): self.MockAction = action self.MockAddressCallback_install = action_installed_cb self.MockInstallErrorCallback = error_cb self.MockActionUpdate = None return None def update(self, action_address, newaction, block=False): self.MockActionAddress = action_address self.MockActionUpdate = newaction return None def uninstall(self, action_address): self.MockAction = None self.MockActionUpdate = None return None def subscribe(self, event_name, event, notif_cb, subscribe_cb, error_cb): #Do like the current Probe self.MockEventName = event_name self.MockEvent = event self.MockCB = notif_cb self.MockSubscribeCB = subscribe_cb self.MockSubscriptionErrorCb = error_cb return str(id(event)) def unsubscribe(self, address): self.MockEventAddr = address return None def detach(self, block=False): self.MockAction = None self.MockActionUpdate = None self.MockEvent = None self.MockCB = None self.MockAlive = False self.MockEventAddr = None return None class MockProxyObject(object): _MockProxyObjects = {} def __new__(cls, name, path): return cls._MockProxyObjects.setdefault((name, path), super(MockProxyObject, cls).__new__(cls)) def __init__(self, name, path): self.MockCall = {} self.MockRet = {} self.MockCB = {} def get_dbus_method(self, name, *args, **kwargs): #FIXME This mockMethod should support asynchronous calling, # and possibly more def mockMethod(*a, **kw): self.MockCall[name] = dict(args=a, kwargs=kw) return self.MockRet.get(name, None) return mockMethod def connect_to_signal(self, signal_name, handler_function, dbus_interface=None, **kw): self.MockCB[signal_name] = dict(handler_function=handler_function, dbus_interface=dbus_interface, **kw) class MockSessionBus(object): def get_object(self, bus_name, object_path, introspect=True, follow_name_owner_changes=False, **kwargs): return MockProxyObject(bus_name, object_path) old_SessionBus = dbus.SessionBus class MockServiceProxy(object): def register_probe(self, process_name, unique_id): pass def unregister_probe(self, unique_id): pass ########################################################################### # Begin Test Cases ########################################################################### class ProbeTest(unittest.TestCase): def setUp(self): global message_box message_box = None #Fix the addon create addon.create = new_addon_create #Set a default dbus mainloop m = DBusGMainLoop(set_as_default=True) dbus.set_default_main_loop(m) #Setup the activity and probe self.activity = MockActivity() self.probe = TProbe(self.activity, MockServiceProxy()) #Override the eventOccured on the Probe... self.old_eO = self.probe.eventOccured def newEo(event): global event_box try: self.old_eO(event) event_box = event except RuntimeError: event_box = None self.probe.eventOccured = newEo def tearDown(self): #Replace addon create addon.create = old_addon_create #Clear the default dbus mainloop dbus.set_default_main_loop(NULL_MAIN_LOOP) #Clear the activity self.probe.remove_from_connection() del self.probe del self.activity def test_ping(self): #Test ping() res = self.probe.ping() assert res == "alive", "Probe should be alive" def test_action(self): global message_box action = MockAddon() action.i, action.s = (5,"woot") assert message_box is None, "Message box should still be empty" #install 1 address = self.probe.install(pickle.dumps(action)) assert type(address) == str, "install should return a string" assert message_box == (5, "woot"), "message box should have (i, s)" #install 2 action.i, action.s = (10, "ahhah!") address2 = self.probe.install(pickle.dumps(action)) assert message_box == (10, "ahhah!"), "message box should have changed" assert address != address2, "action addresses should be different" #uninstall 2 self.probe.uninstall(address2) assert message_box is None, "undo should clear the message box" #update action 1 with action 2 props self.probe.update(address, pickle.dumps(action._props)) assert message_box == (10, "ahhah!"), "message box should have changed(i, s)" #ErrorCase: Update with bad address #try to update 2, should fail self.assertRaises(KeyError, self.probe.update, address2, pickle.dumps(action._props)) self.probe.uninstall(address) assert message_box is None, "undo should clear the message box" message_box = "Test" #ErrorCase: Uninstall bad address (currently silent fail) #Uninstall twice should do nothing self.probe.uninstall(address) assert message_box == "Test", "undo should not have happened again" def test_events(self): global message_box global event_box event = MockAddon() event.i, event.s = (0, "event1") event2 = MockAddon() event2.i, event2.s = (1, "event2") addr = self.probe.subscribe(pickle.dumps(event)) cb1 = message_box addr2 = self.probe.subscribe(pickle.dumps(event2)) cb2 = message_box assert type(addr) == str, "should return a string address" assert addr != addr2, "each subscribe should return a different address" assert event_box is None, "event_box should still be empty" #Do the callback 2 cb2() assert event_box is not None, "event_box should have an event" assert type(event_box) == str, "event should be pickled" assert pickle.loads(event_box) == event2, "event should be event2" #Unsubscribe event 2 self.probe.unsubscribe(addr2) assert message_box is None, "unsubscribe should clear the message_box" #Do the callback 1 cb1() assert pickle.loads(event_box) == event, "event should be event1" #unsubscribe event 1 self.probe.unsubscribe(addr) assert message_box is None, "unsubscribe should clear the message_box" event_box = None #ErrorCase: callback called from unregistered event filter #Do the callback 1 again self.assertRaises(RuntimeWarning, cb1) class ProbeManagerTest(unittest.TestCase): def setUp(self): MockProbeProxy._MockProxyCache = {} self.probeManager = ProbeManager(proxy_class=MockProbeProxy) def test_register_probe(self): assert len(self.probeManager.get_registered_probes_list()) == 0 self.probeManager.register_probe("act1", "unique_id_1") assert len(self.probeManager.get_registered_probes_list()) == 1 assert len(self.probeManager.get_registered_probes_list("act1")) == 1 assert self.probeManager.get_registered_probes_list()[0][0] == "unique_id_1" self.probeManager.register_probe("act2","unique_id_2") assert len(self.probeManager.get_registered_probes_list()) == 2 assert len(self.probeManager.get_registered_probes_list("act1")) == 1 assert self.probeManager.get_registered_probes_list("act1")[0][0] == "unique_id_1" assert len(self.probeManager.get_registered_probes_list("act2")) == 1 assert self.probeManager.get_registered_probes_list("act2")[0][0] == "unique_id_2" def test_register_multiple_probes(self): assert len(self.probeManager.get_registered_probes_list()) == 0 self.probeManager.register_probe("act1", "unique_id_1") self.probeManager.register_probe("act1","unique_id_2") assert len(self.probeManager.get_registered_probes_list()) == 2 assert len(self.probeManager.get_registered_probes_list("act1")) == 2 assert self.probeManager.get_registered_probes_list("act1")[0][0] == "unique_id_1" assert self.probeManager.get_registered_probes_list("act1")[1][0] == "unique_id_2" def test_unregister_probe(self): assert len(self.probeManager.get_registered_probes_list()) == 0 self.probeManager.register_probe("act1", "unique_id_1") self.probeManager.register_probe("act1","unique_id_2") self.probeManager.unregister_probe("unique_id_1") assert len(self.probeManager.get_registered_probes_list("act1")) == 1 assert self.probeManager.get_registered_probes_list("act1")[0][0] == "unique_id_2" self.probeManager.unregister_probe("unique_id_2") assert len(self.probeManager.get_registered_probes_list("act1")) == 0 assert self.probeManager.get_registered_probes_list("act1") == [] def test_actions(self): self.probeManager.register_probe("act1", "unique_id_1") self.probeManager.register_probe("act2", "unique_id_2") act1 = self.probeManager.get_registered_probes_list("act1")[0][1] act2 = self.probeManager.get_registered_probes_list("act2")[0][1] ad1 = MockAddon() ad1_address = "Address1" def callback(value): pass def error_cb(): pass #ErrorCase: install, update, uninstall without currentActivity #Action functions should do a warning if there is no activity self.assertRaises(RuntimeWarning, self.probeManager.install, ad1_address, ad1, callback) self.assertRaises(RuntimeWarning, self.probeManager.update, ad1_address, ad1) self.assertRaises(RuntimeWarning, self.probeManager.uninstall, ad1_address) assert act1.MockAction is None, "Action should not be installed on inactive proxy" assert act2.MockAction is None, "Action should not be installed on inactive proxy" self.probeManager.currentActivity = "act1" self.probeManager.install(ad1, callback, error_cb) assert act1.MockAction == ad1, "Action should have been installed" assert act2.MockAction is None, "Action should not be installed on inactive proxy" self.probeManager.update(ad1_address, ad1) assert act1.MockActionUpdate == ad1, "Action should have been updated" assert act2.MockActionUpdate is None, "Should not update on inactive" self.probeManager.currentActivity = "act2" self.probeManager.uninstall(ad1_address) assert act1.MockActionAddress == ad1_address, "Action should still be installed" self.probeManager.currentActivity = "act1" self.probeManager.uninstall(ad1_address) assert act1.MockAction is None, "Action should be uninstalled" def test_events(self): self.probeManager.register_probe("act1", "unique_id_1") self.probeManager.register_probe("act2", "unique_id_2") act1 = self.probeManager.get_registered_probes_list("act1")[0][1] act2 = self.probeManager.get_registered_probes_list("act2")[0][1] event_name1 = 'State0/event0' event_name2 = 'State0/event1' ad1 = MockAddon() ad2 = MockAddon() ad2.i, ad2.s = (2, "test2") cb1 = lambda *args: None install_cb1 = lambda *args:None error_cb1 = lambda *args:None cb2 = lambda *args: None #ErrorCase: unsubscribe and subscribe without current activity #Event functions should do a warning if there is no activity self.assertRaises(RuntimeWarning, self.probeManager.subscribe, event_name1, ad1, cb1, install_cb1, error_cb1) self.assertRaises(RuntimeWarning, self.probeManager.unsubscribe, None) assert act1.MockEvent is None, "No event should be on act1" assert act2.MockEvent is None, "No event should be on act2" self.probeManager.currentActivity = "act1" self.probeManager.subscribe(event_name1, ad1, cb1, install_cb1, error_cb1) assert act1.MockEvent == ad1, "Event should have been installed" assert act1.MockCB == cb1, "Callback should have been set" assert act2.MockEvent is None, "No event should be on act2" self.probeManager.unsubscribe("SomeAddress") assert act1.MockEventAddr == "SomeAddress", "Unsubscribe should have been called" assert act2.MockEventAddr is None, "Unsubscribe should not have been called" class ProbeProxyTest(unittest.TestCase): def setUp(self): dbus.SessionBus = MockSessionBus self.mockObj = MockProxyObject("unittest.TestCase", "/tutorius/Probe/unique_id_1") self.probeProxy = ProbeProxy("unittest.TestCase", "unique_id_1") def tearDown(self): dbus.SessionBus = old_SessionBus MockProxyObject._MockProxyObjects = {} def test_Alive(self): self.mockObj.MockRet["ping"] = "alive" assert self.probeProxy.isAlive() == True, "Alive should return True" self.mockObj.MockRet["ping"] = "anything else" assert self.probeProxy.isAlive() == False, "Alive should return False" def test_actions(self): action = MockAddon() action.i, action.s = 5, "action" action2 = MockAddon() action2.i, action2.s = 10, "action2" action2_address = "Addr2" #Check if the installed action is the good one address = "Addr1" def action_installed_cb(value): pass def error_cb(value): pass #Set the return value of probe install self.mockObj.MockRet["install"] = address self.probeProxy.install(action, action_installed_cb, error_cb) assert pickle.loads(self.mockObj.MockCall["install"]["args"][0]) == action, "1 argument, the action" self.mockObj.MockCall["install"]["kwargs"]["reply_handler"](address) #ErrorCase: Update should fail on noninstalled actions self.assertRaises(RuntimeWarning, self.probeProxy.update, action2_address, action2) #Test the update self.probeProxy.update(address, action2) args = self.mockObj.MockCall["update"]["args"] assert args[0] == address, "arg 1 should be the action address" assert pickle.loads(args[1]) == action2._props, "arg2 should be the new action properties" #ErrorCase: Uninstall on not installed action (silent fail) #Test the uninstall self.probeProxy.uninstall(action2_address) assert not "uninstall" in self.mockObj.MockCall, "Uninstall should not be called if action is not installed" self.probeProxy.uninstall(address) assert self.mockObj.MockCall["uninstall"]["args"][0] == address, "1 argument, the action address" def test_events(self): event = MockAddon() event.i, event.s = 5, "event" event_address = 'event1' event2 = MockAddon() event2.i, event2.s = 10, "event2" def callback(event): global message_box message_box = event subs_cb = lambda *args : None error_cb = lambda *args : None #Check if the installed event is the good one address = "Addr1" #Set the return value of probe subscribe self.mockObj.MockRet["subscribe"] = address self.probeProxy.subscribe('State0/event0', event, callback, subs_cb, error_cb) self.probeProxy._ProbeProxy__update_event('State0/event0', event, callback, subs_cb, event_address) assert pickle.loads(self.mockObj.MockCall["subscribe"]["args"][0]) == event, "1 argument, the event" #Call the callback with the event global message_box #import rpdb2; rpdb2.start_embedded_debugger('pass') self.mockObj.MockCB["eventOccured"]["handler_function"](pickle.dumps(event)) assert message_box == event, "callback should have been called with event" message_box = None #ErrorCase: eventOccured triggered by a wrong event #Call with a wrong event self.mockObj.MockCB["eventOccured"]["handler_function"](pickle.dumps(event2)) assert message_box is None, "callback should not have been called" #ErrorCase: unsubcribe for non subscribed event #Test the unsubscribe self.probeProxy.unsubscribe("otheraddress") assert not "unsubscribe" in self.mockObj.MockCall, "Unsubscribe should not be called if event is not subscribeed" self.probeProxy.unsubscribe(event_address) assert self.mockObj.MockCall["unsubscribe"]["args"][0] == event_address, "1 argument, the event address" #ErrorCase: eventOccured triggered by uninstalled event #Test the callback with unregistered event self.mockObj.MockCB["eventOccured"]["handler_function"](pickle.dumps(event)) assert message_box is None, "callback should not have been called" if __name__ == "__main__": unittest.main()