#457

Committed 24 Aug 2024 09:15PM UTC coverage: 0.0% (-59.9%) from 59.947%

Build # #457

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 7c6dbce13 into c467d6f14

Run Details

0 of 4902 relevant lines covered (0.0%)

0.0 hits per line

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

0.0

/src/mbed_os_tools/test/host_tests/base_host_test.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 inspect
import six
from time import time
from inspect import isfunction, ismethod


class BaseHostTestAbstract(object):
    """ Base class for each host-test test cases with standard
        setup, test and teardown set of functions
    """

    name = ''                   # name of the host test (used for local registration)
    __event_queue = None        # To main even loop
    __dut_event_queue = None    # To DUT
    script_location = None      # Path to source file used to load host test
    __config = {}

    def __notify_prn(self, text):
        if self.__event_queue:
            self.__event_queue.put(('__notify_prn', text, time()))

    def __notify_conn_lost(self, text):
        if self.__event_queue:
            self.__event_queue.put(('__notify_conn_lost', text, time()))

    def __notify_sync_failed(self, text):
        if self.__event_queue:
            self.__event_queue.put(('__notify_sync_failed', text, time()))

    def __notify_dut(self, key, value):
        """! Send data over serial to DUT """
        if self.__dut_event_queue:
            self.__dut_event_queue.put((key, value, time()))

    def notify_complete(self, result=None):
        """! Notify main even loop that host test finished processing
        @param result True for success, False failure. If None - no action in main even loop
        """
        if self.__event_queue:
            self.__event_queue.put(('__notify_complete', result, time()))

    def reset_dut(self, value):
        """
        Reset device under test
        :return:
        """
        if self.__event_queue:
            self.__event_queue.put(('__reset_dut', value, time()))

    def reset(self):
        """
        Reset the device under test and continue running the host test
        :return:
        """
        if self.__event_queue:
            self.__event_queue.put(("__reset", "0", time()))

    def notify_conn_lost(self, text):
        """! Notify main even loop that there was a DUT-host test connection error
        @param consume If True htrun will process (consume) all remaining events
        """
        self.__notify_conn_lost(text)

    def log(self, text):
        """! Send log message to main event loop """
        self.__notify_prn(text)

    def send_kv(self, key, value):
        """! Send Key-Value data to DUT """
        self.__notify_dut(key, value)

    def setup_communication(self, event_queue, dut_event_queue, config={}):
        """! Setup queues used for IPC """
        self.__event_queue = event_queue         # To main even loop
        self.__dut_event_queue = dut_event_queue # To DUT
        self.__config = config

    def get_config_item(self, name):
        """
        Return test config

        :param name:
        :return:
        """
        return self.__config.get(name, None)

    def setup(self):
        """! Setup your tests and callbacks """
        raise NotImplementedError

    def result(self):
        """! Returns host test result (True, False or None) """
        raise NotImplementedError

    def teardown(self):
        """! Blocking always guaranteed test teardown """
        raise NotImplementedError


def event_callback(key):
    """
    Decorator for defining a event callback method. Adds a property attribute "event_key" with value as the passed key.

    :param key:
    :return:
    """
    def decorator(func):
        func.event_key = key
        return func
    return decorator


class HostTestCallbackBase(BaseHostTestAbstract):

    def __init__(self):
        BaseHostTestAbstract.__init__(self)
        self.__callbacks = {}
        self.__restricted_callbacks = [
            '__coverage_start',
            '__testcase_start',
            '__testcase_finish',
            '__testcase_summary',
            '__exit',
            '__exit_event_queue'
        ]

        self.__consume_by_default = [
            '__coverage_start',
            '__testcase_start',
            '__testcase_finish',
            '__testcase_count',
            '__testcase_name',
            '__testcase_summary',
            '__rxd_line',
        ]

        self.__assign_default_callbacks()
        self.__assign_decorated_callbacks()

    def __callback_default(self, key, value, timestamp):
        """! Default callback """
        #self.log("CALLBACK: key=%s, value=%s, timestamp=%f"% (key, value, timestamp))
        pass

    def __default_end_callback(self, key, value, timestamp):
        """
        Default handler for event 'end' that gives test result from target.
        This callback is not decorated as we don't know then in what order this
        callback would be registered. We want to let users over write this callback.
        Hence it should be registered before registering user defined callbacks.

        :param key:
        :param value:
        :param timestamp:
        :return:
        """
        self.notify_complete(value == 'success')

    def __assign_default_callbacks(self):
        """! Assigns default callback handlers """
        for key in self.__consume_by_default:
            self.__callbacks[key] = self.__callback_default
        # Register default handler for event 'end' before assigning user defined callbacks to let users over write it.
        self.register_callback('end', self.__default_end_callback)

    def __assign_decorated_callbacks(self):
        """
        It looks for any callback methods decorated with @event_callback

        Example:
        Define a method with @event_callback decorator like:

         @event_callback('<event key>')
         def event_handler(self, key, value, timestamp):
            do something..

        :return:
        """
        for name, method in inspect.getmembers(self, inspect.ismethod):
            key = getattr(method, 'event_key', None)
            if key:
                self.register_callback(key, method)

    def register_callback(self, key, callback, force=False):
        """! Register callback for a specific event (key: event name)
            @param key String with name of the event
            @param callback Callable which will be registstered for event "key"
            @param force God mode
        """

        # Non-string keys are not allowed
        if type(key) is not str:
            raise TypeError("event non-string keys are not allowed")

        # And finally callback should be callable
        if not callable(callback):
            raise TypeError("event callback should be callable")

        # Check if callback has all three required parameters (key, value, timestamp)
        # When callback is class method should have 4 arguments (self, key, value, timestamp)
        if ismethod(callback):
            arg_count = six.get_function_code(callback).co_argcount
            if arg_count != 4:
                err_msg = "callback 'self.%s('%s', ...)' defined with %d arguments"% (callback.__name__, key, arg_count)
                err_msg += ", should have 4 arguments: self.%s(self, key, value, timestamp)"% callback.__name__
                raise TypeError(err_msg)

        # When callback is just a function should have 3 arguments func(key, value, timestamp)
        if isfunction(callback):
            arg_count = six.get_function_code(callback).co_argcount
            if arg_count != 3:
                err_msg = "callback '%s('%s', ...)' defined with %d arguments"% (callback.__name__, key, arg_count)
                err_msg += ", should have 3 arguments: %s(key, value, timestamp)"% callback.__name__
                raise TypeError(err_msg)

        if not force:
            # Event starting with '__' are reserved
            if key.startswith('__'):
                raise ValueError("event key starting with '__' are reserved")

            # We predefined few callbacks you can't use
            if key in self.__restricted_callbacks:
                raise ValueError("we predefined few callbacks you can't use e.g. '%s'"% key)

        self.__callbacks[key] = callback

    def get_callbacks(self):
        return self.__callbacks

    def setup(self):
        pass

    def result(self):
        pass

    def teardown(self):
        pass


class BaseHostTest(HostTestCallbackBase):

    __BaseHostTest_Called = False

    def base_host_test_inited(self):
        """ This function will check if BaseHostTest ctor was called
            Call to BaseHostTest is required in order to force required
            interfaces implementation.
            @return Returns True if ctor was called (ok behaviour)
        """
        return self.__BaseHostTest_Called

    def __init__(self):
        HostTestCallbackBase.__init__(self)
        self.__BaseHostTest_Called = True

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

ARMmbed / mbed-os-tools / #457

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