Mbed Host Tests
host_test_plugins.py
Go to the documentation of this file.
1 """
2 mbed SDK
3 Copyright (c) 2011-2015 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 Author: Przemyslaw Wirkus <Przemyslaw.Wirkus@arm.com>
18 """
19 
20 import os
21 import sys
22 import platform
23 import mbed_lstools
24 
25 from os import access, F_OK
26 from sys import stdout
27 from time import sleep
28 from subprocess import call
29 from mbed_host_tests.host_tests_logger import HtrunLogger
30 
31 
33  """! Base class for all plugins used with host tests
34  """
35 
38 
39 
42  name = "HostTestPluginBase" # Plugin name, can be plugin class name
43  type = "BasePlugin" # Plugin type: ResetMethod, CopyMethod etc.
44  capabilities = [] # Capabilities names: what plugin can achieve
45  # (e.g. reset using some external command line tool)
46  required_parameters = [] # Parameters required for 'kwargs' in plugin APIs: e.g. self.execute()
47  stable = False # Determine if plugin is stable and can be used
48 
49  def __init__(self):
50  """ ctor
51  """
52  # Setting Host Test Logger instance
53  ht_loggers = {
54  'BasePlugin' : HtrunLogger('PLGN'),
55  'CopyMethod' : HtrunLogger('COPY'),
56  'ResetMethod' : HtrunLogger('REST'),
57  }
58  self.plugin_loggerplugin_logger = ht_loggers.get(self.typetype, ht_loggers['BasePlugin'])
59 
60 
63 
64  def setup(self, *args, **kwargs):
65  """ Configure plugin, this function should be called before plugin execute() method is used.
66  """
67  return False
68 
69  def execute(self, capability, *args, **kwargs):
70  """! Executes capability by name
71  @param capability Capability name
72  @param args Additional arguments
73  @param kwargs Additional arguments
74  @details Each capability e.g. may directly just call some command line program or execute building pythonic function
75  @return Capability call return value
76  """
77  return False
78 
79  def is_os_supported(self, os_name=None):
80  """!
81  @return Returns true if plugin works (supportes) under certain OS
82  @os_name String describing OS.
83  See self.mbed_os_support() and self.mbed_os_info()
84  @details In some cases a plugin will not work under particular OS
85  mainly because command / software used to implement plugin
86  functionality is not available e.g. on MacOS or Linux.
87  """
88  return True
89 
90 
93  def print_plugin_error(self, text):
94  """! Function prints error in console and exits always with False
95  @param text Text to print
96  """
97  self.plugin_loggerplugin_logger.prn_err(text)
98  return False
99 
100  def print_plugin_info(self, text, NL=True):
101  """! Function prints notification in console and exits always with True
102  @param text Text to print
103  @param NL Deprecated! Newline will be added behind text if this flag is True
104  """
105 
106  self.plugin_loggerplugin_logger.prn_inf(text)
107  return True
108 
109  def print_plugin_char(self, char):
110  """ Function prints char on stdout
111  """
112  stdout.write(char)
113  stdout.flush()
114  return True
115 
116  def check_mount_point_ready(self, destination_disk, init_delay=0.2, loop_delay=0.25, target_id=None, timeout=60):
117  """! Waits until destination_disk is ready and can be accessed by e.g. copy commands
118  @return True if mount point was ready in given time, False otherwise
119  @param destination_disk Mount point (disk) which will be checked for readiness
120  @param init_delay - Initial delay time before first access check
121  @param loop_delay - polling delay for access check
122  @param timeout Mount point pooling timeout in seconds
123  """
124 
125  if target_id:
126  # Wait for mount point to appear with mbed-ls
127  # and if it does check if mount point for target_id changed
128  # If mount point changed, use new mount point and check if its ready (os.access)
129  new_destination_disk = destination_disk
130 
131  # Sometimes OSes take a long time to mount devices (up to one minute).
132  # Current pooling time: 120x 500ms = 1 minute
133  self.print_plugin_infoprint_plugin_info("Waiting up to %d sec for '%s' mount point (current is '%s')..."% (timeout, target_id, destination_disk))
134  timeout_step = 0.5
135  timeout = int(timeout / timeout_step)
136  for i in range(timeout):
137  # mbed_lstools.create() should be done inside the loop.
138  # Otherwise it will loop on same data.
139  mbeds = mbed_lstools.create()
140  mbed_list = mbeds.list_mbeds() #list of mbeds present
141  # get first item in list with a matching target_id, if present
142  mbed_target = next((x for x in mbed_list if x['target_id']==target_id), None)
143 
144  if mbed_target is not None:
145  # Only assign if mount point is present and known (not None)
146  if 'mount_point' in mbed_target and mbed_target['mount_point'] is not None:
147  new_destination_disk = mbed_target['mount_point']
148  break
149  sleep(timeout_step)
150 
151  if new_destination_disk != destination_disk:
152  # Mount point changed, update to new mount point from mbed-ls
153  self.print_plugin_infoprint_plugin_info("Mount point for '%s' changed from '%s' to '%s'..."% (target_id, destination_disk, new_destination_disk))
154  destination_disk = new_destination_disk
155 
156  result = True
157  # Check if mount point we've promoted to be valid one (by optional target_id check above)
158  # Let's wait for 30 * loop_delay + init_delay max
159  if not access(destination_disk, F_OK):
160  self.print_plugin_infoprint_plugin_info("Waiting for mount point '%s' to be ready..."% destination_disk, NL=False)
161  sleep(init_delay)
162  for i in range(30):
163  if access(destination_disk, F_OK):
164  result = True
165  break
166  sleep(loop_delay)
167  self.print_plugin_charprint_plugin_char('.')
168  else:
169  self.print_plugin_errorprint_plugin_error("mount {} is not accessible ...".format(destination_disk))
170  result = False
171  return (result, destination_disk)
172 
173  def check_serial_port_ready(self, serial_port, target_id=None, timeout=60):
174  """! Function checks (using mbed-ls) and updates serial port name information for DUT with specified target_id.
175  If no target_id is specified function returns old serial port name.
176  @param serial_port Current serial port name
177  @param target_id Target ID of a device under test which serial port will be checked and updated if needed
178  @param timeout Serial port pooling timeout in seconds
179  @return Tuple with result (always True) and serial port read from mbed-ls
180  """
181  # If serial port changed (check using mbed-ls), use new serial port
182  new_serial_port = None
183 
184  if target_id:
185  # Sometimes OSes take a long time to mount devices (up to one minute).
186  # Current pooling time: 120x 500ms = 1 minute
187  self.print_plugin_infoprint_plugin_info("Waiting up to %d sec for '%s' serial port (current is '%s')..."% (timeout, target_id, serial_port))
188  timeout_step = 0.5
189  timeout = int(timeout / timeout_step)
190  for i in range(timeout):
191  # mbed_lstools.create() should be done inside the loop. Otherwise it will loop on same data.
192  mbeds = mbed_lstools.create()
193  mbed_list = mbeds.list_mbeds() #list of mbeds present
194  # get first item in list with a matching target_id, if present
195  mbed_target = next((x for x in mbed_list if x['target_id']==target_id), None)
196 
197  if mbed_target is not None:
198  # Only assign if serial port is present and known (not None)
199  if 'serial_port' in mbed_target and mbed_target['serial_port'] is not None:
200  new_serial_port = mbed_target['serial_port']
201  if new_serial_port != serial_port:
202  # Serial port changed, update to new serial port from mbed-ls
203  self.print_plugin_infoprint_plugin_info("Serial port for tid='%s' changed from '%s' to '%s'..." % (target_id, serial_port, new_serial_port))
204  break
205  sleep(timeout_step)
206  else:
207  new_serial_port = serial_port
208 
209  return new_serial_port
210 
211  def check_parameters(self, capability, *args, **kwargs):
212  """! This function should be ran each time we call execute() to check if none of the required parameters is missing
213  @param capability Capability name
214  @param args Additional parameters
215  @param kwargs Additional parameters
216  @return Returns True if all parameters are passed to plugin, else return False
217  """
218  missing_parameters = []
219  for parameter in self.required_parametersrequired_parameters:
220  if parameter not in kwargs:
221  missing_parameters.append(parameter)
222  if len(missing_parameters):
223  self.print_plugin_errorprint_plugin_error("execute parameter(s) '%s' missing!"% (', '.join(missing_parameters)))
224  return False
225  return True
226 
227  def run_command(self, cmd, shell=True):
228  """! Runs command from command line.
229  @param cmd Command to execute
230  @param shell True if shell command should be executed (eg. ls, ps)
231  @details Function prints 'cmd' return code if execution failed
232  @return True if command successfully executed
233  """
234  result = True
235  try:
236  ret = call(cmd, shell=shell)
237  if ret:
238  self.print_plugin_errorprint_plugin_error("[ret=%d] Command: %s"% (int(ret), cmd))
239  return False
240  except Exception as e:
241  result = False
242  self.print_plugin_errorprint_plugin_error("[ret=%d] Command: %s"% (int(ret), cmd))
243  self.print_plugin_errorprint_plugin_error(str(e))
244  return result
245 
246  def mbed_os_info(self):
247  """! Returns information about host OS
248  @return Returns tuple with information about OS and host platform
249  """
250  result = (os.name,
251  platform.system(),
252  platform.release(),
253  platform.version(),
254  sys.platform)
255  return result
256 
257  def mbed_os_support(self):
258  """! Function used to determine host OS
259  @return Returns None if host OS is unknown, else string with name
260  @details This function should be ported for new OS support
261  """
262  result = None
263  os_info = self.mbed_os_infombed_os_info()
264  if (os_info[0] == 'nt' and os_info[1] == 'Windows'):
265  result = 'Windows7'
266  elif (os_info[0] == 'posix' and os_info[1] == 'Linux' and ('Ubuntu' in os_info[3])):
267  result = 'Ubuntu'
268  elif (os_info[0] == 'posix' and os_info[1] == 'Linux'):
269  result = 'LinuxGeneric'
270  elif (os_info[0] == 'posix' and os_info[1] == 'Darwin'):
271  result = 'Darwin'
272  return result
def print_plugin_info(self, text, NL=True)
Function prints notification in console and exits always with True.
def run_command(self, cmd, shell=True)
Runs command from command line.
def check_mount_point_ready(self, destination_disk, init_delay=0.2, loop_delay=0.25, target_id=None, timeout=60)
Waits until destination_disk is ready and can be accessed by e.g.
def execute(self, capability, *args, **kwargs)
Executes capability by name.
def print_plugin_error(self, text)
Interface helper methods - overload only if you need to have custom behaviour.
def check_serial_port_ready(self, serial_port, target_id=None, timeout=60)
Function checks (using mbed-ls) and updates serial port name information for DUT with specified targe...
def check_parameters(self, capability, *args, **kwargs)
This function should be ran each time we call execute() to check if none of the required parameters i...