#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_conn_proxy/conn_proxy.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 re
import sys
import uuid
from time import time
from ..host_tests_logger import HtrunLogger
from .conn_primitive_serial import SerialConnectorPrimitive
from .conn_primitive_remote import RemoteConnectorPrimitive
from .conn_primitive_fastmodel import FastmodelConnectorPrimitive

if (sys.version_info > (3, 0)):
    from queue import Empty as QueueEmpty # Queue here refers to the module, not a class
else:
    from Queue import Empty as QueueEmpty

class KiViBufferWalker():
    """! Simple auxiliary class used to walk through a buffer and search for KV tokens """
    def __init__(self):
        self.KIVI_REGEX = r"\{\{([\w\d_-]+);([^\}]+)\}\}"
        self.buff = str()
        self.kvl = []
        self.re_kv = re.compile(self.KIVI_REGEX)

    def append(self, payload):
        """! Append stream buffer with payload and process. Returns non-KV strings"""
        logger = HtrunLogger('CONN')
        try:
            self.buff += payload.decode('utf-8')
        except UnicodeDecodeError:
            logger.prn_wrn("UnicodeDecodeError encountered!")
            self.buff += payload.decode('utf-8','ignore')
        lines = self.buff.split('\n')
        self.buff = lines[-1]   # remaining
        lines.pop(-1)
        # List of line or strings that did not match K,V pair.
        discarded = []

        for line in lines:
            m = self.re_kv.search(line)
            if m:
                (key, value) = m.groups()
                self.kvl.append((key, value, time()))
                line = line.strip()
                match = m.group(0)
                pos = line.find(match)
                before = line[:pos]
                after = line[pos + len(match):]
                if len(before) > 0:
                    discarded.append(before)
                if len(after) > 0:
                    # not a K,V pair part
                    discarded.append(after)
            else:
                # not a K,V pair
                discarded.append(line)
        return discarded

    def search(self):
        """! Check if there is a KV value in buffer """
        return len(self.kvl) > 0

    def pop_kv(self):
        if len(self.kvl):
            return self.kvl.pop(0)
        return None, None, time()


def conn_primitive_factory(conn_resource, config, event_queue, logger):
    """! Factory producing connectors based on type and config
    @param conn_resource Name of connection primitive (e.g. 'serial' for
           local serial port connection or 'grm' for global resource manager)
    @param event_queue Even queue of Key-Value protocol
    @param config Global configuration for connection process
    @param logger Host Test logger instance
    @return Object of type <ConnectorPrimitive> or None if type of connection primitive unknown (conn_resource)
    """
    polling_timeout = int(config.get('polling_timeout', 60))
    logger.prn_inf("notify event queue about extra %d sec timeout for serial port pooling"%polling_timeout)
    event_queue.put(('__timeout', polling_timeout, time()))

    if conn_resource == 'serial':
        # Standard serial port connection
        # Notify event queue we will wait additional time for serial port to be ready

        # Get extra configuration related to serial port
        port = config.get('port')
        baudrate = config.get('baudrate')

        logger.prn_inf("initializing serial port listener... ")
        connector = SerialConnectorPrimitive(
            'SERI',
            port,
            baudrate,
            config=config)
    elif conn_resource == 'grm':
        # Start GRM (Gloabal Resource Mgr) collection
        logger.prn_inf("initializing global resource mgr listener... ")
        connector = RemoteConnectorPrimitive('GLRM', config=config)
    elif conn_resource == 'fmc':
        # Start Fast Model Connection collection
        logger.prn_inf("initializing fast model connection")
        connector = FastmodelConnectorPrimitive('FSMD', config=config)
    else:
        logger.pn_err("unknown connection resource!")
        raise NotImplementedError("ConnectorPrimitive factory: unknown connection resource '%s'!"% conn_resource)

    return connector


def conn_process(event_queue, dut_event_queue, config):

    def __notify_conn_lost():
        error_msg = connector.error()
        connector.finish()
        event_queue.put(('__notify_conn_lost', error_msg, time()))

    def __notify_sync_failed():
        error_msg = connector.error()
        connector.finish()
        event_queue.put(('__notify_sync_failed', error_msg, time()))

    logger = HtrunLogger('CONN')
    logger.prn_inf("starting connection process...")

    # Send connection process start event to host process
    # NOTE: Do not send any other Key-Value pairs before this!
    event_queue.put(('__conn_process_start', 1, time()))

    # Configuration of conn_opriocess behaviour
    sync_behavior = int(config.get('sync_behavior', 1))
    sync_timeout = config.get('sync_timeout', 1.0)
    conn_resource = config.get('conn_resource', 'serial')
    last_sync = False

    # Create connector instance with proper configuration
    connector = conn_primitive_factory(conn_resource, config, event_queue, logger)

    # If the connector failed, stop the process now
    if not connector.connected():
        logger.prn_err("Failed to connect to resource")
        __notify_conn_lost()
        return 0

    # Create simple buffer we will use for Key-Value protocol data
    kv_buffer = KiViBufferWalker()

    # List of all sent to target UUIDs (if multiple found)
    sync_uuid_list = []

    # We will ignore all kv pairs before we get sync back
    sync_uuid_discovered = False

    def __send_sync(timeout=None):
        sync_uuid = str(uuid.uuid4())
        # Handshake, we will send {{sync;UUID}} preamble and wait for mirrored reply
        if timeout:
            logger.prn_inf("Reset the part and send in new preamble...")
            connector.reset()
            logger.prn_inf("resending new preamble '%s' after %0.2f sec"% (sync_uuid, timeout))
        else:
            logger.prn_inf("sending preamble '%s'"% sync_uuid)

        if connector.write_kv('__sync', sync_uuid):
            return sync_uuid
        else:
            return None

    # Send simple string to device to 'wake up' greentea-client k-v parser
    if not connector.write("mbed" * 10, log=True):
        # Failed to write 'wake up' string, exit conn_process
        __notify_conn_lost()
        return 0


    # Sync packet management allows us to manipulate the way htrun sends __sync packet(s)
    # With current settings we can force on htrun to send __sync packets in this manner:
    #
    # * --sync=0        - No sync packets will be sent to target platform
    # * --sync=-10      - __sync packets will be sent unless we will reach
    #                     timeout or proper response is sent from target platform
    # * --sync=N        - Send up to N __sync packets to target platform. Response
    #                     is sent unless we get response from target platform or
    #                     timeout occur

    if sync_behavior > 0:
        # Sending up to 'n' __sync packets
        logger.prn_inf("sending up to %s __sync packets (specified with --sync=%s)"% (sync_behavior, sync_behavior))
        sync_uuid = __send_sync()

        if sync_uuid:
            sync_uuid_list.append(sync_uuid)
            sync_behavior -= 1
        else:
            __notify_conn_lost()
            return 0
    elif sync_behavior == 0:
        # No __sync packets
        logger.prn_wrn("skipping __sync packet (specified with --sync=%s)"% sync_behavior)
    else:
        # Send __sync until we go reply
        logger.prn_inf("sending multiple __sync packets (specified with --sync=%s)"% sync_behavior)

        sync_uuid = __send_sync()
        if sync_uuid:
            sync_uuid_list.append(sync_uuid)
            sync_behavior -= 1
        else:
            __notify_conn_lost()
            return 0

    loop_timer = time()
    while True:

        # Check if connection is lost to serial
        if not connector.connected():
            __notify_conn_lost()
            break

        # Send data to DUT
        try:
            (key, value, _) = dut_event_queue.get(block=False)
        except QueueEmpty:
            pass # Check if target sent something
        else:
            # Return if state machine in host_test_default has finished to end process
            if key == '__host_test_finished' and value == True:
                logger.prn_inf("received special event '%s' value='%s', finishing"% (key, value))
                connector.finish()
                return 0
            elif key == '__reset':
                logger.prn_inf("received special event '%s', resetting dut" % (key))
                connector.reset()
                event_queue.put(("reset_complete", 0, time()))
            elif not connector.write_kv(key, value):
                connector.write_kv(key, value)
                __notify_conn_lost()
                break

        # Since read is done every 0.2 sec, with maximum baud rate we can receive 2304 bytes in one read in worst case.
        data = connector.read(2304)
        if data:
            # Stream data stream KV parsing
            print_lines = kv_buffer.append(data)
            for line in print_lines:
                logger.prn_rxd(line)
                event_queue.put(('__rxd_line', line, time()))
            while kv_buffer.search():
                key, value, timestamp = kv_buffer.pop_kv()

                if sync_uuid_discovered:
                    event_queue.put((key, value, timestamp))
                    logger.prn_inf("found KV pair in stream: {{%s;%s}}, queued..."% (key, value))
                else:
                    if key == '__sync':
                        if value in sync_uuid_list:
                            sync_uuid_discovered = True
                            event_queue.put((key, value, time()))
                            idx = sync_uuid_list.index(value)
                            logger.prn_inf("found SYNC in stream: {{%s;%s}} it is #%d sent, queued..."% (key, value, idx))
                        else:
                            logger.prn_err("found faulty SYNC in stream: {{%s;%s}}, ignored..."% (key, value))
                            logger.prn_inf("Resetting the part and sync timeout to clear out the buffer...")
                            connector.reset()
                            loop_timer = time()
                    else:
                        logger.prn_wrn("found KV pair in stream: {{%s;%s}}, ignoring..."% (key, value))

        if not sync_uuid_discovered:
            # Resending __sync after 'sync_timeout' secs (default 1 sec)
            # to target platform. If 'sync_behavior' counter is != 0 we
            # will continue to send __sync packets to target platform.
            # If we specify 'sync_behavior' < 0 we will send 'forever'
            # (or until we get reply)

            if sync_behavior != 0:
                time_to_sync_again = time() - loop_timer
                if time_to_sync_again > sync_timeout:
                    sync_uuid = __send_sync(timeout=time_to_sync_again)

                    if sync_uuid:
                        sync_uuid_list.append(sync_uuid)
                        sync_behavior -= 1
                        loop_timer = time()
                        #Sync behavior will be zero and if last sync fails we should report connection
                        #lost
                        if sync_behavior == 0:
                            last_sync = True
                    else:
                        __notify_conn_lost()
                        break
            elif last_sync == True:
                #SYNC lost connection event : Device not responding, send sync failed
                __notify_sync_failed()
                break

    return 0

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 re	×
17	import sys	×
18	import uuid	×
19	from time import time	×
20	from ..host_tests_logger import HtrunLogger	×
21	from .conn_primitive_serial import SerialConnectorPrimitive	×
22	from .conn_primitive_remote import RemoteConnectorPrimitive	×
23	from .conn_primitive_fastmodel import FastmodelConnectorPrimitive	×
24
25	if (sys.version_info > (3, 0)):	×
26	from queue import Empty as QueueEmpty # Queue here refers to the module, not a class	×
27	else:
28	from Queue import Empty as QueueEmpty	×
29
30	class KiViBufferWalker():	×
31	"""! Simple auxiliary class used to walk through a buffer and search for KV tokens """
32	def __init__(self):	×
33	self.KIVI_REGEX = r"\{\{([\w\d_-]+);([^\}]+)\}\}"	×
34	self.buff = str()	×
35	self.kvl = []	×
36	self.re_kv = re.compile(self.KIVI_REGEX)	×
37
38	def append(self, payload):	×
39	"""! Append stream buffer with payload and process. Returns non-KV strings"""
40	logger = HtrunLogger('CONN')	×
41	try:	×
42	self.buff += payload.decode('utf-8')	×
43	except UnicodeDecodeError:	×
44	logger.prn_wrn("UnicodeDecodeError encountered!")	×
45	self.buff += payload.decode('utf-8','ignore')	×
46	lines = self.buff.split('\n')	×
47	self.buff = lines[-1] # remaining	×
48	lines.pop(-1)	×
49	# List of line or strings that did not match K,V pair.
50	discarded = []	×
51
52	for line in lines:	×
53	m = self.re_kv.search(line)	×
54	if m:	×
55	(key, value) = m.groups()	×
56	self.kvl.append((key, value, time()))	×
57	line = line.strip()	×
58	match = m.group(0)	×
59	pos = line.find(match)	×
60	before = line[:pos]	×
61	after = line[pos + len(match):]	×
62	if len(before) > 0:	×
63	discarded.append(before)	×
64	if len(after) > 0:	×
65	# not a K,V pair part
66	discarded.append(after)	×
67	else:
68	# not a K,V pair
69	discarded.append(line)	×
70	return discarded	×
71
72	def search(self):	×
73	"""! Check if there is a KV value in buffer """
74	return len(self.kvl) > 0	×
75
76	def pop_kv(self):	×
77	if len(self.kvl):	×
78	return self.kvl.pop(0)	×
79	return None, None, time()	×
80
81
82	def conn_primitive_factory(conn_resource, config, event_queue, logger):	×
83	"""! Factory producing connectors based on type and config
84	@param conn_resource Name of connection primitive (e.g. 'serial' for
85	local serial port connection or 'grm' for global resource manager)
86	@param event_queue Even queue of Key-Value protocol
87	@param config Global configuration for connection process
88	@param logger Host Test logger instance
89	@return Object of type <ConnectorPrimitive> or None if type of connection primitive unknown (conn_resource)
90	"""
91	polling_timeout = int(config.get('polling_timeout', 60))	×
92	logger.prn_inf("notify event queue about extra %d sec timeout for serial port pooling"%polling_timeout)	×
93	event_queue.put(('__timeout', polling_timeout, time()))	×
94
95	if conn_resource == 'serial':	×
96	# Standard serial port connection
97	# Notify event queue we will wait additional time for serial port to be ready
98
99	# Get extra configuration related to serial port
100	port = config.get('port')	×
101	baudrate = config.get('baudrate')	×
102
103	logger.prn_inf("initializing serial port listener... ")	×
104	connector = SerialConnectorPrimitive(	×
105	'SERI',
106	port,
107	baudrate,
108	config=config)
109	elif conn_resource == 'grm':	×
110	# Start GRM (Gloabal Resource Mgr) collection
111	logger.prn_inf("initializing global resource mgr listener... ")	×
112	connector = RemoteConnectorPrimitive('GLRM', config=config)	×
113	elif conn_resource == 'fmc':	×
114	# Start Fast Model Connection collection
115	logger.prn_inf("initializing fast model connection")	×
116	connector = FastmodelConnectorPrimitive('FSMD', config=config)	×
117	else:
118	logger.pn_err("unknown connection resource!")	×
119	raise NotImplementedError("ConnectorPrimitive factory: unknown connection resource '%s'!"% conn_resource)	×
120
121	return connector	×
122
123
124	def conn_process(event_queue, dut_event_queue, config):	×
125
126	def __notify_conn_lost():	×
127	error_msg = connector.error()	×
128	connector.finish()	×
129	event_queue.put(('__notify_conn_lost', error_msg, time()))	×
130
131	def __notify_sync_failed():	×
132	error_msg = connector.error()	×
133	connector.finish()	×
134	event_queue.put(('__notify_sync_failed', error_msg, time()))	×
135
136	logger = HtrunLogger('CONN')	×
137	logger.prn_inf("starting connection process...")	×
138
139	# Send connection process start event to host process
140	# NOTE: Do not send any other Key-Value pairs before this!
141	event_queue.put(('__conn_process_start', 1, time()))	×
142
143	# Configuration of conn_opriocess behaviour
144	sync_behavior = int(config.get('sync_behavior', 1))	×
145	sync_timeout = config.get('sync_timeout', 1.0)	×
146	conn_resource = config.get('conn_resource', 'serial')	×
147	last_sync = False	×
148
149	# Create connector instance with proper configuration
150	connector = conn_primitive_factory(conn_resource, config, event_queue, logger)	×
151
152	# If the connector failed, stop the process now
153	if not connector.connected():	×
154	logger.prn_err("Failed to connect to resource")	×
155	__notify_conn_lost()	×
156	return 0	×
157
158	# Create simple buffer we will use for Key-Value protocol data
159	kv_buffer = KiViBufferWalker()	×
160
161	# List of all sent to target UUIDs (if multiple found)
162	sync_uuid_list = []	×
163
164	# We will ignore all kv pairs before we get sync back
165	sync_uuid_discovered = False	×
166
167	def __send_sync(timeout=None):	×
168	sync_uuid = str(uuid.uuid4())	×
169	# Handshake, we will send {{sync;UUID}} preamble and wait for mirrored reply
170	if timeout:	×
171	logger.prn_inf("Reset the part and send in new preamble...")	×
172	connector.reset()	×
173	logger.prn_inf("resending new preamble '%s' after %0.2f sec"% (sync_uuid, timeout))	×
174	else:
175	logger.prn_inf("sending preamble '%s'"% sync_uuid)	×
176
177	if connector.write_kv('__sync', sync_uuid):	×
178	return sync_uuid	×
179	else:
180	return None	×
181
182	# Send simple string to device to 'wake up' greentea-client k-v parser
183	if not connector.write("mbed" * 10, log=True):	×
184	# Failed to write 'wake up' string, exit conn_process
185	__notify_conn_lost()	×
186	return 0	×
187
188
189	# Sync packet management allows us to manipulate the way htrun sends __sync packet(s)
190	# With current settings we can force on htrun to send __sync packets in this manner:
191	#
192	# * --sync=0 - No sync packets will be sent to target platform
193	# * --sync=-10 - __sync packets will be sent unless we will reach
194	# timeout or proper response is sent from target platform
195	# * --sync=N - Send up to N __sync packets to target platform. Response
196	# is sent unless we get response from target platform or
197	# timeout occur
198
199	if sync_behavior > 0:	×
200	# Sending up to 'n' __sync packets
201	logger.prn_inf("sending up to %s __sync packets (specified with --sync=%s)"% (sync_behavior, sync_behavior))	×
202	sync_uuid = __send_sync()	×
203
204	if sync_uuid:	×
205	sync_uuid_list.append(sync_uuid)	×
206	sync_behavior -= 1	×
207	else:
208	__notify_conn_lost()	×
209	return 0	×
210	elif sync_behavior == 0:	×
211	# No __sync packets
212	logger.prn_wrn("skipping __sync packet (specified with --sync=%s)"% sync_behavior)	×
213	else:
214	# Send __sync until we go reply
215	logger.prn_inf("sending multiple __sync packets (specified with --sync=%s)"% sync_behavior)	×
216
217	sync_uuid = __send_sync()	×
218	if sync_uuid:	×
219	sync_uuid_list.append(sync_uuid)	×
220	sync_behavior -= 1	×
221	else:
222	__notify_conn_lost()	×
223	return 0	×
224
225	loop_timer = time()	×
226	while True:
227
228	# Check if connection is lost to serial
229	if not connector.connected():	×
230	__notify_conn_lost()	×
231	break	×
232
233	# Send data to DUT
234	try:	×
235	(key, value, _) = dut_event_queue.get(block=False)	×
236	except QueueEmpty:	×
237	pass # Check if target sent something	×
238	else:
239	# Return if state machine in host_test_default has finished to end process
240	if key == '__host_test_finished' and value == True:	×
241	logger.prn_inf("received special event '%s' value='%s', finishing"% (key, value))	×
242	connector.finish()	×
243	return 0	×
244	elif key == '__reset':	×
245	logger.prn_inf("received special event '%s', resetting dut" % (key))	×
246	connector.reset()	×
247	event_queue.put(("reset_complete", 0, time()))	×
248	elif not connector.write_kv(key, value):	×
249	connector.write_kv(key, value)	×
250	__notify_conn_lost()	×
251	break	×
252
253	# Since read is done every 0.2 sec, with maximum baud rate we can receive 2304 bytes in one read in worst case.
254	data = connector.read(2304)	×
255	if data:	×
256	# Stream data stream KV parsing
257	print_lines = kv_buffer.append(data)	×
258	for line in print_lines:	×
259	logger.prn_rxd(line)	×
260	event_queue.put(('__rxd_line', line, time()))	×
261	while kv_buffer.search():	×
262	key, value, timestamp = kv_buffer.pop_kv()	×
263
264	if sync_uuid_discovered:	×
265	event_queue.put((key, value, timestamp))	×
266	logger.prn_inf("found KV pair in stream: {{%s;%s}}, queued..."% (key, value))	×
267	else:
268	if key == '__sync':	×
269	if value in sync_uuid_list:	×
270	sync_uuid_discovered = True	×
271	event_queue.put((key, value, time()))	×
272	idx = sync_uuid_list.index(value)	×
273	logger.prn_inf("found SYNC in stream: {{%s;%s}} it is #%d sent, queued..."% (key, value, idx))	×
274	else:
275	logger.prn_err("found faulty SYNC in stream: {{%s;%s}}, ignored..."% (key, value))	×
276	logger.prn_inf("Resetting the part and sync timeout to clear out the buffer...")	×
277	connector.reset()	×
278	loop_timer = time()	×
279	else:
280	logger.prn_wrn("found KV pair in stream: {{%s;%s}}, ignoring..."% (key, value))	×
281
282	if not sync_uuid_discovered:	×
283	# Resending __sync after 'sync_timeout' secs (default 1 sec)
284	# to target platform. If 'sync_behavior' counter is != 0 we
285	# will continue to send __sync packets to target platform.
286	# If we specify 'sync_behavior' < 0 we will send 'forever'
287	# (or until we get reply)
288
289	if sync_behavior != 0:	×
290	time_to_sync_again = time() - loop_timer	×
291	if time_to_sync_again > sync_timeout:	×
292	sync_uuid = __send_sync(timeout=time_to_sync_again)	×
293
294	if sync_uuid:	×
295	sync_uuid_list.append(sync_uuid)	×
296	sync_behavior -= 1	×
297	loop_timer = time()	×
298	#Sync behavior will be zero and if last sync fails we should report connection
299	#lost
300	if sync_behavior == 0:	×
301	last_sync = True	×
302	else:
303	__notify_conn_lost()	×
304	break	×
305	elif last_sync == True:	×
306	#SYNC lost connection event : Device not responding, send sync failed
307	__notify_sync_failed()	×
308	break	×
309
310	return 0	×

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