From b1ff3d93cd809ce434f911ad004746d8c5998df5 Mon Sep 17 00:00:00 2001 From: mike Date: Tue, 24 Nov 2009 03:08:50 +0000 Subject: LP 448319 : Adding engine tests, fixing probe tests, correcting bugs found with tests --- (limited to 'tests/probetests.py') diff --git a/tests/probetests.py b/tests/probetests.py index bdb9bc9..d43da8e 100644 --- a/tests/probetests.py +++ b/tests/probetests.py @@ -85,6 +85,7 @@ class MockProbeProxy(object): @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 @@ -95,9 +96,10 @@ class MockProbeProxy(object): def isAlive(self): return self.MockAlive - def install(self, action, callback, block=False): + def install(self, action, action_installed_cb, error_cb): self.MockAction = action - self.MockAddressCallback_install = callback + self.MockAddressCallback_install = action_installed_cb + self.MockInstallErrorCallback = error_cb self.MockActionUpdate = None return None @@ -106,21 +108,21 @@ class MockProbeProxy(object): self.MockActionUpdate = newaction return None - def uninstall(self, action_address, block=False): + def uninstall(self, action_address): self.MockAction = None self.MockActionUpdate = None return None - def subscribe(self, event, callback, block=True): + def subscribe(self, event_name, event, notif_cb, subscribe_cb, error_cb): #Do like the current Probe - if not block: - raise RuntimeError("This function does not allow non-blocking mode yet") - + self.MockEventName = event_name self.MockEvent = event - self.MockCB = callback + self.MockCB = notif_cb + self.MockSubscribeCB = subscribe_cb + self.MockSubscriptionErrorCb = error_cb return str(id(event)) - def unsubscribe(self, address, block=True): + def unsubscribe(self, address): self.MockEventAddr = address return None @@ -348,16 +350,18 @@ class ProbeManagerTest(unittest.TestCase): 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, callback) + 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) + 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" @@ -379,22 +383,26 @@ class ProbeManagerTest(unittest.TestCase): 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, ad1, cb1) + 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(ad1, cb1) + 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" @@ -431,50 +439,58 @@ class ProbeProxyTest(unittest.TestCase): #Check if the installed action is the good one address = "Addr1" - def callback(value): + 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, callback, block=True) + 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, block=True) + self.assertRaises(RuntimeWarning, self.probeProxy.update, action2_address, action2) #Test the update - self.probeProxy.update(address, action2, block=True) + 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, block=True) + 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, block=True) + 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(event, callback, block=True) + 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 @@ -487,11 +503,11 @@ class ProbeProxyTest(unittest.TestCase): #ErrorCase: unsubcribe for non subscribed event #Test the unsubscribe - self.probeProxy.unsubscribe("otheraddress", block=True) + self.probeProxy.unsubscribe("otheraddress") 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" + 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 -- cgit v0.9.1