Web   ·   Wiki   ·   Activities   ·   Blog   ·   Lists   ·   Chat   ·   Meeting   ·   Bugs   ·   Git   ·   Translate   ·   Archive   ·   People   ·   Donate
summaryrefslogtreecommitdiffstats
path: root/tests/enginetests.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/enginetests.py')
-rw-r--r--tests/enginetests.py165
1 files changed, 159 insertions, 6 deletions
diff --git a/tests/enginetests.py b/tests/enginetests.py
index a035c2f..ca86c9f 100644
--- a/tests/enginetests.py
+++ b/tests/enginetests.py
@@ -25,18 +25,74 @@ and event filters. Those are in their separate test module
"""
import unittest
+from functools import partial
+from uuid import uuid1
from sugar.tutorius.tutorial import Tutorial
from sugar.tutorius.engine import TutorialRunner
+import sugar.tutorius.engine as engine
+from sugar.tutorius.actions import Action
from sugar.tutorius.filters import EventFilter
from actiontests import CountAction
+class MockProbeMgrMultiAddons(object):
+ def __init__(self):
+ self.action_dict = {}
+ self.event_dict = {}
+ self.event_cb_dict = {}
+
+ self._action_installed_cb_list = []
+ self._install_error_cb_list = []
+ self._event_subscribed_cb_list = []
+ self._subscribe_error_cb_list = []
+
+ currentActivity = property(fget=lambda s:s, fset=lambda s, v: v)
+
+ def run_install_cb(self, action_number, action):
+ self._action_installed_cb_list[action_number](action, str(uuid1()))
+
+ def run_install_error_cb(self, action_number):
+ self._install_error_cb_list[action_number](Exception("Could not install action..."))
+
+ def run_subscribe_cb(self, event_number):
+ self._event_subscribed_cb_list[event_number](str(uuid1()))
+
+ def run_subscribe_error(self, event_number):
+ self._subscribe_error_cb_list[event_number](str(uuid1()))
+
+ def install(self, action, action_installed_cb, error_cb):
+ action_address = str(uuid1())
+ self.action_dict[action_address] = action
+ self._action_installed_cb_list.append(action_installed_cb)
+ self._install_error_cb_list.append(error_cb)
+
+ def update(self, action_address, new_action):
+ self.action_dict[action_address] = new_action
+
+ def uninstall(self, action_address):
+ del self.action_dict[action_address]
+
+ def subscribe(self, event_name, event, notif_cb, subscribe_cb, error_cb):
+ event_address = str(uuid1())
+ self.event_dict[event_name] = event_address
+ self.event_cb_dict[event_name] = notif_cb
+ self._event_subscribed_cb_list.append(subscribe_cb)
+ self._subscribe_error_cb_list.append(error_cb)
+
+ def unsubscribe(self, address):
+ for (event_name, other_event) in self.event_dict.values():
+ if event == othet_event:
+ del self.event_dict[event_name]
+ break
class MockProbeMgr(object):
def __init__(self):
self.action = None
self.event = None
self.cB = None
+
+ self._action_installed_cb = None
+ self._install_error_cb = None
def doCB(self):
self.cB(self.event)
@@ -45,7 +101,8 @@ class MockProbeMgr(object):
def install(self, action, action_installed_cb, error_cb):
self.action = action
- action_installed_cb(action, 'Action1')
+ self._action_installed_cb = partial(action_installed_cb, action)
+ self._install_error_cb = partial(error_cb, action)
def update(self, action_address, newaction):
self.action = newaction
@@ -53,11 +110,11 @@ class MockProbeMgr(object):
def uninstall(self, action_address):
self.action = None
- def subscribe(self, event, notif_cb, event_sub_cb, error_cb):
+ def subscribe(self, event_name, event, notif_cb, event_sub_cb, error_cb):
self.event = event
self.cB = notif_cb
self.event.install_handlers(notif_cb)
- # Trigger the installation callback
+ # Save the callbacks for this action
self.event_sub_cB = event_sub_cb
self._subscribe_error_cb = error_cb
return str(event)
@@ -67,9 +124,105 @@ class MockProbeMgr(object):
class MockEvent(EventFilter):
pass
+
+class TestRunnerStates(unittest.TestCase):
+ def setUp(self):
+ self.pM = MockProbeMgr()
+ self.tutorial = Tutorial("TutorialRunner")
+ self.state_name = self.tutorial.add_state()
+ self.tutorial.update_transition(Tutorial.INITIAL_TRANSITION_NAME,
+ None, self.state_name)
+ self.action = CountAction()
+ self.tutorial.add_action(self.state_name, self.action)
+ self.event = MockEvent()
+ self.tutorial.add_transition(self.state_name, (self.event, Tutorial.END))
+
+ self.runner = TutorialRunner(self.tutorial, self.pM)
+
+ def test_setup_states(self):
+ assert self.runner._runner_state == engine.RUNNER_STATE_IDLE, "Idle should be the initial state for the runner"
+
+ self.runner.start()
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_ACTIONS, "Setup Actions State should be entered after start"
+ self.pM._action_installed_cb('action1')
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_EVENTS, "State should be Setup Events after all actions are installed"
+
+ self.pM.event_sub_cB('event1')
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_AWAITING_NOTIFICATIONS, "State should be Awaiting Notifications once all events are installed"
+
+ def test_setup_actions_errors(self):
+ self.runner.start()
+
+ self.pM._install_error_cb(Exception("Fake Exception"))
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_EVENTS, "Setup Events should be reached after error on action installation"
+
+ self.pM._subscribe_error_cb(Exception("Fake Exception"))
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_AWAITING_NOTIFICATIONS, "State Awaiting Notifications should be reached after event subscribe error"
+
+ def test_stop_in_actions(self):
+ self.runner.start()
+
+ self.runner.stop()
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_ACTIONS, "Stop state should not be reached"
+
+ self.pM._action_installed_cb('action1')
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_STOPPED
+
+ def test_stop_in_events(self):
+ self.runner.start()
+ self.pM._action_installed_cb('action1')
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_EVENTS, "Setup events state should be reached after all actions installed"
+
+ self.runner.stop()
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_EVENTS, "Tutorial should not be stopped until all events have been confirmed"
+ self.pM.event_sub_cB('event1')
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_STOPPED, "Tutorial should have been stopped right after the last event was confirmed"
+
+class TestInstallationStates(unittest.TestCase):
+ def setUp(self):
+ self.pM = MockProbeMgrMultiAddons()
+ self.tutorial = Tutorial("TutorialRunner")
+ self.state_name = self.tutorial.add_state()
+ self.tutorial.update_transition(Tutorial.INITIAL_TRANSITION_NAME,
+ None, self.state_name)
+ #import rpdb2; rpdb2.start_embedded_debugger('pass')
+ self.action1 = CountAction()
+ self.tutorial.add_action(self.state_name, self.action1)
+ self.action2 = CountAction()
+ self.tutorial.add_action(self.state_name, self.action2)
+
+ self.event = MockEvent()
+ self.tutorial.add_transition(self.state_name, (self.event, Tutorial.END))
+ self.event2 = MockEvent()
+ self.tutorial.add_transition(self.state_name, (self.event2, Tutorial.INIT))
+
+ self.runner = TutorialRunner(self.tutorial, self.pM)
+
+ def test_multiple_actions(self):
+ self.runner.start()
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_ACTIONS, "Runner should be in Setup Actions state"
+
+ self.pM.run_install_cb(1, self.action2)
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_ACTIONS, "Runner should still be in Setup Actions state after a single action confirmation callback"
+
+ self.pM.run_install_cb(0, self.action1)
+
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_EVENTS, "Runner should be in Setup Events state after all actions are installed"
+ self.pM.run_subscribe_cb(1)
+ assert self.runner._runner_state == engine.RUNNER_STATE_SETUP_EVENTS, "Runner should still be in Setup Events state when not all event installations are confirmed"
-
class TutorialRunnerTest(unittest.TestCase):
"""
This class needs to test the TutorialRunner
@@ -94,13 +247,13 @@ class TutorialRunnerTest(unittest.TestCase):
runner.start()
self.pM.event_sub_cB('event1')
- assert runner._state == state_name, "Current state is: %s"%runner._state
+ assert runner._state == state_name, "Current tutorial state is: %s"%runner._state
assert self.pM.action == None
assert self.pM.event == event
event.do_callback()
- assert runner._state == Tutorial.END, "Current state is: %s"%runner._state
+ assert runner._state == Tutorial.END, "Current tutorial state is: %s"%runner._state
assert self.pM.action == None, "Current action is %s"%str(self.pM.action)
assert self.pM.event == None, "Current event is %s"%str(self.pM.event)