Web   ·   Wiki   ·   Activities   ·   Blog   ·   Lists   ·   Chat   ·   Meeting   ·   Bugs   ·   Git   ·   Translate   ·   Archive   ·   People   ·   Donate
summaryrefslogtreecommitdiffstats
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/coretests.py17
-rw-r--r--tests/filterstests.py20
-rw-r--r--tests/probetests.py483
3 files changed, 473 insertions, 47 deletions
diff --git a/tests/coretests.py b/tests/coretests.py
index 4d5055e..b9e04e5 100644
--- a/tests/coretests.py
+++ b/tests/coretests.py
@@ -28,7 +28,7 @@ and event filters. Those are in their separate test module
import unittest
-import copy
+from copy import deepcopy
import logging
from sugar.tutorius.actions import *
from sugar.tutorius.addon import *
@@ -275,22 +275,21 @@ class StateTest(unittest.TestCase):
assert not(st1 == st2), "Different state names should give different states"
st2.name = "Identical"
- st3 = copy.deepcopy(st1)
+ st3 = deepcopy(st1)
st3.add_action(addon.create("BubbleMessage", "Hi!", [128,264]))
assert not (st1 == st3), "States having a different number of actions should be different"
- st4 = copy.deepcopy(st1)
- st4.add_event_filter(addon.create("GtkWidgetEventFilter", "next_state", "0.0.1.1.2.2.3", "clicked"))
+ st4 = deepcopy(st1)
+ st4.add_event_filter(addon.create("GtkWidgetEventFilter", "0.0.1.1.2.2.3", "clicked"), "next_state")
assert not (st1 == st4), "States having a different number of events should be different"
- st5 = copy.deepcopy(st1)
+ st5 = deepcopy(st1)
st5._event_filters = []
- st5.add_event_filter(addon.create("GtkWidgetEventFilter", "other_state", "0.1.2.3.4.1.2", "pressed"))
+ st5.add_event_filter(addon.create("GtkWidgetEventFilter", "0.1.2.3.4.1.2", "pressed"), "other_state")
- #import rpdb2; rpdb2.start_embedded_debugger('pass')
assert not (st1 == st5), "States having the same number of event filters" \
+ " but those being different should be different"
@@ -523,7 +522,7 @@ class FSMTest(unittest.TestCase):
fsm.add_action(act1)
- fsm2 = copy.deepcopy(fsm)
+ fsm2 = deepcopy(fsm)
assert fsm == fsm2
@@ -548,7 +547,7 @@ class FSMTest(unittest.TestCase):
fsm.add_state(st1)
fsm.add_state(st2)
- fsm4 = copy.deepcopy(fsm)
+ fsm4 = deepcopy(fsm)
assert fsm == fsm4
diff --git a/tests/filterstests.py b/tests/filterstests.py
index c45f924..ee6033b 100644
--- a/tests/filterstests.py
+++ b/tests/filterstests.py
@@ -32,20 +32,10 @@ from gtkutilstests import SignalCatcher
class BaseEventFilterTests(unittest.TestCase):
"""Test the behavior of the Base EventFilter class"""
- def test_properties(self):
- """Test EventFilter properties"""
- e = EventFilter("NEXTSTATE")
-
- assert e.next_state == "NEXTSTATE", "next_state should have value used in constructor"
-
- e.next_state = "NEWSTATE"
-
- assert e.next_state == "NEWSTATE", "next_state should have been changed by setter"
-
def test_callback(self):
"""Test the callback mechanism"""
- e = EventFilter("Next")
+ e = EventFilter()
s = SignalCatcher()
#Trigger the do_callback, shouldn't do anything
@@ -79,7 +69,7 @@ class TestTimerEvent(unittest.TestCase):
ctx = gobject.MainContext()
main = gobject.MainLoop(ctx)
- e = addon.create('TimerEvent', "Next", 2) # 2 seconds should be enough :s
+ e = addon.create('TimerEvent', 2) # 2 seconds should be enough :s
s = SignalCatcher()
e.install_handlers(s.callback)
@@ -122,7 +112,7 @@ class TestTimerEvent(unittest.TestCase):
ctx = gobject.MainContext()
main = gobject.MainLoop(ctx)
- e = addon.create('TimerEvent', "Next", 2) # 2 seconds should be enough :s
+ e = addon.create('TimerEvent', 2) # 2 seconds should be enough :s
s = SignalCatcher()
e.install_handlers(s.callback)
@@ -169,7 +159,7 @@ class TestGtkWidgetEventFilter(unittest.TestCase):
self.top.add(self.btn1)
def test_install(self):
- h = addon.create('GtkWidgetEventFilter', "Next","0","whatever")
+ h = addon.create('GtkWidgetEventFilter', "0","whatever")
try:
h.install_handlers(None)
@@ -178,7 +168,7 @@ class TestGtkWidgetEventFilter(unittest.TestCase):
assert True, "Install should have failed"
def test_button_clicks(self):
- h = addon.create('GtkWidgetEventFilter', "Next","0.0","clicked")
+ h = addon.create('GtkWidgetEventFilter', "0.0","clicked")
s = SignalCatcher()
h.install_handlers(s.callback, activity=self.top)
diff --git a/tests/probetests.py b/tests/probetests.py
index a440334..e1a587b 100644
--- a/tests/probetests.py
+++ b/tests/probetests.py
@@ -19,45 +19,482 @@ Probe Tests
"""
import unittest
-import os, sys
-import gtk
-import time
+import pickle
from dbus.mainloop.glib import DBusGMainLoop
+from dbus.mainloop import NULL_MAIN_LOOP
import dbus
-from sugar.tutorius.TProbe import TProbe, ProbeProxy
+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
-class FakeActivity(object):
- def __init__(self):
- self.top = gtk.Window(type=gtk.WINDOW_TOPLEVEL)
- self.top.set_name("Top")
+#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):
+ pass
+
+class MockProbeProxy(object):
+ _MockProxyCache = {}
+ def __new__(cls, activityName):
+ #For testing, use only one instance per activityName
+ return cls._MockProxyCache.setdefault(activityName, super(MockProbeProxy, cls).__new__(cls))
+
+ def __init__(self, activityName):
+ """
+ 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
- hbox = gtk.HBox()
- self.top.add(hbox)
- hbox.show()
+ 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
- btn1 = gtk.Button()
- btn1.set_name("Button1")
- hbox.pack_start(btn1)
- btn1.show()
- self.button = btn1
+ 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
+
+###########################################################################
+# Begin Test Cases
+###########################################################################
class ProbeTest(unittest.TestCase):
- def test_ping(self):
+ 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)
- activity = FakeActivity()
- probe = TProbe("localhost.unittest.ProbeTest", activity.top)
+ #Setup the activity and probe
+ self.activity = MockActivity()
+ self.probe = TProbe("localhost.unittest.ProbeTest", self.activity)
- #Parent, ping the probe
- proxy = ProbeProxy("localhost.unittest.ProbeTest")
- res = probe.ping()
-
+ #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_attach(self):
+ #ErrorCase: Set currentActivity to unattached activity
+ #Attempt to set to a non existing activity
+ try:
+ self.probeManager.currentActivity = "act1"
+ assert False, "Exception expected"
+ except RuntimeError, e:
+ pass
+
+ #Attach an activity
+ self.probeManager.attach("act1")
+
+ #Should have been created
+ assert "act1" in MockProbeProxy._MockProxyCache.keys(), "Proxy not created"
+
+ #ErrorCase: Attach multiple times to same activity
+ #Try to attach again
+ self.assertRaises(RuntimeWarning, self.probeManager.attach, "act1")
+
+ #Set current activity should work
+ self.probeManager.currentActivity = "act1"
+
+ #TODO Fill in the alive/notalive behavior at creation time once
+ # it is fixed in the ProbeManager
+
+ def test_detach(self):
+ #attach an activity
+ self.probeManager.attach("act1")
+ self.probeManager.currentActivity = "act1"
+ act1 = MockProbeProxy("act1")
+
+ #Now we detach
+ self.probeManager.detach("act1")
+ assert act1.MockAlive == False, "ProbeProxy should have been detached"
+ assert self.probeManager.currentActivity is None, "Current activity should be None"
+
+ #Attempt to detach again, should do nothing
+ #ErrorCase: detach already detached (currently silent fail)
+ self.probeManager.detach("act1")
+
+ #Now, attach 2 activities
+ self.probeManager.attach("act2")
+ self.probeManager.attach("act3")
+ act2 = MockProbeProxy("act2")
+ act3 = MockProbeProxy("act3")
+
+ self.probeManager.currentActivity = "act2"
+
+ assert act2.MockAlive and act3.MockAlive, "Both ProbeProxy instances should be alive"
+
+ #Detach the not active activity
+ self.probeManager.detach("act3")
+ #Check the statuses
+ assert act2.MockAlive and not act3.MockAlive, "Only act2 should be alive"
+ assert self.probeManager.currentActivity == "act2", "act2 should not have failed"
+
+ def test_actions(self):
+ self.probeManager.attach("act1")
+ self.probeManager.attach("act2")
+ act1 = MockProbeProxy("act1")
+ act2 = MockProbeProxy("act2")
+
+ 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.attach("act1")
+ self.probeManager.attach("act2")
+ act1 = MockProbeProxy("act1")
+ act2 = MockProbeProxy("act2")
+
+ 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")
+ self.probeProxy = ProbeProxy("unittest.TestCase")
+
+ 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"
+
+ #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()