Merge branch 'fix/optimize_the_ot_ci' into 'master'

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

See merge request espressif/esp-idf!48676

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(

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}')