Mbed Host Tests
base_host_test.py
Go to the documentation of this file.
1 """
2 mbed SDK
3 Copyright (c) 2011-2016 ARM Limited
4 
5 Licensed under the Apache License, Version 2.0 (the "License");
6 you may not use this file except in compliance with the License.
7 You may obtain a copy of the License at
8 
9  http://www.apache.org/licenses/LICENSE-2.0
10 
11 Unless required by applicable law or agreed to in writing, software
12 distributed under the License is distributed on an "AS IS" BASIS,
13 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 See the License for the specific language governing permissions and
15 limitations under the License.
16 """
17 
18 import inspect
19 import six
20 from time import time
21 from 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__event_queue:
37  self.__event_queue__event_queue.put(('__notify_prn', text, time()))
38 
39  def __notify_conn_lost(self, text):
40  if self.__event_queue__event_queue:
41  self.__event_queue__event_queue.put(('__notify_conn_lost', text, time()))
42 
43  def __notify_sync_failed(self, text):
44  if self.__event_queue__event_queue:
45  self.__event_queue__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__dut_event_queue:
50  self.__dut_event_queue__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__event_queue:
57  self.__event_queue__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__event_queue:
65  self.__event_queue__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__event_queue:
73  self.__event_queue__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__notify_conn_lost(text)
80 
81  def log(self, text):
82  """! Send log message to main event loop """
83  self.__notify_prn__notify_prn(text)
84 
85  def send_kv(self, key, value):
86  """! Send Key-Value data to DUT """
87  self.__notify_dut__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 = event_queue # To main even loop
92  self.__dut_event_queue__dut_event_queue = dut_event_queue # To DUT
93  self.__config__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__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 
117 def 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__callbacks = {}
135  self.__restricted_callbacks__restricted_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__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 
154  self.__assign_default_callbacks__assign_default_callbacks()
155  self.__assign_decorated_callbacks__assign_decorated_callbacks()
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__restricted_callbacks:
240  raise ValueError("we predefined few callbacks you can't use e.g. '%s'"% key)
241 
242  self.__callbacks__callbacks[key] = callback
243 
244  def get_callbacks(self):
245  return self.__callbacks__callbacks
246 
247  def setup(self):
248  pass
249 
250  def result(self):
251  pass
252 
253  def teardown(self):
254  pass
255 
256 
257 class 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  """
267  return self.__BaseHostTest_Called__BaseHostTest_Called__BaseHostTest_Called
268 
269  def __init__(self):
270  HostTestCallbackBase.__init__(self)
271  self.__BaseHostTest_Called__BaseHostTest_Called__BaseHostTest_Called = True
272 
def notify_conn_lost(self, text)
Notify main even loop that there was a DUT-host test connection error.
def send_kv(self, key, value)
Send Key-Value data to DUT.
def log(self, text)
Send log message to main event loop.
def result(self)
Returns host test result (True, False or None)
def setup_communication(self, event_queue, dut_event_queue, config={})
Setup queues used for IPC.
def notify_complete(self, result=None)
Notify main even loop that host test finished processing.
def teardown(self)
Blocking always guaranteed test teardown.
def register_callback(self, key, callback, force=False)
Register callback for a specific event (key: event name)
def teardown(self)
Blocking always guaranteed test teardown.
def result(self)
Returns host test result (True, False or None)