From 74b735dde54d0c7c3adc3dd62cf509ad1630c163 Mon Sep 17 00:00:00 2001
From: Xu Si Yu <xusiyu@espressif.com>
Date: Tue, 19 May 2026 11:55:51 +0800
Subject: [PATCH] fix(openthread): wait for all onlink GUAs to finish DAD
 before tests

---
 examples/openthread/ot_ci_function.py | 35 ++++++++++++++++++++++-----
 examples/openthread/pytest_otbr.py    |  6 +----
 2 files changed, 30 insertions(+), 11 deletions(-)

diff --git a/examples/openthread/ot_ci_function.py b/examples/openthread/ot_ci_function.py
index ac65e624b8e..3dd15fd7616 100644
--- a/examples/openthread/ot_ci_function.py
+++ b/examples/openthread/ot_ci_function.py
@@ -12,6 +12,8 @@ import subprocess
 import time
 from functools import wraps
 from typing import Callable
+from typing import Dict
+from typing import List
 from typing import Optional
 from typing import Tuple
 
@@ -416,17 +418,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:
@@ -436,11 +440,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 eef30af8b8f..700ab2e1344 100644
--- a/examples/openthread/pytest_otbr.py
+++ b/examples/openthread/pytest_otbr.py
@@ -227,11 +227,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}')