OpenThread CI: add a function for executing commands

examples/openthread/ot_ci_function.py

@@ -36,37 +36,40 @@ class wifi_parameter:
 def joinThreadNetwork(dut:IdfDut, thread:thread_parameter) -> None:
     if thread.dataset != '':
         command = 'dataset set active ' + thread.dataset
-        dut.write(command)
+        execute_command(dut, command)
         dut.expect('Done', timeout=5)
     else:
-        dut.write('dataset init new')
+        execute_command(dut, 'dataset init new')
         dut.expect('Done', timeout=5)
-        dut.write('dataset commit active')
+        execute_command(dut, 'dataset commit active')
         dut.expect('Done', timeout=5)
     if thread.channel != '':
         command = 'channel ' + thread.channel
-        dut.write(command)
+        execute_command(dut, command)
         dut.expect('Done', timeout=5)
     if thread.exaddr != '':
         command = 'extaddr ' + thread.exaddr
-        dut.write(command)
+        execute_command(dut, command)
         dut.expect('Done', timeout=5)
     if thread.bbr:
-        dut.write('bbr enable')
+        execute_command(dut, 'bbr enable')
         dut.expect('Done', timeout=5)
-    dut.write('ifconfig up')
+    if thread.deviceRole == 'router':
+        execute_command(dut, 'routerselectionjitter 1')
+        dut.expect('Done', timeout=5)
+    execute_command(dut, 'ifconfig up')
     dut.expect('Done', timeout=5)
-    dut.write('thread start')
-    dut.expect('Role detached ->', timeout=20)
-    if thread.deviceRole == 'leader':
-        assert getDeviceRole(dut) == 'leader'
-    elif thread.deviceRole == 'router':
-        if getDeviceRole(dut) != 'router':
-            changeDeviceRole(dut, 'router')
-            wait(dut, 10)
-            assert getDeviceRole(dut) == 'router'
-    else:
-        assert False
+    execute_command(dut, 'thread start')
+    assert wait_for_join(dut, thread.deviceRole)
+
+
+def wait_for_join(dut:IdfDut, role:str) -> bool:
+    for _ in range(1, 30):
+        if getDeviceRole(dut) == role:
+            wait(dut, 5)
+            return True
+        wait(dut, 1)
+    return False
 
 
 def joinWiFiNetwork(dut:IdfDut, wifi:wifi_parameter) -> Tuple[str, int]:
@@ -74,11 +77,12 @@ def joinWiFiNetwork(dut:IdfDut, wifi:wifi_parameter) -> Tuple[str, int]:
     ip_address = ''
     information = ''
     for order in range(1, wifi.retry_times):
-        dut.write('wifi connect -s ' + str(wifi.ssid) + ' -p ' + str(wifi.psk))
-        tmp = dut.expect(pexpect.TIMEOUT, timeout=10)
+        command = 'wifi connect -s ' + str(wifi.ssid) + ' -p ' + str(wifi.psk)
+        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]
-        information = dut.expect(r'wifi sta (\w+ \w+ \w+)\W', timeout=20)[1].decode()
+        if 'wifi sta' in str(tmp):
+            information = re.findall(r'wifi sta (\w+ \w+ \w+)\W', str(tmp))[0]
         if information == 'is connected successfully':
             break
     assert information == 'is connected successfully'
@@ -87,38 +91,40 @@ def joinWiFiNetwork(dut:IdfDut, wifi:wifi_parameter) -> Tuple[str, int]:
 
 def getDeviceRole(dut:IdfDut) -> str:
     clean_buffer(dut)
-    dut.write('state')
-    role = dut.expect(r'state\W+(\w+)\W+Done', timeout=5)[1].decode()
+    execute_command(dut, 'state')
+    role = dut.expect(r'\W+(\w+)\W+Done', timeout=5)[1].decode()
     print(role)
     return str(role)
 
 
 def changeDeviceRole(dut:IdfDut, role:str) -> None:
     command = 'state ' + role
-    dut.write(command)
+    execute_command(dut, command)
 
 
 def getDataset(dut:IdfDut) -> str:
     clean_buffer(dut)
-    dut.write('dataset active -x')
+    execute_command(dut, 'dataset active -x')
     dut_data = dut.expect(r'\n(\w{212})\r', timeout=5)[1].decode()
     return str(dut_data)
 
 
 def reset_thread(dut:IdfDut) -> None:
