Web   ·   Wiki   ·   Activities   ·   Blog   ·   Lists   ·   Chat   ·   Meeting   ·   Bugs   ·   Git   ·   Translate   ·   Archive   ·   People   ·   Donate
summaryrefslogtreecommitdiffstats
path: root/tests/probetests.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/probetests.py')
-rw-r--r--tests/probetests.py62
1 files changed, 39 insertions, 23 deletions
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