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
|
#!/usr/bin/env python
"""Module that manipulates test state as a state machine
Note that this module has been superceded by the pluggable
test configuration framework and the apptest module.
Copyright (C) 2011, Digium, Inc.
Matt Jordan <mjordan@digium.com>
This program is free software, distributed under the terms of
the GNU General Public License Version 2.
"""
import logging
LOGGER = logging.getLogger(__name__)
def print_test_event(event):
"""Log a test event
Keyword Arguments:
event The TestEvent
"""
LOGGER.debug("Test Event received:")
for key, value in event.items():
LOGGER.debug("\t" + key + "\t=\t" + value)
class TestStateController(object):
"""The controller for the TestEvent state machine"""
def __init__(self, test_case, ami_receiver):
"""Constructor
Keyword Arguments:
test_case The TestCase derived class that owns this controller
ami_receiver The AMI instance that will send the controller TestEvent
notifications
"""
self._test_case = test_case
self._current_state = None
self._assert_handler = None
# Register for TestEvent updates
ami_receiver.registerEvent('TestEvent', self.handle_test_event)
def handle_test_event(self, ami, event):
"""Handler for a TestEvent
Keyword Arguments:
ami The AMI instance that sent us the TestEvent
event The TestEvent
"""
print_test_event(event)
if event['type'] == 'StateChange':
if (self._current_state != None):
self._current_state.handle_state_change(ami, event)
else:
LOGGER.error("No initial state set before TestEvent received")
self._current_state = FailureTestState(self)
elif event['type'] == 'Assert':
if (self._assert_handler != None):
self._assert_handler(ami, event)
else:
LOGGER.warn("ASSERT received but no handler defined; " \
"test will now fail")
self.fail_test()
def change_state(self, test_state):
"""Change the current state machine state to a new state
Keyword Arguments:
test_state The TestState to change to
"""
self._current_state = test_state
def fail_test(self):
"""
Fail and stop the test
"""
LOGGER.info("Setting test state to Fail")
self._test_case.passed = False
LOGGER.info("Stopping reactor")
self._test_case.stop_reactor()
def add_assert_handler(self, assert_handler_func):
"""Add an assert handler for Assert TestEvent types
Keyword Arguments:
assert_handler_func The handler function that takes in an AMI instance
and an event instance and receives the Asserts
Note that without a handler function, receiving any assert will
automatically fail a test
"""
self._assert_handler = assert_handler_func
class TestState(object):
"""Base class for the TestEvent state machine objects"""
def __init__(self, controller):
"""Constructor
Keyword Arguments:
controller The TestStateController instance
"""
self.controller = controller
if (self.controller == None):
LOGGER.error("Controller is none")
raise RuntimeError('Controller is none')
def handle_state_change(self, ami, event):
"""Handle a state change.
Called whenever a state change is received by the TestStateController.
Concrete implementations should override this method and use it to
change the state of the test by calling the change_state method
Keyword Arguments:
ami The instance of AMI that sent us the TestEvent
event The TestEvent object
"""
pass
def change_state(self, new_state):
"""Inform the TestStateController that the test state needs to change
Keyword Arguments:
new_state The new TestState to change to
"""
self.controller.change_state(new_state)
class FailureTestState(TestState):
"""A generic failure state.
Once transitioned to, the test will automatically fail. No further
state changes will be processed.
"""
def __init__(self, controller):
"""Constructor
Keyword Arguments:
controller The TestStateController instance
"""
super(FailureTestState, self).__init__(controller)
controller.fail_test()
def handle_state_change(self, ami, event):
"""Handle a state change.
This class ignores all subsequent state changes.
Keyword Arguments:
ami The instance of AMI that sent us the TestEvent
event The TestEvent object
"""
pass
def change_state(self, new_state):
"""Inform the TestStateController that the test state needs to change
This class ignores all changes of state.
Keyword Arguments:
new_state The new TestState to change to
"""
return
|
