diff --git a/examples/openthread/ot_ci_function.py b/examples/openthread/ot_ci_function.py index 74b84998a36..5233cc14bc4 100644 --- a/examples/openthread/ot_ci_function.py +++ b/examples/openthread/ot_ci_function.py @@ -413,17 +413,19 @@ def wait_for_host_ra_route( 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: +def _list_host_onlink_global_address_entries(interface_name: str, onlinkprefix: str) -> list[tuple[str, bool]]: + """Return (address, is_usable) for each global address in the onlink /64.""" onlinkprefix = onlinkprefix.strip() if not onlinkprefix: - return False + return [] 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 + return [] + entries: dict[str, bool] = {} 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: @@ -433,11 +435,30 @@ def host_global_address_has_onlink_prefix(interface_name: str, onlinkprefix: str continue addr_s = m.group(1).split('%')[0] try: - if ipaddress.IPv6Address(addr_s) in network: - return True + if ipaddress.IPv6Address(addr_s) not in network: + continue except ValueError: continue - return False + is_usable = 'tentative' not in line and 'dadfailed' not in line + if addr_s in entries: + entries[addr_s] = entries[addr_s] and is_usable + else: + entries[addr_s] = is_usable + return list(entries.items()) + + +def host_global_address_has_onlink_prefix(interface_name: str, onlinkprefix: str) -> bool: + entries = _list_host_onlink_global_address_entries(interface_name, onlinkprefix) + if not entries: + return False + return all(usable for _, usable in entries) + + +def list_host_usable_onlink_global_addresses(interface_name: str, onlinkprefix: str) -> list[str]: + entries = _list_host_onlink_global_address_entries(interface_name, onlinkprefix) + if not entries or not all(usable for _, usable in entries): + return [] + return [addr for addr, _ in entries] def wait_for_host_onlink_global_address( diff --git a/examples/openthread/pytest_otbr.py b/examples/openthread/pytest_otbr.py index b2cf83bd8d0..26aca26040d 100644 --- a/examples/openthread/pytest_otbr.py +++ b/examples/openthread/pytest_otbr.py @@ -246,11 +246,7 @@ 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 = 'ifconfig ' + interface_name + ' | grep inet6 | grep global' - out_bytes = subprocess.check_output(command, shell=True, timeout=5) - out_str = out_bytes.decode('utf-8') - pattern = rf'\W+({onlinkprefix}(?:\w+:){{3}}\w+)\W+' - host_global_unicast_addr = re.findall(pattern, out_str) + host_global_unicast_addr = ocf.list_host_usable_onlink_global_addresses(interface_name, onlinkprefix) 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}')