Mbed Host Tests
base_host_test.py
Go to the documentation of this file.
1"""
2mbed SDK
3Copyright (c) 2011-2016 ARM Limited
4
5Licensed under the Apache License, Version 2.0 (the "License");
6you may not use this file except in compliance with the License.
7You may obtain a copy of the License at
8
9 http://www.apache.org/licenses/LICENSE-2.0
10
11Unless required by applicable law or agreed to in writing, software
12distributed under the License is distributed on an "AS IS" BASIS,
13WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14See the License for the specific language governing permissions and
15limitations under the License.
16"""
17
18import inspect
19import six
20from time import time
21from inspect import isfunction, ismethod
22
23
25 """ Base class for each host-test test cases with standard
26 setup, test and teardown set of functions
27 """
28
29 name = '' # name of the host test (used for local registration)
30 __event_queue = None # To main even loop
31 __dut_event_queue = None # To DUT
32 script_location = None # Path to source file used to load host test
33 __config = {}
34
35 def __notify_prn(self, text):
36 if self.__event_queue:
37 self.__event_queue.put(('__notify_prn', text, time()))
38
39 def __notify_conn_lost(self, text):
40 if self.__event_queue:
41 self.__event_queue.put(('__notify_conn_lost', text, time()))
42
43 def __notify_sync_failed(self, text):
44 if self.__event_queue:
45 self.__event_queue.put(('__notify_sync_failed', text, time()))
46
47 def __notify_dut(self, key, value):
48 """! Send data over serial to DUT """
49 if self.__dut_event_queue:
50 self.__dut_event_queue.put((key, value, time()))
51
52 def notify_complete(self, result=None):
53 """! Notify main even loop that host test finished processing
54 @param result True for success, False failure. If None - no action in main even loop
55 """
56 if self.__event_queue:
57 self.__event_queue.put(('__notify_complete', result, time()))
58
59 def reset_dut(self, value):
60 """
61 Reset device under test
62 :return:
63 """
64 if self.__event_queue:
65 self.__event_queue.put(('__reset_dut', value, time()))
66
67 def reset(self):
68 """
69 Reset the device under test and continue running the host test
70 :return:
71 """
72 if self.__event_queue:
73 self.__event_queue.put(("__reset", "0", time()))
74
75 def notify_conn_lost(self, text):
76 """! Notify main even loop that there was a DUT-host test connection error
77 @param consume If True htrun will process (consume) all remaining events
78 """
79 self.__notify_conn_lost(text)
80
81 def log(self, text):
82 """! Send log message to main event loop """
83 self.__notify_prn(text)
84
85 def send_kv(self, key, value):
86 """! Send Key-Value data to DUT """
87 self.__notify_dut(key, value)
88
89 def setup_communication(self, event_queue, dut_event_queue, config={}):
90 """! Setup queues used for IPC """
91 self.__event_queue = event_queue # To main even loop
92 self.__dut_event_queue = dut_event_queue # To DUT
93 self.__config__config = config
94
95 def get_config_item(self, name):
96 """
97 Return test config
98
99 :param name:
100 :return:
101 """
102 return self.__config__config.get(name, None)
103
104 def setup(self):
105 """! Setup your tests and callbacks """
106 raise NotImplementedError
107
108 def result(self):
109 """! Returns host test result (True, False or None) """
110 raise NotImplementedError
111
112 def teardown(self):
113 """! Blocking always guaranteed test teardown """
114 raise NotImplementedError
115
116
117def event_callback(key):
118 """
119 Decorator for defining a event callback method. Adds a property attribute "event_key" with value as the passed key.
120
121 :param key:
122 :return:
123 """
124 def decorator(func):
125 func.event_key = key
126 return func
127 return decorator
128
129
131
132 def __init__(self):
133 BaseHostTestAbstract.__init__(self)
134 self.__callbacks = {}
136 '__coverage_start',
137 '__testcase_start',
138 '__testcase_finish',
139 '__testcase_summary',
140 '__exit',
141 '__exit_event_queue'
142 ]
143
144 self.__consume_by_default = [
145 '__coverage_start',
146 '__testcase_start',
147 '__testcase_finish',
148 '__testcase_count',
149 '__testcase_name',
150 '__testcase_summary',
151 '__rxd_line',
152 ]
153
156
157 def __callback_default(self, key, value, timestamp):
158 """! Default callback """
159 #self.log("CALLBACK: key=%s, value=%s, timestamp=%f"% (key, value, timestamp))
160 pass
161
162 def __default_end_callback(self, key, value, timestamp):
163 """
164 Default handler for event 'end' that gives test result from target.
165 This callback is not decorated as we don't know then in what order this
166 callback would be registered. We want to let users over write this callback.
167 Hence it should be registered before registering user defined callbacks.
168
169 :param key:
170 :param value:
171 :param timestamp:
172 :return:
173 """
174 self.notify_complete(value == 'success')
175
176 def __assign_default_callbacks(self):
177 """! Assigns default callback handlers """
178 for key in self.__consume_by_default:
179 self.__callbacks[key] = self.__callback_default
180 # Register default handler for event 'end' before assigning user defined callbacks to let users over write it.
181 self.register_callback('end', self.__default_end_callback)
182
183 def __assign_decorated_callbacks(self):
184 """
185 It looks for any callback methods decorated with @event_callback
186
187 Example:
188 Define a method with @event_callback decorator like:
189
190 @event_callback('<event key>')
191 def event_handler(self, key, value, timestamp):
192 do something..
193
194 :return:
195 """
196 for name, method in inspect.getmembers(self, inspect.ismethod):
197 key = getattr(method, 'event_key', None)
198 if key:
199 self.register_callback(key, method)
200
201 def register_callback(self, key, callback, force=False):
202 """! Register callback for a specific event (key: event name)
203 @param key String with name of the event
204 @param callback Callable which will be registstered for event "key"
205 @param force God mode
206 """
207
208 # Non-string keys are not allowed
209 if type(key) is not str:
210 raise TypeError("event non-string keys are not allowed")
211
212 # And finally callback should be callable
213 if not callable(callback):
214 raise TypeError("event callback should be callable")
215
216 # Check if callback has all three required parameters (key, value, timestamp)
217 # When callback is class method should have 4 arguments (self, key, value, timestamp)
218 if ismethod(callback):
219 arg_count = six.get_function_code(callback).co_argcount
220 if arg_count != 4:
221 err_msg = "callback 'self.%s('%s', ...)' defined with %d arguments"% (callback.__name__, key, arg_count)
222 err_msg += ", should have 4 arguments: self.%s(self, key, value, timestamp)"% callback.__name__
223 raise TypeError(err_msg)
224
225 # When callback is just a function should have 3 arguments func(key, value, timestamp)
226 if isfunction(callback):
227 arg_count = six.get_function_code(callback).co_argcount
228 if arg_count != 3:
229 err_msg = "callback '%s('%s', ...)' defined with %d arguments"% (callback.__name__, key, arg_count)
230 err_msg += ", should have 3 arguments: %s(key, value, timestamp)"% callback.__name__
231 raise TypeError(err_msg)
232
233 if not force:
234 # Event starting with '__' are reserved
235 if key.startswith('__'):
236 raise ValueError("event key starting with '__' are reserved")
237
238 # We predefined few callbacks you can't use
239 if key in self.__restricted_callbacks:
240 raise ValueError("we predefined few callbacks you can't use e.g. '%s'"% key)
241
242 self.__callbacks[key] = callback
243
244 def get_callbacks(self):
245 return self.__callbacks
246
247 def setup(self):
248 pass
249
250 def result(self):
251 pass
252
253 def teardown(self):
254 pass
255
256
257class BaseHostTest(HostTestCallbackBase):
258
259 __BaseHostTest_Called = False
260
262 """ This function will check if BaseHostTest ctor was called
263 Call to BaseHostTest is required in order to force required
264 interfaces implementation.
265 @return Returns True if ctor was called (ok behaviour)
266 """
268
269 def __init__(self):
270 HostTestCallbackBase.__init__(self)
272
send_kv(self, key, value)
Send Key-Value data to DUT.
result(self)
Returns host test result (True, False or None)
notify_complete(self, result=None)
Notify main even loop that host test finished processing.
teardown(self)
Blocking always guaranteed test teardown.
log(self, text)
Send log message to main event loop.
notify_conn_lost(self, text)
Notify main even loop that there was a DUT-host test connection error.
setup_communication(self, event_queue, dut_event_queue, config={})
Setup queues used for IPC.
result(self)
Returns host test result (True, False or None)
teardown(self)
Blocking always guaranteed test teardown.
register_callback(self, key, callback, force=False)
Register callback for a specific event (key: event name)