From ab35004c08bc87c702bb92a1e353a2b6aa6e3b25 Mon Sep 17 00:00:00 2001 From: Xu Si Yu Date: Fri, 8 May 2026 15:54:43 +0800 Subject: [PATCH] fix(openthread): fix openthread ci ping case --- examples/openthread/ot_ci_function.py | 2 +- examples/openthread/pytest_otbr.py | 17 ++++++++++++----- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/examples/openthread/ot_ci_function.py b/examples/openthread/ot_ci_function.py index ef5e4ddbf44..ac65e624b8e 100644 --- a/examples/openthread/ot_ci_function.py +++ b/examples/openthread/ot_ci_function.py @@ -187,7 +187,7 @@ def changeDeviceRole(dut: IdfDut, role: str) -> None: def getDataset(dut: IdfDut) -> str: execute_command(dut, 'dataset active -x') - dut_data = dut.expect(r'\n(\w+)\r', timeout=5)[1].decode() + dut_data = dut.expect(r'\n([0-9A-Fa-f]+)\r', timeout=5)[1].decode() return str(dut_data) diff --git a/examples/openthread/pytest_otbr.py b/examples/openthread/pytest_otbr.py index e7d4df5a52c..eef30af8b8f 100644 --- a/examples/openthread/pytest_otbr.py +++ b/examples/openthread/pytest_otbr.py @@ -227,11 +227,6 @@ def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: Tuple[IdfDut cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br) logging.info(f'cli_global_unicast_addr {cli_global_unicast_addr}') interface_name = ocf.get_host_interface_name() - command = 'ping ' + str(cli_global_unicast_addr) + ' -c 10' - out_str = subprocess.getoutput(command) - logging.info(f'ping result:\n{out_str}') - role = re.findall(r' (\d+)%', str(out_str))[0] - assert role != '100' command = 'ifconfig ' + interface_name + ' | grep inet6 | grep global' out_bytes = subprocess.check_output(command, shell=True, timeout=5) out_str = out_bytes.decode('utf-8') @@ -240,6 +235,18 @@ def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: Tuple[IdfDut logging.info(f'host_global_unicast_addr: {host_global_unicast_addr}') if not host_global_unicast_addr: raise Exception(f'onlinkprefix: {onlinkprefix}, host_global_unicast_addr: {host_global_unicast_addr}') + host_ping_loss_rates = [] + for src_addr in host_global_unicast_addr: + command = f'ping -6 -I {src_addr} {cli_global_unicast_addr} -c 10' + result = subprocess.run(command, shell=True, capture_output=True, text=True) + ping_output = (result.stdout or '') + (f'\n{result.stderr}' if result.stderr else '') + logging.info('ping result (source=%s):\n%s', src_addr, ping_output) + loss_match = re.search(r'(\d+)\s*%', ping_output) + loss_rate = int(loss_match.group(1)) if loss_match else 100 + host_ping_loss_rates.append(loss_rate) + logging.info('ping parsed result (source=%s): loss_rate=%d%%', src_addr, loss_rate) + logging.info('host_ping_loss_rates: %s', host_ping_loss_rates) + assert any(loss_rate != 100 for loss_rate in host_ping_loss_rates) rx_nums = 0 for ip_addr in host_global_unicast_addr: txrx_nums = ocf.ot_ping(cli, str(ip_addr), count=10)