fix(openthread): wait for all onlink GUAs to finish DAD before tests

This commit is contained in:
Xu Si Yu
2026-05-19 17:39:41 +08:00
parent 8f3d8b41c3
commit 0ec7c487bc
2 changed files with 28 additions and 11 deletions
+27 -6
View File
@@ -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(