espressif/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-04-27 19:13:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'bugfix/fix_ot_ci_testcases' into 'master'

Browse Source 
ci: improve BR host related test cases

See merge request espressif/esp-idf!47089
This commit is contained in:
Zhang Wen Xu
2026-04-08 02:35:01 +00:00
parent ae710978ae fd9b2de370
commit 710f133604
2 changed files with 148 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/openthread/ot_ci_function.py
+129 -19
View File
@@ -1,7 +1,8 @@
# 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
# this file defines some functions for testing cli and br under pytest framework
import ipaddress
import logging
import os
import re
@@ -377,6 +378,78 @@ def is_joined_wifi_network(br: IdfDut) -> bool:
return check_if_host_receive_ra(br)
def wait_for_host_ra_route(
br: IdfDut,
*,
retries: int = 12,
interval_s: int = 5,
) -> None:
interface_name = get_host_interface_name()
log_ipv6_addr_route_by_interface(interface_name, title='Wait RA (initial)')
for attempt in range(1, retries + 1):
if is_joined_wifi_network(br):
log_ipv6_addr_route_by_interface(interface_name, title='RA Ready!')
return
logging.info(f'Host route not ready yet, retry {attempt}/{retries}...')
log_ipv6_addr_route_by_interface(interface_name, title=f'Wait RA ({attempt}/{retries})')
time.sleep(interval_s)
raise AssertionError('Host did not receive RA / OMR route in time')
def host_global_address_has_onlink_prefix(interface_name: str, onlinkprefix: str) -> bool:
onlinkprefix = onlinkprefix.strip()
if not onlinkprefix:
return False
base = onlinkprefix.rstrip(':')
try:
network = ipaddress.IPv6Network(f'{base}::/64', strict=False)
except ValueError:
logging.warning(f'Invalid onlinkprefix for /64 check: {onlinkprefix}')
return False
out = subprocess.getoutput(f'ip -6 addr show dev {interface_name}')
for line in out.splitlines():
if 'inet6' not in line or 'scope global' not in line:
continue
m = re.search(r'inet6 ([^\s]+)/\d+', line)
if not m:
continue
addr_s = m.group(1).split('%')[0]
try:
if ipaddress.IPv6Address(addr_s) in network:
return True
except ValueError:
continue
return False
def wait_for_host_onlink_global_address(
br: IdfDut,
*,
retries: int = 12,
interval_s: int = 5,
) -> str:
interface_name = get_host_interface_name()
onlinkprefix = get_onlinkprefix(br)
logging.info(f'Wait for host GUA in BR onlink prefix {onlinkprefix!r} on {interface_name}')
log_ipv6_addr_route_by_interface(interface_name, title='Wait onlink GUA (initial)')
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!')
return onlinkprefix
logging.info(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)
raise AssertionError(f'Host did not get a global IPv6 address in onlink prefix {onlinkprefix!r} in time')
thread_ipv6_group = 'ff04:0:0:0:0:0:0:125'
@@ -543,35 +616,72 @@ def open_host_interface() -> None:
assert flag
def ensure_avahi_running(restart_if_needed: bool = True) -> bool:
out_str = subprocess.getoutput('pgrep -a avahi-daemon 2>/dev/null')
if out_str.strip():
logging.info(f'avahi process list:\n{out_str}')
return True
logging.warning('avahi-daemon not running')
if restart_if_needed:
logging.warning('restarting avahi-daemon once...')
restart_avahi()
out_str = subprocess.getoutput('pgrep -a avahi-daemon 2>/dev/null')
if out_str.strip():
logging.info(f'avahi process list after restart:\n{out_str}')
return True
logging.error('avahi-daemon is still not running')
return False
def get_domain() -> str:
hostname = socket.gethostname()
logging.info(f'hostname is: {hostname}')
command = 'ps -auxww | grep avahi-daemon | grep running'
out_str = subprocess.getoutput(command)
logging.info(f'avahi status:\n {out_str}')
role = re.findall(r'\[([\w\W]+)\.local\]', str(out_str))[0]
logging.info(f'active host is: {role}')
return str(role)
out_str = subprocess.getoutput('pgrep -a avahi-daemon 2>/dev/null')
if not out_str.strip():
out_str = subprocess.getoutput('ps -C avahi-daemon -o args= --no-headers 2>/dev/null')
logging.info(f'avahi status:\n{out_str}')
matches = re.findall(r'\[([\w\W]+?)\.local\]', str(out_str))
if matches:
role = matches[0]
logging.info(f'active host is: {role}')
return str(role)
short = subprocess.getoutput('hostname -s').strip() or hostname.split('.')[0]
logging.warning(
'Could not parse [.local] from avahi process args; using short hostname %r',
short,
)
return short
def flush_ipv6_addr_by_interface() -> None:
interface_name = get_host_interface_name()
logging.info(f'flush ipv6 addr : {interface_name}')
def log_ipv6_addr_route_by_interface(interface_name: str, title: str = '') -> tuple[str, str]:
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)
logging.info(f'Before flush, IPv6 addresses: \n{addr_before}')
logging.info(f'Before flush, IPv6 routes: \n{route_before}')
addr = subprocess.getoutput(command_show_addr)
route = subprocess.getoutput(command_show_route)
prefix = f'{title} ' if title else ''
logging.info(f'{prefix}IPv6 addresses on {interface_name}:\n{addr}')
logging.info(f'{prefix}IPv6 routes on {interface_name}:\n{route}')
return addr, route
def flush_ipv6_addr_route_by_interface(interface_name: str, down_up_wait_s: int = 5) -> None:
logging.info(f'flush ipv6 addr/route: {interface_name}')
log_ipv6_addr_route_by_interface(interface_name, title='Before flush')
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])
subprocess.run(['ip', 'link', 'set', interface_name, 'up'])
time.sleep(5)
addr_after = subprocess.getoutput(command_show_addr)
route_after = subprocess.getoutput(command_show_route)
logging.info(f'After flush, IPv6 addresses: \n{addr_after}')
logging.info(f'After flush, IPv6 routes: \n{route_after}')
time.sleep(down_up_wait_s)
log_ipv6_addr_route_by_interface(interface_name, title='After flush')
def flush_ipv6_addr_by_interface() -> None:
interface_name = get_host_interface_name()
flush_ipv6_addr_route_by_interface(interface_name)
class tcp_parameter:
examples/openthread/pytest_otbr.py
+19 -13
View File
@@ -215,19 +215,22 @@ def test_Bidirectional_IPv6_connectivity(Init_interface: bool, dut: tuple[IdfDut
formBasicWiFiThreadNetwork(br, cli)
try:
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
onlinkprefix = ocf.wait_for_host_onlink_global_address(br)
logging.info(f'br onlinkprefix: {onlinkprefix}')
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()
ocf.log_ipv6_addr_route_by_interface(interface_name, title='Before ping test')
command = 'ping ' + str(cli_global_unicast_addr) + ' -c 10'
out_str = subprocess.getoutput(command)
ocf.log_ipv6_addr_route_by_interface(interface_name, title='After ping test')
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()
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')
onlinkprefix = ocf.get_onlinkprefix(br)
pattern = rf'\W+({onlinkprefix}(?:\w+:){{3}}\w+)\W+'
host_global_unicast_addr = re.findall(pattern, out_str)
logging.info(f'host_global_unicast_addr: {host_global_unicast_addr}')
@@ -272,7 +275,7 @@ def test_multicast_forwarding_A(Init_interface: bool, dut: tuple[IdfDut, IdfDut,
formBasicWiFiThreadNetwork(br, cli)
try:
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
assert ocf.thread_is_joined_group(cli)
@@ -325,7 +328,7 @@ def test_multicast_forwarding_B(Init_interface: bool, dut: tuple[IdfDut, IdfDut,
formBasicWiFiThreadNetwork(br, cli)
try:
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
ocf.execute_command(cli, 'udp open')
@@ -382,7 +385,7 @@ def test_service_discovery_of_Thread_device(
formBasicWiFiThreadNetwork(br, cli)
try:
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
command = 'avahi-browse -rt _testyyy._udp'
out_str = subprocess.getoutput(command)
logging.info(f'avahi-browse:\n{out_str}')
@@ -415,7 +418,7 @@ def test_service_discovery_of_Thread_device(
# Case 6: discover dervice published by Wi-Fi device
@pytest.mark.openthread_br
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.flaky(reruns=3, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
[
@@ -442,13 +445,15 @@ def test_service_discovery_of_WiFi_device(
dut[0].serial.stop_redirect_thread()
formBasicWiFiThreadNetwork(br, cli)
sp: subprocess.Popen | None = None
try:
assert ocf.is_joined_wifi_network(br)
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)
cli.expect('Done', timeout=5)
ocf.wait(cli, 1)
assert ocf.ensure_avahi_running(restart_if_needed=True), 'avahi-daemon is not running on this runner'
domain_name = ocf.get_domain()
logging.info(f'domain name is: {domain_name}')
command = 'dns resolve ' + domain_name + '.default.service.arpa.'
@@ -477,7 +482,8 @@ def test_service_discovery_of_WiFi_device(
assert 'Port:12347' in str(tmp)
finally:
ocf.host_close_service()
sp.terminate()
if sp is not None:
sp.terminate()
ocf.stop_thread(cli)
ocf.stop_thread(br)
time.sleep(3)
@@ -510,7 +516,7 @@ def test_ICMP_NAT64(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) ->
formBasicWiFiThreadNetwork(br, cli)
try:
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
host_ipv4_address = ocf.get_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]
@@ -548,7 +554,7 @@ def test_UDP_NAT64(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) ->
formBasicWiFiThreadNetwork(br, cli)
try:
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
ocf.execute_command(cli, 'udp open')
@@ -604,7 +610,7 @@ def test_TCP_NAT64(Init_interface: bool, dut: tuple[IdfDut, IdfDut, IdfDut]) ->
formBasicWiFiThreadNetwork(br, cli)
try:
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5)
ocf.execute_command(cli, 'tcpsockclient open')
@@ -832,7 +838,7 @@ def test_br_meshcop(Init_interface: bool, Init_avahi: bool, dut: tuple[IdfDut, I
br_thread_para.setnetworkname(networkname)
ocf.joinThreadNetwork(br, br_thread_para)
ocf.wait(br, 10)
assert ocf.is_joined_wifi_network(br)
ocf.wait_for_host_ra_route(br)
command = 'timeout 3 avahi-browse -r _meshcop._udp'
try:
result = subprocess.run(command, capture_output=True, check=True, shell=True)