-    dut.expect('>')
+    dut.expect('>', timeout=10)
     wait(dut, 3)
-    dut.write('factoryreset')
+    clean_buffer(dut)
+    execute_command(dut, 'factoryreset')
     dut.expect('OpenThread attached to netif', timeout=20)
-    dut.expect('>')
+    dut.expect('>', timeout=10)
     wait(dut, 3)
+    clean_buffer(dut)
 
 
 # get the mleid address of the thread
 def get_mleid_addr(dut:IdfDut) -> str:
     dut_adress = ''
     clean_buffer(dut)
-    dut.write('ipaddr mleid')
+    execute_command(dut, 'ipaddr mleid')
     dut_adress = dut.expect(r'\n((?:\w+:){7}\w+)\r', timeout=5)[1].decode()
     return dut_adress
 
@@ -127,7 +133,7 @@ def get_mleid_addr(dut:IdfDut) -> str:
 def get_rloc_addr(dut:IdfDut) -> str:
     dut_adress = ''
     clean_buffer(dut)
-    dut.write('ipaddr rloc')
+    execute_command(dut, 'ipaddr rloc')
     dut_adress = dut.expect(r'\n((?:\w+:){7}\w+)\r', timeout=5)[1].decode()
     return dut_adress
 
@@ -136,7 +142,7 @@ def get_rloc_addr(dut:IdfDut) -> str:
 def get_linklocal_addr(dut:IdfDut) -> str:
     dut_adress = ''
     clean_buffer(dut)
-    dut.write('ipaddr linklocal')
+    execute_command(dut, 'ipaddr linklocal')
     dut_adress = dut.expect(r'\n((?:\w+:){7}\w+)\r', timeout=5)[1].decode()
     return dut_adress
 
@@ -147,7 +153,7 @@ def get_global_unicast_addr(dut:IdfDut, br:IdfDut) -> str:
     clean_buffer(br)
     omrprefix = get_omrprefix(br)
     clean_buffer(dut)
-    dut.write('ipaddr')
+    execute_command(dut, 'ipaddr')
     dut_adress = dut.expect(r'(%s(?:\w+:){3}\w+)\r' % str(omrprefix), timeout=5)[1].decode()
     return dut_adress
 
@@ -155,7 +161,7 @@ def get_global_unicast_addr(dut:IdfDut, br:IdfDut) -> str:
 # ping of thread
 def ot_ping(dut:IdfDut, target:str, times:int) -> Tuple[int, int]:
     command = 'ping ' + str(target) + ' 0 ' + str(times)
-    dut.write(command)
+    execute_command(dut, command)
     transmitted = dut.expect(r'(\d+) packets transmitted', timeout=30)[1].decode()
     tx_count = int(transmitted)
     received = dut.expect(r'(\d+) packets received', timeout=30)[1].decode()
@@ -250,9 +256,7 @@ thread_ipv6_group = 'ff04:0:0:0:0:0:0:125'
 
 
 def check_ipmaddr(dut:IdfDut) -> bool:
-    clean_buffer(dut)
-    dut.write('ipmaddr')
-    info = dut.expect(pexpect.TIMEOUT, timeout=2)
+    info = get_ouput_string(dut, 'ipmaddr', 2)
     if thread_ipv6_group in str(info):
         return True
     return False
@@ -260,13 +264,13 @@ def check_ipmaddr(dut:IdfDut) -> bool:
 
 def thread_is_joined_group(dut:IdfDut) -> bool:
     command = 'mcast join ' + thread_ipv6_group
-    dut.write(command)
+    execute_command(dut, command)
     dut.expect('Done', timeout=5)
     order = 0
     while order < 3:
         if check_ipmaddr(dut):
             return True
-        dut.write(command)
+        execute_command(dut, command)
         wait(dut, 2)
         order = order + 1
     return False
@@ -451,12 +455,34 @@ def decimal_to_hex(decimal_str:str) -> str:
 
 
 def get_omrprefix(br:IdfDut) -> str:
-    br.write('br omrprefix')
+    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:
-    br.write('br nat64prefix')
+    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:
+    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)
+    return str(tmp)