diff --git a/examples/openthread/ot_ci_function.py b/examples/openthread/ot_ci_function.py index 2b267e62ec..9c594894c3 100644 --- a/examples/openthread/ot_ci_function.py +++ b/examples/openthread/ot_ci_function.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Unlicense OR CC0-1.0 # !/usr/bin/env python3 # this file defines some functions for testing cli and br under pytest framework +import logging import os import re import socket @@ -28,7 +29,7 @@ def extract_address( try: result = dut.expect(pattern, timeout=5)[1].decode() except Exception as e: - print(f'Error: {e}') + logging.error(f'Error: {e}') return default_return return func(result) @@ -151,7 +152,7 @@ def getDeviceRole(dut: IdfDut) -> str: wait(dut, 1) execute_command(dut, 'state') role = dut.expect(r'\W+(\w+)\W+Done', timeout=5)[1].decode() - print(role) + logging.info(role) return str(role) @@ -310,7 +311,7 @@ def get_host_interface_name() -> str: interface_name = config.get('interface_name') if interface_name: if interface_name == 'eth0': - print( + logging.warning( f"Warning: 'eth0' is not recommended as a valid network interface. " f"Please check and update the 'interface_name' in the configuration file: " f'{config_path}' @@ -318,9 +319,9 @@ def get_host_interface_name() -> str: else: return str(interface_name) else: - print("Warning: Configuration file found but 'interface_name' is not defined.") + logging.warning("Warning: Configuration file found but 'interface_name' is not defined.") except Exception as e: - print(f'Error: Failed to read or parse {config_path}. Details: {e}') + logging.error(f'Error: Failed to read or parse {config_path}. Details: {e}') if 'eth1' in netifaces.interfaces(): return 'eth1' @@ -338,8 +339,8 @@ def check_if_host_receive_ra(br: IdfDut) -> bool: omrprefix = get_omrprefix(br) command = 'ip -6 route | grep ' + str(interface_name) out_str = subprocess.getoutput(command) - print('br omrprefix: ', str(omrprefix)) - print('host route table:\n', str(out_str)) + logging.info(f'br omrprefix: {omrprefix}') + logging.info(f'host route table:\n {out_str}') return str(omrprefix) in str(out_str) @@ -404,7 +405,7 @@ def create_host_udp_server(myudp: udp_parameter) -> None: AF_INET = socket.AF_INET6 else: AF_INET = socket.AF_INET - print('The host start to create udp server!') + logging.info('The host start to create udp server!') if_index = socket.if_nametoindex(interface_name) sock = socket.socket(AF_INET, socket.SOCK_DGRAM) sock.bind((myudp.addr, myudp.port)) @@ -417,13 +418,14 @@ def create_host_udp_server(myudp: udp_parameter) -> None: ) sock.settimeout(myudp.timeout) myudp.init_flag = True - print('The host start to receive message!') + logging.info('The host start to receive message!') myudp.udp_bytes = (sock.recvfrom(1024))[0] - print('The host has received message: ', myudp.udp_bytes) + udp_str = str(myudp.udp_bytes) + logging.info(f'The host has received message: {udp_str}') except OSError: - print('The host did not receive message!') + logging.error('The host did not receive message!') finally: - print('Close the socket.') + logging.info('Close the socket.') sock.close() @@ -438,10 +440,10 @@ def host_udp_send_message(udp_target: udp_parameter) -> None: sock.bind(('::', 12350)) sock.setsockopt(socket.SOL_SOCKET, socket.SO_BINDTODEVICE, interface_name.encode()) sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 32) - print('Host is sending message') + logging.info('Host is sending message') sock.sendto(udp_target.udp_bytes, (udp_target.addr, udp_target.port)) except OSError: - print('Host cannot send message') + logging.error('Host cannot send message') finally: sock.close() @@ -481,13 +483,13 @@ def host_close_service() -> None: command = 'ps auxww | grep avahi-publish-s' out_bytes = subprocess.check_output(command, shell=True, timeout=5) out_str = out_bytes.decode('utf-8') - print('host close service avahi status:\n', out_str) + logging.info(f'host close service avahi status:\n {out_str}') service_info = [line for line in out_str.splitlines() if 'testxxx _testxxx._udp' in line] for line in service_info: - print('Process:', line) + logging.info(f'Process:{line}') pid = line.split()[1] command = 'kill -9 ' + pid - print('kill ', pid) + logging.info(f'kill {pid}') subprocess.call(command, shell=True, timeout=5) time.sleep(1) @@ -520,24 +522,24 @@ def open_host_interface() -> None: def get_domain() -> str: hostname = socket.gethostname() - print('hostname is: ', hostname) + logging.info(f'hostname is: {hostname}') command = 'ps -auxww | grep avahi-daemon | grep running' out_str = subprocess.getoutput(command) - print('avahi status:\n', out_str) + logging.info(f'avahi status:\n {out_str}') role = re.findall(r'\[([\w\W]+)\.local\]', str(out_str))[0] - print('active host is: ', role) + logging.info(f'active host is: {role}') return str(role) def flush_ipv6_addr_by_interface() -> None: interface_name = get_host_interface_name() - print(f'flush ipv6 addr : {interface_name}') + logging.info(f'flush ipv6 addr : {interface_name}') command_show_addr = f'ip -6 addr show dev {interface_name}' command_show_route = f'ip -6 route show dev {interface_name}' addr_before = subprocess.getoutput(command_show_addr) route_before = subprocess.getoutput(command_show_route) - print(f'Before flush, IPv6 addresses: \n{addr_before}') - print(f'Before flush, IPv6 routes: \n{route_before}') + logging.info(f'Before flush, IPv6 addresses: \n{addr_before}') + logging.info(f'Before flush, IPv6 routes: \n{route_before}') subprocess.run(['ip', 'link', 'set', interface_name, 'down']) subprocess.run(['ip', '-6', 'addr', 'flush', 'dev', interface_name]) subprocess.run(['ip', '-6', 'route', 'flush', 'dev', interface_name]) @@ -545,8 +547,8 @@ def flush_ipv6_addr_by_interface() -> None: time.sleep(5) addr_after = subprocess.getoutput(command_show_addr) route_after = subprocess.getoutput(command_show_route) - print(f'After flush, IPv6 addresses: \n{addr_after}') - print(f'After flush, IPv6 routes: \n{route_after}') + logging.info(f'After flush, IPv6 addresses: \n{addr_after}') + logging.info(f'After flush, IPv6 routes: \n{route_after}') class tcp_parameter: @@ -575,28 +577,29 @@ def create_host_tcp_server(mytcp: tcp_parameter) -> None: AF_INET = socket.AF_INET6 else: AF_INET = socket.AF_INET - print('The host start to create a tcp server!') + logging.info('The host start to create a tcp server!') sock = socket.socket(AF_INET, socket.SOCK_STREAM) sock.bind((mytcp.addr, mytcp.port)) sock.listen(5) mytcp.listen_flag = True - print('The tcp server is waiting for connection!') + logging.info('The tcp server is waiting for connection!') sock.settimeout(mytcp.timeout) connfd, addr = sock.accept() - print('The tcp server connected with ', addr) + logging.info(f'The tcp server connected with {addr}') mytcp.recv_flag = True mytcp.tcp_bytes = connfd.recv(1024) - print('The tcp server has received message: ', mytcp.tcp_bytes) + tcp_str = str(mytcp.tcp_bytes) + logging.info(f'The tcp server has received message: {tcp_str}') except OSError: if mytcp.recv_flag: - print('The tcp server did not receive message!') + logging.error('The tcp server did not receive message!') else: - print('The tcp server fail to connect!') + logging.error('The tcp server fail to connect!') finally: - print('Close the socket.') + logging.info('Close the socket.') sock.close() diff --git a/examples/openthread/pytest_otbr.py b/examples/openthread/pytest_otbr.py index e3a968eec7..46d081e9f3 100644 --- a/examples/openthread/pytest_otbr.py +++ b/examples/openthread/pytest_otbr.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Unlicense OR CC0-1.0 # !/usr/bin/env python3 import copy +import logging import os.path import random import re @@ -9,7 +10,6 @@ import secrets import subprocess import threading import time -from typing import Tuple import ot_ci_function as ocf import pexpect @@ -76,7 +76,7 @@ from pytest_embedded_idf.dut import IdfDut @pytest.fixture(scope='module', name='Init_avahi') def fixture_Init_avahi() -> bool: - print('Init Avahi') + logging.info('Init Avahi') ocf.start_avahi() time.sleep(10) return True @@ -84,7 +84,7 @@ def fixture_Init_avahi() -> bool: @pytest.fixture(name='Init_interface') def fixture_Init_interface() -> bool: - print('Init interface') + logging.info('Init interface') ocf.flush_ipv6_addr_by_interface() # The sleep time is set based on experience; reducing it might cause the host to be unready. time.sleep(30) @@ -131,7 +131,7 @@ PORT_MAPPING = {'ESPPORT1': 'esp32h2', 'ESPPORT2': 'esp32s3', 'ESPPORT3': 'esp32 ], indirect=True, ) -def test_thread_connect(dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_thread_connect(dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli_h2 = dut[1] dut[0].serial.stop_redirect_thread() @@ -204,7 +204,7 @@ def formBasicWiFiThreadNetwork(br: IdfDut, cli: IdfDut) -> None: ], indirect=True, ) -def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -214,10 +214,10 @@ def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: Tuple[IdfDut try: assert ocf.is_joined_wifi_network(br) cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br) - print('cli_global_unicast_addr', cli_global_unicast_addr) + logging.info(f'cli_global_unicast_addr {cli_global_unicast_addr}') command = 'ping ' + str(cli_global_unicast_addr) + ' -c 10' out_str = subprocess.getoutput(command) - print('ping result:\n', str(out_str)) + logging.info(f'ping result:\n{out_str}') role = re.findall(r' (\d+)%', str(out_str))[0] assert role != '100' interface_name = ocf.get_host_interface_name() @@ -225,7 +225,8 @@ def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: Tuple[IdfDut out_bytes = subprocess.check_output(command, shell=True, timeout=5) out_str = out_bytes.decode('utf-8') onlinkprefix = ocf.get_onlinkprefix(br) - host_global_unicast_addr = re.findall(r'\W+(%s(?:\w+:){3}\w+)\W+' % onlinkprefix, str(out_str)) + pattern = rf'\W+({onlinkprefix}(?:\w+:){{3}}\w+)\W+' + host_global_unicast_addr = re.findall(pattern, out_str) rx_nums = 0 for ip_addr in host_global_unicast_addr: txrx_nums = ocf.ot_ping(cli, str(ip_addr), count=5) @@ -256,7 +257,7 @@ def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: Tuple[IdfDut ], indirect=True, ) -def test_multicast_forwarding_A(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_multicast_forwarding_A(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -271,7 +272,7 @@ def test_multicast_forwarding_A(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, interface_name = ocf.get_host_interface_name() command = 'ping -I ' + str(interface_name) + ' -t 64 ff04::125 -c 10' out_str = subprocess.getoutput(command) - print('ping result:\n', str(out_str)) + logging.info(f'ping result:\n{out_str}') role = re.findall(r' (\d+)%', str(out_str))[0] assert role != '100' ocf.execute_command(cli, 'udp open') @@ -309,7 +310,7 @@ def test_multicast_forwarding_A(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, ], indirect=True, ) -def test_multicast_forwarding_B(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_multicast_forwarding_B(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -364,7 +365,7 @@ def test_multicast_forwarding_B(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, indirect=True, ) def test_service_discovery_of_Thread_device( - Init_interface: bool, Init_avahi: bool, dut: Tuple[IdfDut, IdfDut, IdfDut] + Init_interface: bool, Init_avahi: bool, dut: tuple[IdfDut, IdfDut, IdfDut] ) -> None: br = dut[2] cli = dut[1] @@ -377,14 +378,14 @@ def test_service_discovery_of_Thread_device( assert ocf.is_joined_wifi_network(br) command = 'avahi-browse -rt _testyyy._udp' out_str = subprocess.getoutput(command) - print('avahi-browse:\n', str(out_str)) + logging.info(f'avahi-browse:\n{out_str}') assert 'myTest' not in str(out_str) hostname = 'myTest' command = 'srp client host name ' + hostname ocf.execute_command(cli, command) cli.expect('Done', timeout=5) cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br) - print('cli_global_unicast_addr', cli_global_unicast_addr) + logging.info(f'cli_global_unicast_addr {cli_global_unicast_addr}') command = 'srp client host address ' + str(cli_global_unicast_addr) ocf.execute_command(cli, command) cli.expect('Done', timeout=5) @@ -397,7 +398,7 @@ def test_service_discovery_of_Thread_device( ocf.wait(cli, 3) command = 'avahi-browse -rt _testyyy._udp' out_str = subprocess.getoutput(command) - print('avahi-browse:\n', str(out_str)) + logging.info(f'avahi-browse:\n {out_str}') assert 'myTest' in str(out_str) finally: ocf.execute_command(br, 'factoryreset') @@ -425,7 +426,7 @@ def test_service_discovery_of_Thread_device( indirect=True, ) def test_service_discovery_of_WiFi_device( - Init_interface: bool, Init_avahi: bool, dut: Tuple[IdfDut, IdfDut, IdfDut] + Init_interface: bool, Init_avahi: bool, dut: tuple[IdfDut, IdfDut, IdfDut] ) -> None: br = dut[2] cli = dut[1] @@ -442,7 +443,7 @@ def test_service_discovery_of_WiFi_device( cli.expect('Done', timeout=5) ocf.wait(cli, 1) domain_name = ocf.get_domain() - print('domain name is: ', domain_name) + logging.info(f'domain name is: {domain_name}') command = 'dns resolve ' + domain_name + '.default.service.arpa.' ocf.execute_command(cli, command) @@ -494,7 +495,7 @@ def test_service_discovery_of_WiFi_device( ], indirect=True, ) -def test_ICMP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_ICMP_NAT64(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -504,7 +505,7 @@ def test_ICMP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> try: assert ocf.is_joined_wifi_network(br) host_ipv4_address = ocf.get_host_ipv4_address() - print('host_ipv4_address: ', host_ipv4_address) + logging.info(f'host_ipv4_address: {host_ipv4_address}') rx_nums = ocf.ot_ping(cli, str(host_ipv4_address), count=5)[1] assert rx_nums != 0 finally: @@ -532,7 +533,7 @@ def test_ICMP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> ], indirect=True, ) -def test_UDP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_UDP_NAT64(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -547,7 +548,7 @@ def test_UDP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> cli.expect('Done', timeout=5) ocf.wait(cli, 3) host_ipv4_address = ocf.get_host_ipv4_address() - print('host_ipv4_address: ', host_ipv4_address) + logging.info(f'host_ipv4_address: {host_ipv4_address}') myudp = ocf.udp_parameter('INET4', host_ipv4_address, 5090, '', False, 15.0, b'') udp_mission = threading.Thread(target=ocf.create_host_udp_server, args=(myudp,)) udp_mission.start() @@ -588,7 +589,7 @@ def test_UDP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> ], indirect=True, ) -def test_TCP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_TCP_NAT64(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -604,7 +605,7 @@ def test_TCP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> ocf.wait(cli, 3) host_ipv4_address = ocf.get_host_ipv4_address() connect_address = ocf.get_ipv6_from_ipv4(host_ipv4_address, br) - print('connect_address is: ', connect_address) + logging.info(f'connect_address is: {connect_address}') mytcp = ocf.tcp_parameter('INET4', host_ipv4_address, 12345, False, False, 15.0, b'') tcp_mission = threading.Thread(target=ocf.create_host_tcp_server, args=(mytcp,)) tcp_mission.start() @@ -657,7 +658,7 @@ def test_TCP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> ], indirect=True, ) -def test_ot_sleepy_device(dut: Tuple[IdfDut, IdfDut]) -> None: +def test_ot_sleepy_device(dut: tuple[IdfDut, IdfDut]) -> None: leader = dut[0] sleepy_device = dut[1] fail_info = re.compile(r'Core\W*?\d\W*?register dump') @@ -706,7 +707,7 @@ def test_ot_sleepy_device(dut: Tuple[IdfDut, IdfDut]) -> None: ], indirect=True, ) -def test_basic_startup(dut: Tuple[IdfDut, IdfDut]) -> None: +def test_basic_startup(dut: tuple[IdfDut, IdfDut]) -> None: br = dut[1] dut[0].serial.stop_redirect_thread() try: @@ -748,7 +749,7 @@ def test_basic_startup(dut: Tuple[IdfDut, IdfDut]) -> None: ], indirect=True, ) -def test_NAT64_DNS(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_NAT64_DNS(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -787,7 +788,7 @@ def test_NAT64_DNS(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> ], indirect=True, ) -def test_br_meshcop(Init_interface: bool, Init_avahi: bool, dut: Tuple[IdfDut, IdfDut]) -> None: +def test_br_meshcop(Init_interface: bool, Init_avahi: bool, dut: tuple[IdfDut, IdfDut]) -> None: br = dut[1] assert Init_interface assert Init_avahi @@ -813,9 +814,9 @@ def test_br_meshcop(Init_interface: bool, Init_avahi: bool, dut: Tuple[IdfDut, I except subprocess.CalledProcessError as e: output_bytes = e.stdout finally: - print('out_bytes: ', output_bytes) + logging.info(f'out_bytes: {output_bytes!r}') output_str = str(output_bytes) - print('out_str: ', output_str) + logging.info(f'out_str: {output_str}') assert 'hostname = [esp-ot-br.local]' in str(output_str) assert ('address = [' + ipv4_address + ']') in str(output_str) @@ -849,7 +850,7 @@ def test_br_meshcop(Init_interface: bool, Init_avahi: bool, dut: Tuple[IdfDut, I ], indirect=True, ) -def test_https_NAT64_DNS(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None: +def test_https_NAT64_DNS(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) -> None: br = dut[2] cli = dut[1] assert Init_interface @@ -887,7 +888,7 @@ def test_https_NAT64_DNS(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut ], indirect=True, ) -def test_trel_connect(dut: Tuple[IdfDut, IdfDut]) -> None: +def test_trel_connect(dut: tuple[IdfDut, IdfDut]) -> None: trel_s3 = dut[1] trel_c6 = dut[0] trel_list = [trel_c6] @@ -940,7 +941,7 @@ def test_trel_connect(dut: Tuple[IdfDut, IdfDut]) -> None: ], indirect=True, ) -def test_br_lib_check(dut: Tuple[IdfDut, IdfDut]) -> None: +def test_br_lib_check(dut: tuple[IdfDut, IdfDut]) -> None: br = dut[1] dut[0].serial.stop_redirect_thread() try: @@ -978,7 +979,7 @@ def test_br_lib_check(dut: Tuple[IdfDut, IdfDut]) -> None: ], indirect=True, ) -def test_ot_ssed_device(dut: Tuple[IdfDut, IdfDut]) -> None: +def test_ot_ssed_device(dut: tuple[IdfDut, IdfDut]) -> None: leader = dut[0] ssed_device = dut[1] try: