Mirrors/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-10-03 03:31:41 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(openthread): optimize the process of openthread pytest cases

Browse Source
This commit is contained in:
Xu Si Yu
2026-06-05 03:50:26 +00:00
committed by Zhang Wen Xu
parent f3dac0b82d
commit ecd72ef115
2 changed files with 63 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/openthread/ot_ci_function.py
+25 -13
View File
@@ -90,7 +90,7 @@ class wifi_parameter:
self.retry_times = retry_times
def joinThreadNetwork(dut: IdfDut, thread: thread_parameter) -> None:
def SetThreadNetworkPara(dut: IdfDut, thread: thread_parameter) -> None:
if thread.dataset:
command = 'dataset set active ' + thread.dataset
execute_command(dut, command)
@@ -128,6 +128,9 @@ def joinThreadNetwork(dut: IdfDut, thread: thread_parameter) -> None:
dut.expect('Done', timeout=5)
execute_command(dut, 'dataset commit active')
dut.expect('Done', timeout=5)
def StartThreadNetwork(dut: IdfDut, thread: thread_parameter) -> None:
if thread.bbr:
execute_command(dut, 'bbr enable')
dut.expect('Done', timeout=5)
@@ -158,7 +161,7 @@ def joinWiFiNetwork(dut: IdfDut, wifi: wifi_parameter) -> Tuple[str, int]:
ip_address = ''
for order in range(1, wifi.retry_times):
command = 'wifi connect -s ' + str(wifi.ssid) + ' -p ' + str(wifi.psk)
tmp = get_ouput_string(dut, command, 10)
tmp = get_output_string(dut, command, 10)
if 'sta ip' in str(tmp):
ip_address = re.findall(r'sta ip: (\w+.\w+.\w+.\w+),', str(tmp))[0]
wait(dut, 2)
@@ -363,11 +366,16 @@ def check_if_host_receive_ra(br: IdfDut) -> bool:
interface_name = get_host_interface_name()
clean_buffer(br)
omrprefix = get_omrprefix(br)
command = 'ip -6 route | grep ' + str(interface_name)
onlinkprefix = get_onlinkprefix(br)
command = f'ip -6 route show dev {interface_name}'
out_str = subprocess.getoutput(command)
print('br omrprefix: ', str(omrprefix))
print('host route table:\n', str(out_str))
return str(omrprefix) in str(out_str)
has_omr_route = str(omrprefix) in str(out_str)
has_onlink_gua = host_global_address_has_onlink_prefix(interface_name, onlinkprefix)
print(
f'RA check: omrprefix={omrprefix} onlinkprefix={onlinkprefix} '
f'has_omr_route={has_omr_route} has_onlink_gua={has_onlink_gua}'
)
return has_omr_route and has_onlink_gua
def host_connect_wifi() -> None:
@@ -392,14 +400,16 @@ def wait_for_host_ra_route(
for attempt in range(1, retries + 1):
if is_joined_wifi_network(br):
log_ipv6_addr_route_by_interface(interface_name, title='RA Ready!')
print(f'Host RA is ready on attempt {attempt}/{retries}.')
return
print(f'Host route not ready yet, retry {attempt}/{retries}...')
log_ipv6_addr_route_by_interface(interface_name, title=f'Wait RA ({attempt}/{retries})')
if attempt < retries:
time.sleep(interval_s)
time.sleep(interval_s)
raise AssertionError('Host did not receive RA / OMR route in time')
log_ipv6_addr_route_by_interface(interface_name, title='RA check failed')
raise AssertionError('Host did not receive valid RA in time (OMR route and onlink GUA both required)')
def host_global_address_has_onlink_prefix(interface_name: str, onlinkprefix: str) -> bool:
@@ -442,13 +452,15 @@ def wait_for_host_onlink_global_address(
for attempt in range(1, retries + 1):
if host_global_address_has_onlink_prefix(interface_name, onlinkprefix):
log_ipv6_addr_route_by_interface(interface_name, title='Onlink GUA ready!')
print(f'Host onlink GUA is ready on attempt {attempt}/{retries}.')
return onlinkprefix
print(f'Host onlink GUA not ready yet, retry {attempt}/{retries}...')
log_ipv6_addr_route_by_interface(interface_name, title=f'Wait onlink GUA ({attempt}/{retries})')
time.sleep(interval_s)
if attempt < retries:
time.sleep(interval_s)
log_ipv6_addr_route_by_interface(interface_name, title='Onlink GUA check failed')
raise AssertionError(f'Host did not get a global IPv6 address in onlink prefix {onlinkprefix!r} in time')
@@ -456,7 +468,7 @@ thread_ipv6_group = 'ff04:0:0:0:0:0:0:125'
def check_ipmaddr(dut: IdfDut) -> bool:
info = get_ouput_string(dut, 'ipmaddr', 2)
info = get_output_string(dut, 'ipmaddr', 2)
if thread_ipv6_group in str(info):
return True
return False
@@ -772,7 +784,7 @@ def execute_command(dut: IdfDut, command: str, prefix: str = 'ot ') -> None:
dut.write(prefix + command)
def get_ouput_string(dut: IdfDut, command: str, wait_time: int) -> str:
def get_output_string(dut: IdfDut, command: str, wait_time: int) -> str:
execute_command(dut, command)
tmp = dut.expect(pexpect.TIMEOUT, timeout=wait_time)
clean_buffer(dut)
examples/openthread/pytest_otbr.py
+38 -29
View File
@@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: 2022-2025 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
# !/usr/bin/env python3
import copy
@@ -157,7 +157,8 @@ def test_thread_connect(dut:Tuple[IdfDut, IdfDut, IdfDut]) -> None:
for cli in cli_list:
ocf.init_thread(cli)
br_ot_para = copy.copy(default_br_ot_para)
ocf.joinThreadNetwork(br, br_ot_para)
ocf.SetThreadNetworkPara(br, br_ot_para)
ocf.StartThreadNetwork(br, br_ot_para)
cli_ot_para = copy.copy(default_cli_ot_para)
cli_ot_para.dataset = ocf.getDataset(br)
try:
@@ -165,7 +166,8 @@ def test_thread_connect(dut:Tuple[IdfDut, IdfDut, IdfDut]) -> None:
for cli in cli_list:
cli_ot_para.exaddr = router_extaddr_list[order]
order = order + 1
ocf.joinThreadNetwork(cli, cli_ot_para)
ocf.SetThreadNetworkPara(cli, cli_ot_para)
ocf.StartThreadNetwork(cli, cli_ot_para)
for cli in cli_list:
cli_mleid_addr = ocf.get_mleid_addr(cli)
br_mleid_addr = ocf.get_mleid_addr(br)
@@ -189,15 +191,20 @@ def test_thread_connect(dut:Tuple[IdfDut, IdfDut, IdfDut]) -> None:
def formBasicWiFiThreadNetwork(br:IdfDut, cli:IdfDut) -> None:
ocf.init_thread(br)
ocf.init_thread(cli)
otbr_wifi_para = copy.copy(default_br_wifi_para)
ocf.joinWiFiNetwork(br, otbr_wifi_para)
otbr_thread_para = copy.copy(default_br_ot_para)
ocf.joinThreadNetwork(br, otbr_thread_para)
otbr_wifi_para = copy.copy(default_br_wifi_para)
otbr_thread_para.extpanid = secrets.token_hex(8)
ocf.SetThreadNetworkPara(br, otbr_thread_para)
ocf.joinWiFiNetwork(br, otbr_wifi_para)
ocf.StartThreadNetwork(br, otbr_thread_para)
otcli_thread_para = copy.copy(default_cli_ot_para)
otcli_thread_para.dataset = ocf.getDataset(br)
otcli_thread_para.exaddr = '7766554433221101'
ocf.joinThreadNetwork(cli, otcli_thread_para)
ocf.wait(cli,10)
ocf.SetThreadNetworkPara(cli, otcli_thread_para)
ocf.StartThreadNetwork(cli, otcli_thread_para)
ocf.wait(cli, 10)
# Polling wait for host RA readiness: both OMR route and onlink GUA must exist.
ocf.wait_for_host_ra_route(br)
# Case 2: Bidirectional IPv6 connectivity
@@ -238,8 +245,7 @@ def test_Bidirectional_IPv6_connectivity(Init_interface:bool, dut: Tuple[IdfDut,
formBasicWiFiThreadNetwork(br, cli)
try:
ocf.wait_for_host_ra_route(br)
onlinkprefix = ocf.wait_for_host_onlink_global_address(br)
onlinkprefix = ocf.get_onlinkprefix(br)
print(f'br onlinkprefix: {onlinkprefix}')
cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br)
print(f'cli_global_unicast_addr {cli_global_unicast_addr}')
@@ -250,7 +256,7 @@ def test_Bidirectional_IPv6_connectivity(Init_interface:bool, dut: Tuple[IdfDut,
pattern = rf'\W+({onlinkprefix}(?:\w+:){{3}}\w+)\W+'
host_global_unicast_addr = re.findall(pattern, out_str)
print(f'host_global_unicast_addr: {host_global_unicast_addr}')
if host_global_unicast_addr is None:
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:
@@ -314,7 +320,6 @@ def test_multicast_forwarding_A(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
formBasicWiFiThreadNetwork(br, cli)
try:
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
assert ocf.thread_is_joined_group(cli)
@@ -378,7 +383,6 @@ def test_multicast_forwarding_B(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
formBasicWiFiThreadNetwork(br, cli)
try:
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
ocf.execute_command(cli, 'udp open')
@@ -444,7 +448,6 @@ def test_service_discovery_of_Thread_device(Init_interface:bool, Init_avahi:bool
formBasicWiFiThreadNetwork(br, cli)
try:
ocf.wait_for_host_ra_route(br)
command = 'avahi-browse -rt _testyyy._udp'
out_str = subprocess.getoutput(command)
print('avahi-browse:\n', str(out_str))
@@ -515,7 +518,6 @@ def test_service_discovery_of_WiFi_device(Init_interface:bool, Init_avahi:bool,
formBasicWiFiThreadNetwork(br, cli)
sp: subprocess.Popen | None = None
try:
ocf.wait_for_host_ra_route(br)
br_global_unicast_addr = ocf.get_global_unicast_addr(br, br)
command = 'dns config ' + br_global_unicast_addr
ocf.execute_command(cli, command)
@@ -526,12 +528,17 @@ def test_service_discovery_of_WiFi_device(Init_interface:bool, Init_avahi:bool,
print('domain name is: ', domain_name)
command = 'dns resolve ' + domain_name + '.default.service.arpa.'
ocf.execute_command(cli, command)
cli.expect('TTL', timeout=10)
cli.expect('Done', timeout=10)
for _ in range(3):
tmp = ocf.get_output_string(cli, command, 5)
if 'TTL' in tmp and 'Done' in str(tmp):
break
time.sleep(1)
else:
print('DNS resolution failed after 3 retries, with no response received.')
assert False
command = 'dns browse _testxxx._udp.default.service.arpa'
tmp = ocf.get_ouput_string(cli, command, 10)
tmp = ocf.get_output_string(cli, command, 10)
assert 'Port:12347' not in str(tmp)
ocf.restart_avahi()
command = 'avahi-publish-service testxxx _testxxx._udp 12347 test=1235 dn="for_ci_br_test"'
@@ -540,12 +547,12 @@ def test_service_discovery_of_WiFi_device(Init_interface:bool, Init_avahi:bool,
ocf.wait(cli, 5)
command = 'dns browse _testxxx._udp.default.service.arpa'
tmp = ocf.get_ouput_string(cli, command, 10)
tmp = ocf.get_output_string(cli, command, 10)
assert 'response for _testxxx' in str(tmp)
assert 'Port:12347' in str(tmp)
command = 'dns service testxxx _testxxx._udp.default.service.arpa.'
tmp = ocf.get_ouput_string(cli, command, 10)
tmp = ocf.get_output_string(cli, command, 10)
assert 'response for testxxx' in str(tmp)
assert 'Port:12347' in str(tmp)
finally:
@@ -585,7 +592,6 @@ def test_ICMP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) ->
formBasicWiFiThreadNetwork(br, cli)
try:
ocf.wait_for_host_ra_route(br)
host_ipv4_address = ocf.get_host_ipv4_address()
print('host_ipv4_address: ', host_ipv4_address)
rx_nums = ocf.ot_ping(cli, str(host_ipv4_address), count=5)[1]
@@ -624,7 +630,6 @@ def test_UDP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N
formBasicWiFiThreadNetwork(br, cli)
try:
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
ocf.execute_command(cli, 'udp open')
@@ -681,7 +686,6 @@ def test_TCP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N
formBasicWiFiThreadNetwork(br, cli)
try:
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
ocf.execute_command(cli, 'tcpsockclient open')
@@ -754,7 +758,8 @@ def test_ot_sleepy_device(dut: Tuple[IdfDut, IdfDut]) -> None:
ocf.init_thread(leader)
time.sleep(3)
leader_para = ocf.thread_parameter('leader', '', '12', '7766554433221100', False)
ocf.joinThreadNetwork(leader, leader_para)
ocf.SetThreadNetworkPara(leader, leader_para)
ocf.StartThreadNetwork(leader, leader_para)
ocf.wait(leader, 5)
dataset = ocf.getDataset(leader)
ocf.execute_command(sleepy_device, 'mode -')
@@ -858,7 +863,8 @@ def test_br_meshcop(Init_interface:bool, Init_avahi:bool, dut: Tuple[IdfDut, Idf
br_thread_para = copy.copy(default_br_ot_para)
networkname = 'OTCI-' + str(secrets.token_hex(1))
br_thread_para.setnetworkname(networkname)
ocf.joinThreadNetwork(br, br_thread_para)
ocf.SetThreadNetworkPara(br, br_thread_para)
ocf.StartThreadNetwork(br, br_thread_para)
ocf.wait(br, 10)
ocf.wait_for_host_ra_route(br)
command = 'timeout 3 avahi-browse -r _meshcop._udp'
@@ -921,7 +927,8 @@ def test_trel_connect(dut: Tuple[IdfDut, IdfDut]) -> None:
ocf.init_thread(trel)
trel_leader_para = copy.copy(default_br_ot_para)
trel_leader_para.bbr = False
ocf.joinThreadNetwork(trel_s3, trel_leader_para)
ocf.SetThreadNetworkPara(trel_s3, trel_leader_para)
ocf.StartThreadNetwork(trel_s3, trel_leader_para)
trel_para = copy.copy(default_cli_ot_para)
trel_para.dataset = ocf.getDataset(trel_s3)
try:
@@ -929,7 +936,8 @@ def test_trel_connect(dut: Tuple[IdfDut, IdfDut]) -> None:
for trel in trel_list:
trel_para.exaddr = router_extaddr_list[order]
order = order + 1
ocf.joinThreadNetwork(trel, trel_para)
ocf.SetThreadNetworkPara(trel, trel_para)
ocf.StartThreadNetwork(trel, trel_para)
for trel in trel_list:
trel_mleid_addr = ocf.get_mleid_addr(trel)
trel_s3_mleid_addr = ocf.get_mleid_addr(trel_s3)
@@ -1012,7 +1020,8 @@ def test_ot_ssed_device(dut: Tuple[IdfDut, IdfDut]) -> None:
ocf.init_thread(leader)
time.sleep(3)
leader_para = ocf.thread_parameter('leader', '', '12', '7766554433221100', False)
ocf.joinThreadNetwork(leader, leader_para)
ocf.SetThreadNetworkPara(leader, leader_para)
ocf.StartThreadNetwork(leader, leader_para)
ocf.wait(leader, 5)
ocf.execute_command(leader, 'networkkey')
dataset = ocf.getDataset(leader)