feat(openthread): add delay before execute wifi command in openthread ci cases

examples/openthread/ot_ci_function.py

@@ -119,14 +119,15 @@ def joinWiFiNetwork(dut:IdfDut, wifi:wifi_parameter) -> Tuple[str, int]:
         tmp = get_ouput_string(dut, command, 10)
         if 'sta ip' in str(tmp):
             ip_address = re.findall(r'sta ip: (\w+.\w+.\w+.\w+),', str(tmp))[0]
+        wait(dut, 2)
         execute_command(dut, 'wifi state')
-        if dut.expect('\nconnected\r', timeout=5):
+        if dut.expect('connected', timeout=5):
             return ip_address, order
     raise Exception(f'{dut} connect wifi {str(wifi.ssid)} with password {str(wifi.psk)} fail')
 
 
 def getDeviceRole(dut:IdfDut) -> str:
-    clean_buffer(dut)
+    wait(dut, 1)
     execute_command(dut, 'state')
     role = dut.expect(r'\W+(\w+)\W+Done', timeout=5)[1].decode()
     print(role)
@@ -139,7 +140,6 @@ def changeDeviceRole(dut:IdfDut, role:str) -> None:
 
 
 def getDataset(dut:IdfDut) -> str:
-    clean_buffer(dut)
     execute_command(dut, 'dataset active -x')
     dut_data = dut.expect(r'\n(\w+)\r', timeout=5)[1].decode()
     return str(dut_data)
@@ -152,7 +152,6 @@ def init_thread(dut:IdfDut) -> None:
 
 
 def reset_thread(dut:IdfDut) -> None:
-    clean_buffer(dut)
     execute_command(dut, 'factoryreset')
     dut.expect('OpenThread attached to netif', timeout=20)
     dut.expect('>', timeout=10)
@@ -163,7 +162,6 @@ def reset_thread(dut:IdfDut) -> None:
 # get the mleid address of the thread
 def get_mleid_addr(dut:IdfDut) -> str:
     dut_adress = ''
-    clean_buffer(dut)
     execute_command(dut, 'ipaddr mleid')
     dut_adress = dut.expect(r'\n((?:\w+:){7}\w+)\r', timeout=5)[1].decode()
     return dut_adress
@@ -172,7 +170,6 @@ def get_mleid_addr(dut:IdfDut) -> str:
 # get the rloc address of the thread
 def get_rloc_addr(dut:IdfDut) -> str:
     dut_adress = ''
-    clean_buffer(dut)
     execute_command(dut, 'ipaddr rloc')
     dut_adress = dut.expect(r'\n((?:\w+:){7}\w+)\r', timeout=5)[1].decode()
     return dut_adress
@@ -181,7 +178,6 @@ def get_rloc_addr(dut:IdfDut) -> str:
 # get the linklocal address of the thread
 def get_linklocal_addr(dut:IdfDut) -> str:
     dut_adress = ''
-    clean_buffer(dut)
     execute_command(dut, 'ipaddr linklocal')
     dut_adress = dut.expect(r'\n((?:\w+:){7}\w+)\r', timeout=5)[1].decode()
     return dut_adress
@@ -192,7 +188,6 @@ def get_global_unicast_addr(dut:IdfDut, br:IdfDut) -> str:
     dut_adress = ''
     clean_buffer(br)
     omrprefix = get_omrprefix(br)
-    clean_buffer(dut)
     execute_command(dut, 'ipaddr')
     dut_adress = dut.expect(r'(%s(?:\w+:){3}\w+)\r' % str(omrprefix), timeout=5)[1].decode()
     return dut_adress
@@ -537,33 +532,29 @@ def decimal_to_hex(decimal_str:str) -> str:
 
 
 def get_omrprefix(br:IdfDut) -> str:
-    clean_buffer(br)
     execute_command(br, 'br omrprefix')
     omrprefix = br.expect(r'Local: ((?:\w+:){4}):/\d+\r', timeout=5)[1].decode()
     return str(omrprefix)
 
 
 def get_onlinkprefix(br:IdfDut) -> str:
-    clean_buffer(br)
     execute_command(br, 'br onlinkprefix')
     onlinkprefix = br.expect(r'Local: ((?:\w+:){4}):/\d+\r', timeout=5)[1].decode()
     return str(onlinkprefix)
 
 
 def get_nat64prefix(br:IdfDut) -> str:
-    clean_buffer(br)
     execute_command(br, 'br nat64prefix')
     nat64prefix = br.expect(r'Local: ((?:\w+:){6}):/\d+', timeout=5)[1].decode()
     return str(nat64prefix)
 
 
 def execute_command(dut:IdfDut, command:str) -> None:
+    clean_buffer(dut)
     dut.write(command)
-    dut.expect(command, timeout=3)
 
 
 def get_ouput_string(dut:IdfDut, command:str, wait_time:int) -> str:
-    clean_buffer(dut)
     execute_command(dut, command)
     tmp = dut.expect(pexpect.TIMEOUT, timeout=wait_time)
     clean_buffer(dut)

examples/openthread/pytest_otbr.py

@@ -618,7 +618,6 @@ def test_ot_sleepy_device(dut: Tuple[IdfDut, IdfDut]) -> None:
         assert not bool(fail_info.search(str(info)))
         output = sleepy_device.expect(pexpect.TIMEOUT, timeout=20)
         assert not bool(fail_info.search(str(output)))
-        ocf.clean_buffer(sleepy_device)
         ocf.execute_command(leader, 'factoryreset')
         output = sleepy_device.expect(pexpect.TIMEOUT, timeout=5)
         assert not bool(fail_info.search(str(output)))
@@ -648,7 +647,6 @@ def test_basic_startup(dut: Tuple[IdfDut, IdfDut]) -> None:
     try:
         ocf.init_thread(br)
         time.sleep(3)
-        ocf.clean_buffer(br)
         ocf.execute_command(br, 'ifconfig up')
         br.expect('Done', timeout=5)
         ocf.execute_command(br, 'thread start')