Web   ·   Wiki   ·   Activities   ·   Blog   ·   Lists   ·   Chat   ·   Meeting   ·   Bugs   ·   Git   ·   Translate   ·   Archive   ·   People   ·   Donate
summaryrefslogtreecommitdiffstats
path: root/tests/enginetests.py
blob: 4a8a3ca6cd3731d0ddd31605ef7a3055a7318c19 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
# Copyright (C) 2009, Tutorius.org
# Copyright (C) 2009, Erick Lavoie <erick.lavoie@gmail.com> 
#
# 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 MockProbeMgrMultiAddons(object):
    """ 
    Mock probe manager that supports installing more than one action
    at the time.
    """
    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_IDLE

    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_IDLE, "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()