#447

Committed 09 Jun 2024 08:56PM UTC coverage: 58.956% (-1.0%) from 59.947%

Build # #447

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge eddec8328 into c467d6f14

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

384 existing lines in 14 files now uncovered.

2890 of 4902 relevant lines covered (58.96%)

0.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.17

/src/mbed_os_tools/test/host_tests_plugins/host_test_plugins.py

# Copyright (c) 2018, Arm Limited and affiliates.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import sys
import platform

from os import access, F_OK
from sys import stdout
from time import sleep
from subprocess import call

from ... import detect
from ..host_tests_logger import HtrunLogger


class HostTestPluginBase:
    """! Base class for all plugins used with host tests
    """
    ###########################################################################
    # Interface:
    ###########################################################################

    ###########################################################################
    # Interface attributes defining plugin name, type etc.
    ###########################################################################
    name = "HostTestPluginBase" # Plugin name, can be plugin class name
    type = "BasePlugin"         # Plugin type: ResetMethod, CopyMethod etc.
    capabilities = []           # Capabilities names: what plugin can achieve
                                # (e.g. reset using some external command line tool)
    required_parameters = []    # Parameters required for 'kwargs' in plugin APIs: e.g. self.execute()
    stable = False              # Determine if plugin is stable and can be used

    def __init__(self):
        """ ctor
        """
        # Setting Host Test Logger instance
        ht_loggers = {
            'BasePlugin' : HtrunLogger('PLGN'),
            'CopyMethod' : HtrunLogger('COPY'),
            'ResetMethod' : HtrunLogger('REST'),
        }
        self.plugin_logger = ht_loggers.get(self.type, ht_loggers['BasePlugin'])

    ###########################################################################
    # Interface methods
    ###########################################################################

    def setup(self, *args, **kwargs):
        """ Configure plugin, this function should be called before plugin execute() method is used.
        """
        return False

    def execute(self, capability, *args, **kwargs):
        """! Executes capability by name
        @param capability Capability name
        @param args Additional arguments
        @param kwargs Additional arguments
        @details Each capability e.g. may directly just call some command line program or execute building pythonic function
        @return Capability call return value
        """
        return False

    def is_os_supported(self, os_name=None):
        """!
        @return Returns true if plugin works (supportes) under certain OS
        @os_name String describing OS.
                 See self.mbed_os_support() and self.mbed_os_info()
        @details In some cases a plugin will not work under particular OS
                 mainly because command / software used to implement plugin
                 functionality is not available e.g. on MacOS or Linux.
        """
        return True

    ###########################################################################
    # Interface helper methods - overload only if you need to have custom behaviour
    ###########################################################################
    def print_plugin_error(self, text):
        """! Function prints error in console and exits always with False
        @param text Text to print
        """
        self.plugin_logger.prn_err(text)
        return False

    def print_plugin_info(self, text, NL=True):
        """! Function prints notification in console and exits always with True
        @param text Text to print
        @param NL Deprecated! Newline will be added behind text if this flag is True
        """

        self.plugin_logger.prn_inf(text)
        return True

    def print_plugin_char(self, char):
        """ Function prints char on stdout
        """
        stdout.write(char)
        stdout.flush()
        return True

    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. copy commands
        @return True if mount point was ready in given time, False otherwise
        @param destination_disk Mount point (disk) which will be checked for readiness
        @param init_delay - Initial delay time before first access check
        @param loop_delay - polling delay for access check
        @param timeout Mount point pooling timeout in seconds
        """

        if target_id:
            # Wait for mount point to appear with mbed-ls
            # and if it does check if mount point for target_id changed
            # If mount point changed, use new mount point and check if its ready (os.access)
            new_destination_disk = destination_disk

            # Sometimes OSes take a long time to mount devices (up to one minute).
            # Current pooling time: 120x 500ms = 1 minute
            self.print_plugin_info("Waiting up to %d sec for '%s' mount point (current is '%s')..."% (timeout, target_id, destination_disk))
            timeout_step = 0.5
            timeout = int(timeout / timeout_step)
            for i in range(timeout):
                # mbed_os_tools.detect.create() should be done inside the loop.
                # Otherwise it will loop on same data.
                mbeds = detect.create()
                mbed_list = mbeds.list_mbeds() #list of mbeds present
                # get first item in list with a matching target_id, if present
                mbed_target = next((x for x in mbed_list if x['target_id']==target_id), None)

                if mbed_target is not None:
                    # Only assign if mount point is present and known (not None)
                    if 'mount_point' in mbed_target and mbed_target['mount_point'] is not None:
                        new_destination_disk = mbed_target['mount_point']
                        break
                sleep(timeout_step)

            if new_destination_disk != destination_disk:
                # Mount point changed, update to new mount point from mbed-ls
                self.print_plugin_info("Mount point for '%s' changed from '%s' to '%s'..."% (target_id, destination_disk, new_destination_disk))
                destination_disk = new_destination_disk

        result = True
        # Check if mount point we've promoted to be valid one (by optional target_id check above)
        # Let's wait for 30 * loop_delay + init_delay max
        if not access(destination_disk, F_OK):
            self.print_plugin_info("Waiting for mount point '%s' to be ready..."% destination_disk, NL=False)
            sleep(init_delay)
            for i in range(30):
                if access(destination_disk, F_OK):
                    result = True
                    break
                sleep(loop_delay)
                self.print_plugin_char('.')
            else:
                self.print_plugin_error("mount {} is not accessible ...".format(destination_disk))
                result = False
        return (result, destination_disk)

    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 target_id.
        If no target_id is specified function returns old serial port name.
        @param serial_port Current serial port name
        @param target_id Target ID of a device under test which serial port will be checked and updated if needed
        @param timeout Serial port pooling timeout in seconds
        @return Tuple with result (always True) and serial port read from mbed-ls
        """
        # If serial port changed (check using mbed-ls), use new serial port
        new_serial_port = None

        if target_id:
            # Sometimes OSes take a long time to mount devices (up to one minute).
            # Current pooling time: 120x 500ms = 1 minute
            self.print_plugin_info("Waiting up to %d sec for '%s' serial port (current is '%s')..."% (timeout, target_id, serial_port))
            timeout_step = 0.5
            timeout = int(timeout / timeout_step)
            for i in range(timeout):
                # mbed_os_tools.detect.create() should be done inside the loop. Otherwise it will loop on same data.
                mbeds = detect.create()
                mbed_list = mbeds.list_mbeds() #list of mbeds present
                # get first item in list with a matching target_id, if present
                mbed_target = next((x for x in mbed_list if x['target_id']==target_id), None)

                if mbed_target is not None:
                    # Only assign if serial port is present and known (not None)
                    if 'serial_port' in mbed_target and mbed_target['serial_port'] is not None:
                        new_serial_port = mbed_target['serial_port']
                        if new_serial_port != serial_port:
                            # Serial port changed, update to new serial port from mbed-ls
                            self.print_plugin_info("Serial port for tid='%s' changed from '%s' to '%s'..." % (target_id, serial_port, new_serial_port))
                        break
                sleep(timeout_step)
        else:
            new_serial_port = serial_port

        return new_serial_port

    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 is missing
        @param capability Capability name
        @param args Additional parameters
        @param kwargs Additional parameters
        @return Returns True if all parameters are passed to plugin, else return False
        """
        missing_parameters = []
        for parameter in self.required_parameters:
            if parameter not in kwargs:
                missing_parameters.append(parameter)
        if len(missing_parameters):
            self.print_plugin_error("execute parameter(s) '%s' missing!"% (', '.join(missing_parameters)))
            return False
        return True

    def run_command(self, cmd, shell=True, stdin = None):
        """! Runs command from command line.
        @param cmd Command to execute
        @param shell True if shell command should be executed (eg. ls, ps)
        @param stdin A custom stdin for the process running the command (defaults to None)
        @details Function prints 'cmd' return code if execution failed
        @return True if command successfully executed
        """
        result = True
        try:
            ret = call(cmd, shell=shell, stdin=stdin)
            if ret:
                self.print_plugin_error("[ret=%d] Command: %s"% (int(ret), cmd))
                return False
        except Exception as e:
            result = False
            self.print_plugin_error("[ret=%d] Command: %s"% (int(ret), cmd))
            self.print_plugin_error(str(e))
        return result

    def mbed_os_info(self):
        """! Returns information about host OS
        @return Returns tuple with information about OS and host platform
        """
        result = (os.name,
                  platform.system(),
                  platform.release(),
                  platform.version(),
                  sys.platform)
        return result

    def mbed_os_support(self):
        """! Function used to determine host OS
        @return Returns None if host OS is unknown, else string with name
        @details This function should be ported for new OS support
        """
        result = None
        os_info = self.mbed_os_info()
        if (os_info[0] == 'nt' and os_info[1] == 'Windows'):
            result = 'Windows7'
        elif (os_info[0] == 'posix' and os_info[1] == 'Linux' and ('Ubuntu' in os_info[3])):
            result = 'Ubuntu'
        elif (os_info[0] == 'posix' and os_info[1] == 'Linux'):
            result = 'LinuxGeneric'
        elif (os_info[0] == 'posix' and os_info[1] == 'Darwin'):
            result = 'Darwin'
        return result

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

ARMmbed / mbed-os-tools / #447

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous