tools/cp-caps/device_manager.py - third_party/openthread - Git at Google

 #!/usr/bin/env python3
 #
 #  Copyright (c) 2025, The OpenThread Authors.
 #  All rights reserved.
 #
 #  Redistribution and use in source and binary forms, with or without
 #  modification, are permitted provided that the following conditions are met:
 #  1. Redistributions of source code must retain the above copyright
 #     notice, this list of conditions and the following disclaimer.
 #  2. Redistributions in binary form must reproduce the above copyright
 #     notice, this list of conditions and the following disclaimer in the
 #     documentation and/or other materials provided with the distribution.
 #  3. Neither the name of the copyright holder nor the
 #     names of its contributors may be used to endorse or promote products
 #     derived from this software without specific prior written permission.
 #
 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 #  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 #  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 #  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 #  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 #  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 #  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 #  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 #  POSSIBILITY OF SUCH DAMAGE.
 #

 import logging
 import os
 import sys
 import threading
 import queue

 from dataclasses import dataclass
 from typing import Optional

 import otci
 from otci import OTCI
 from otci.errors import ExpectLineTimeoutError, InvalidArgumentsError

 CP_CAPABILITY_VERSION = "0.1.1-dev"

 logging.basicConfig(level=logging.WARNING)


 @dataclass
 class Frame:
     """Represents a Thread radio frame.

     Attributes:
       name: The description of the frame.
       tx_frame: The psdu of the frame that is to be sent.
       dst_address: The destination MAC address of the `tx_frame`. It is used by the receiver to filter
           out the `tx_frame`.
       is_security_processed: The value of the `otRadioFrame.mInfo.mTxInfo.mIsSecurityProcessed` field.
           If it is set to False, the `active_dataset` and `src_ext_address` should also be set for the
           radio driver to encrypt the `tx_frame`.
       expect_rx_frame: The frame expected to be received. The expected frame should be the same as the
           `tx_frame` if the `expect_rx_frame` is set to None.
       active_dataset: The active dataset.
       src_ext_address: The source extended MAC address of the transmitter.
     """
     name: str
     tx_frame: str
     dst_address: str
     is_security_processed: Optional[bool] = True
     expect_rx_frame: Optional[str] = None
     active_dataset: Optional[str] = None
     src_ext_address: Optional[str] = None


 class DeviceManager(object):
     """Provides the DUT, reference device and common methods for test cases."""

     DEFAULT_FORMAT_ALIGN_LENGTH = 58  # The default formatted string alignment length

     def __init__(self):
         self.__dut = self.__connect_dut()
         self.__ref = self.__connect_reference_device()

         if self.__get_debug_enabled():
             logger = logging.getLogger()
             logger.setLevel(logging.DEBUG)

     @property
     def dut(self) -> OTCI:
         return self.__dut

     @property
     def ref(self) -> OTCI:
         return self.__ref

     def output_test_result_string(self, name: str, value: str, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH):
         prefix = (name + ' ').ljust(align_length, '-')
         print(f'{prefix} {value}')

     def output_test_result_bool(self, name: str, value: bool, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH):
         self.output_test_result_string(name, 'OK' if value else 'NotSupported', align_length)

     def get_default_dataset(self):
         return self.__dut.create_dataset(channel=20, network_key='00112233445566778899aabbccddcafe')

     def send_formatted_frames_retries(self, sender: OTCI, receiver: OTCI, frame: Frame, max_send_retries: int = 5):
         for i in range(0, max_send_retries):
             if self.__send_formatted_frame(sender, receiver, frame):
                 return True

         return False

     #
     # Private methods
     #
     def __get_debug_enabled(self) -> bool:
         enabled = os.getenv('DEBUG', None)
         return enabled == 'on'

     def __get_adb_key(self) -> Optional[str]:
         return os.getenv('ADB_KEY', None)

     def __connect_dut(self) -> OTCI:
         if os.getenv('DUT_ADB_TCP'):
             node = otci.connect_otbr_adb_tcp(os.getenv('DUT_ADB_TCP'), adb_key=self.__get_adb_key())
         elif os.getenv('DUT_ADB_USB'):
             node = otci.connect_otbr_adb_usb(os.getenv('DUT_ADB_USB'), adb_key=self.__get_adb_key())
         elif os.getenv('DUT_CLI_SERIAL'):
             node = otci.connect_cli_serial(os.getenv('DUT_CLI_SERIAL'))
         elif os.getenv('DUT_SSH'):
             node = otci.connect_otbr_ssh(os.getenv('DUT_SSH'))
         else:
             raise InvalidArgumentsError(
                 "Please set DUT_ADB_TCP, DUT_ADB_USB, DUT_CLI_SERIAL or DUT_SSH to connect to the DUT device.")

         return node

     def __connect_reference_device(self) -> OTCI:
         if os.getenv('REF_CLI_SERIAL'):
             node = otci.connect_cli_serial(os.getenv('REF_CLI_SERIAL'))
         elif os.getenv('REF_SSH'):
             node = otci.connect_otbr_ssh(os.getenv('REF_SSH'))
         elif os.getenv('REF_ADB_USB'):
             node = otci.connect_otbr_adb_usb(os.getenv('REF_ADB_USB'), adb_key=self.__get_adb_key())
         else:
             raise InvalidArgumentsError(
                 "Please set REF_CLI_SERIAL, REF_SSH or REF_ADB_USB to connect to the reference device.")

         return node

     def __send_formatted_frame(self, sender: OTCI, receiver: OTCI, frame: Frame):
         # When `is_security_processed` is False, the frame may need the radio driver
         # to encrypt the frame. Here sets the active dataset and the MAC source address for the
         # radio driver to encrypt the frame.
         if frame.is_security_processed is False:
             if frame.active_dataset is None or frame.src_ext_address is None:
                 raise InvalidArgumentsError(
                     "When the 'is_security_processed' is 'False', the 'active_dataset' and 'src_ext_address' must be set"
                 )
             # Do the factory reset to set the frame counter to 0.
             sender.factory_reset()
             receiver.factory_reset()
             sender.set_dataset_bytes('active', frame.active_dataset)
             sender.set_extaddr(frame.src_ext_address)

         sender.diag_start()
         receiver.diag_start()

         channel = 11
         num_sent_frames = 1

         sender.diag_set_channel(channel)
         receiver.diag_set_channel(channel)
         receiver.diag_radio_receive()
         receiver.diag_set_radio_receive_filter_dest_mac_address(frame.dst_address)
         receiver.diag_enable_radio_receive_filter()

         result_queue = queue.Queue()
         receive_task = threading.Thread(target=self.__radio_receive_task,
                                         args=(receiver, num_sent_frames, result_queue),
                                         daemon=True)
         receive_task.start()

         sender.wait(0.1)

         sender.diag_frame(frame.tx_frame,
                           is_security_processed=frame.is_security_processed,
                           max_csma_backoffs=4,
                           max_frame_retries=4,
                           csma_ca_enabled=True)
         sender.diag_send(num_sent_frames, is_async=False)

         receive_task.join()

         if result_queue.empty():
             ret = False  # No frame is received.
         else:
             received_frames = result_queue.get()
             if len(received_frames) != num_sent_frames:
                 ret = False
             else:
                 # The radio driver may not append the FCF field to the received frame. Do not check the FCF field here.
                 FCF_LENGTH = 4
                 expect_frame = frame.expect_rx_frame or frame.tx_frame
                 ret = expect_frame[:-FCF_LENGTH] == received_frames[0]['psdu'][:-FCF_LENGTH]

         if ret:
             sender.diag_stop()
             receiver.diag_stop()
         else:
             # The command `diag radio receive <number>` may fail to receive specified number of frames in default
             # timeout time. In this case, the diag module still in `sync` mode, and it only allows users to run the
             # command `factoryreset` to terminate the test.
             sender.factory_reset()
             receiver.factory_reset()

         return ret

     def __radio_receive_task(self, receiver: OTCI, number: int, result_queue: queue):
         try:
             receiver.set_execute_command_retry(0)
             result = receiver.diag_radio_receive_number(number)
         except ExpectLineTimeoutError:
             pass
         else:
             result_queue.put(result)
         finally:
             receiver.set_execute_command_retry(OTCI.DEFAULT_EXEC_COMMAND_RETRY)
	#!/usr/bin/env python3
	#
	# Copyright (c) 2025, The OpenThread Authors.
	# All rights reserved.
	#
	# Redistribution and use in source and binary forms, with or without
	# modification, are permitted provided that the following conditions are met:
	# 1. Redistributions of source code must retain the above copyright
	# notice, this list of conditions and the following disclaimer.
	# 2. Redistributions in binary form must reproduce the above copyright
	# notice, this list of conditions and the following disclaimer in the
	# documentation and/or other materials provided with the distribution.
	# 3. Neither the name of the copyright holder nor the
	# names of its contributors may be used to endorse or promote products
	# derived from this software without specific prior written permission.
	#
	# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
	# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
	# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
	# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
	# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
	# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
	# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
	# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
	# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
	# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
	# POSSIBILITY OF SUCH DAMAGE.
	#

	import logging
	import os
	import sys
	import threading
	import queue

	from dataclasses import dataclass
	from typing import Optional

	import otci
	from otci import OTCI
	from otci.errors import ExpectLineTimeoutError, InvalidArgumentsError

	CP_CAPABILITY_VERSION = "0.1.1-dev"

	logging.basicConfig(level=logging.WARNING)


	@dataclass
	class Frame:
	"""Represents a Thread radio frame.

	Attributes:
	name: The description of the frame.
	tx_frame: The psdu of the frame that is to be sent.
	dst_address: The destination MAC address of the `tx_frame`. It is used by the receiver to filter
	out the `tx_frame`.
	is_security_processed: The value of the `otRadioFrame.mInfo.mTxInfo.mIsSecurityProcessed` field.
	If it is set to False, the `active_dataset` and `src_ext_address` should also be set for the
	radio driver to encrypt the `tx_frame`.
	expect_rx_frame: The frame expected to be received. The expected frame should be the same as the
	`tx_frame` if the `expect_rx_frame` is set to None.
	active_dataset: The active dataset.
	src_ext_address: The source extended MAC address of the transmitter.
	"""
	name: str
	tx_frame: str
	dst_address: str
	is_security_processed: Optional[bool] = True
	expect_rx_frame: Optional[str] = None
	active_dataset: Optional[str] = None
	src_ext_address: Optional[str] = None


	class DeviceManager(object):
	"""Provides the DUT, reference device and common methods for test cases."""

	DEFAULT_FORMAT_ALIGN_LENGTH = 58 # The default formatted string alignment length

	def __init__(self):
	self.__dut = self.__connect_dut()
	self.__ref = self.__connect_reference_device()

	if self.__get_debug_enabled():
	logger = logging.getLogger()
	logger.setLevel(logging.DEBUG)

	@property
	def dut(self) -> OTCI:
	return self.__dut

	@property
	def ref(self) -> OTCI:
	return self.__ref

	def output_test_result_string(self, name: str, value: str, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH):
	prefix = (name + ' ').ljust(align_length, '-')
	print(f'{prefix} {value}')

	def output_test_result_bool(self, name: str, value: bool, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH):
	self.output_test_result_string(name, 'OK' if value else 'NotSupported', align_length)

	def get_default_dataset(self):
	return self.__dut.create_dataset(channel=20, network_key='00112233445566778899aabbccddcafe')

	def send_formatted_frames_retries(self, sender: OTCI, receiver: OTCI, frame: Frame, max_send_retries: int = 5):
	for i in range(0, max_send_retries):
	if self.__send_formatted_frame(sender, receiver, frame):
	return True

	return False

	#
	# Private methods
	#
	def __get_debug_enabled(self) -> bool:
	enabled = os.getenv('DEBUG', None)
	return enabled == 'on'

	def __get_adb_key(self) -> Optional[str]:
	return os.getenv('ADB_KEY', None)

	def __connect_dut(self) -> OTCI:
	if os.getenv('DUT_ADB_TCP'):
	node = otci.connect_otbr_adb_tcp(os.getenv('DUT_ADB_TCP'), adb_key=self.__get_adb_key())
	elif os.getenv('DUT_ADB_USB'):
	node = otci.connect_otbr_adb_usb(os.getenv('DUT_ADB_USB'), adb_key=self.__get_adb_key())
	elif os.getenv('DUT_CLI_SERIAL'):
	node = otci.connect_cli_serial(os.getenv('DUT_CLI_SERIAL'))
	elif os.getenv('DUT_SSH'):
	node = otci.connect_otbr_ssh(os.getenv('DUT_SSH'))
	else:
	raise InvalidArgumentsError(
	"Please set DUT_ADB_TCP, DUT_ADB_USB, DUT_CLI_SERIAL or DUT_SSH to connect to the DUT device.")

	return node

	def __connect_reference_device(self) -> OTCI:
	if os.getenv('REF_CLI_SERIAL'):
	node = otci.connect_cli_serial(os.getenv('REF_CLI_SERIAL'))
	elif os.getenv('REF_SSH'):
	node = otci.connect_otbr_ssh(os.getenv('REF_SSH'))
	elif os.getenv('REF_ADB_USB'):
	node = otci.connect_otbr_adb_usb(os.getenv('REF_ADB_USB'), adb_key=self.__get_adb_key())
	else:
	raise InvalidArgumentsError(
	"Please set REF_CLI_SERIAL, REF_SSH or REF_ADB_USB to connect to the reference device.")

	return node

	def __send_formatted_frame(self, sender: OTCI, receiver: OTCI, frame: Frame):
	# When `is_security_processed` is False, the frame may need the radio driver
	# to encrypt the frame. Here sets the active dataset and the MAC source address for the
	# radio driver to encrypt the frame.
	if frame.is_security_processed is False:
	if frame.active_dataset is None or frame.src_ext_address is None:
	raise InvalidArgumentsError(
	"When the 'is_security_processed' is 'False', the 'active_dataset' and 'src_ext_address' must be set"
	)
	# Do the factory reset to set the frame counter to 0.
	sender.factory_reset()
	receiver.factory_reset()
	sender.set_dataset_bytes('active', frame.active_dataset)
	sender.set_extaddr(frame.src_ext_address)

	sender.diag_start()
	receiver.diag_start()

	channel = 11
	num_sent_frames = 1

	sender.diag_set_channel(channel)
	receiver.diag_set_channel(channel)
	receiver.diag_radio_receive()
	receiver.diag_set_radio_receive_filter_dest_mac_address(frame.dst_address)
	receiver.diag_enable_radio_receive_filter()

	result_queue = queue.Queue()
	receive_task = threading.Thread(target=self.__radio_receive_task,
	args=(receiver, num_sent_frames, result_queue),
	daemon=True)
	receive_task.start()

	sender.wait(0.1)

	sender.diag_frame(frame.tx_frame,
	is_security_processed=frame.is_security_processed,
	max_csma_backoffs=4,
	max_frame_retries=4,
	csma_ca_enabled=True)
	sender.diag_send(num_sent_frames, is_async=False)

	receive_task.join()

	if result_queue.empty():
	ret = False # No frame is received.
	else:
	received_frames = result_queue.get()
	if len(received_frames) != num_sent_frames:
	ret = False
	else:
	# The radio driver may not append the FCF field to the received frame. Do not check the FCF field here.
	FCF_LENGTH = 4
	expect_frame = frame.expect_rx_frame or frame.tx_frame
	ret = expect_frame[:-FCF_LENGTH] == received_frames[0]['psdu'][:-FCF_LENGTH]

	if ret:
	sender.diag_stop()
	receiver.diag_stop()
	else:
	# The command `diag radio receive <number>` may fail to receive specified number of frames in default
	# timeout time. In this case, the diag module still in `sync` mode, and it only allows users to run the
	# command `factoryreset` to terminate the test.
	sender.factory_reset()
	receiver.factory_reset()

	return ret

	def __radio_receive_task(self, receiver: OTCI, number: int, result_queue: queue):
	try:
	receiver.set_execute_command_retry(0)
	result = receiver.diag_radio_receive_number(number)
	except ExpectLineTimeoutError:
	pass
	else:
	result_queue.put(result)
	finally:
	receiver.set_execute_command_retry(OTCI.DEFAULT_EXEC_COMMAND_RETRY)