# Copyright (C) 2009, Tutorius.org # Copyright (C) 2009, Erick Lavoie # # 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 """ Engine Tests Usage of actions and event filters is tested, but not the concrete actions 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) currentActivity = property(fget=lambda s:s, fset=lambda s, v: v) def install(self, action, action_installed_cb, error_cb): self.action = action 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 def uninstall(self, action_address): self.action = None 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) # Save the callbacks for this action self.event_sub_cB = event_sub_cb self._subscribe_error_cb = error_cb return str(event) def unsubscribe(self, address): self.event = None 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 """ def setUp(self): self.pM = MockProbeMgr() def tearDown(self): self.pM = None # Basic interface cases def testOneStateTutorial(self): tutorial = Tutorial("TutorialRunner") state_name = tutorial.add_state() tutorial.update_transition(Tutorial.INITIAL_TRANSITION_NAME, None, state_name) event = MockEvent() tutorial.add_transition(state_name, (event, Tutorial.END)) runner = TutorialRunner(tutorial, self.pM) runner.start() self.pM.event_sub_cB('event1') 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 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) # Limit cases def testEmptyTutorial(self): tutorial = Tutorial("TutorialRunner") runner = TutorialRunner(tutorial, self.pM) runner.start() assert runner._state == Tutorial.END, "Current state is: %s"%runner._state assert self.pM.action == None assert self.pM.event == None # Error cases if __name__ == "__main__": unittest.main()