# 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): 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.MockActionUpdate = None self.MockEvent = None self.MockCB = None self.MockAlive = True self.MockEventAddr = None def isAlive(self): return self.MockAlive def install(self, action, block=False): self.MockAction = action self.MockActionUpdate = None return None def update(self, action, newaction, block=False): self.MockAction = action self.MockActionUpdate = newaction return None def uninstall(self, action, block=False): self.MockAction = None self.MockActionUpdate = None return None def subscribe(self, event, callback, block=True): #Do like the current Probe if not block: raise RuntimeError("This function does not allow non-blocking mode yet") self.MockEvent= event self.MockCB = callback return str(id(event)) def unsubscribe(self, address, block=True): 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() #ErrorCase: install, update, uninstall without currentActivity #Action functions should do a warning if there is no activity self.assertRaises(RuntimeWarning, self.probeManager.install, ad1) self.assertRaises(RuntimeWarning, self.probeManager.update, ad1, ad1) self.assertRaises(RuntimeWarning, self.probeManager.uninstall, ad1) 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) 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, 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) assert act1.MockAction == ad1, "Action should still be installed" self.probeManager.currentActivity = "act1" self.probeManager.uninstall(ad1) 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] ad1 = MockAddon() ad2 = MockAddon() ad2.i, ad2.s = (2, "test2") 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, ad1, 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(ad1, 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 = {} # TODO: Clean-Up the dbus session bus ??? 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" #Check if the installed action is the good one address = "Addr1" #Set the return value of probe install self.mockObj.MockRet["install"] = address self.probeProxy.install(action, block=True) assert pickle.loads(self.mockObj.MockCall["install"]["args"][0]) == action, "1 argument, the action" #ErrorCase: Update should fail on noninstalled actions self.assertRaises(RuntimeWarning, self.probeProxy.update, action2, action2, block=True) #Test the update self.probeProxy.update(action, action2, block=True) 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, block=True) assert not "uninstall" in self.mockObj.MockCall, "Uninstall should not be called if action is not installed" self.probeProxy.uninstall(action, block=True) 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" event2 = MockAddon() event2.i, event2.s = 10, "event2" def callback(event): global message_box message_box = event #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(event, callback, block=True) assert pickle.loads(self.mockObj.MockCall["subscribe"]["args"][0]) == event, "1 argument, the event" #Call the callback with the event global message_box 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", block=True) assert not "unsubscribe" in self.mockObj.MockCall, "Unsubscribe should not be called if event is not subscribeed" self.probeProxy.unsubscribe(address, block=True) assert self.mockObj.MockCall["unsubscribe"]["args"][0] == 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()