Mbed Host Tests
host_test_plugins.py
Go to the documentation of this file.
1"""
2mbed SDK
3Copyright (c) 2011-2015 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
17Author: Przemyslaw Wirkus <Przemyslaw.Wirkus@arm.com>
18"""
19
20import os
21import sys
22import platform
23import mbed_lstools
24
25from os import access, F_OK
26from sys import stdout
27from time import sleep
28from subprocess import call
29from 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_logger = ht_loggers.get(self.type, 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_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_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_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_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_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_char('.')
168 else:
169 self.print_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_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_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_parameters:
220 if parameter not in kwargs:
221 missing_parameters.append(parameter)
222 if len(missing_parameters):
223 self.print_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_error("[ret=%d] Command: %s"% (int(ret), cmd))
239 return False
240 except Exception as e:
241 result = False
242 self.print_plugin_error("[ret=%d] Command: %s"% (int(ret), cmd))
243 self.print_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
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_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
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...
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.
print_plugin_error(self, text)
Interface helper methods - overload only if you need to have custom behaviour.
run_command(self, cmd, shell=True)
Runs command from command line.
execute(self, capability, *args, **kwargs)
Executes capability by name.
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...
print_plugin_info(self, text, NL=True)
Function prints notification in console and exits always with True.
print_plugin_info()
Prints plugins' information in user friendly way.
Definition __init__.py:97