Merge branch 'bugfix/modify_state_check_method' into 'master'

OpenThread CI: add a function for executing commands

See merge request espressif/esp-idf!23529

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)

examples/openthread/pytest_otbr.py

@@ -112,9 +112,9 @@ def test_thread_connect(dut:Tuple[IdfDut, IdfDut, IdfDut]) -> None:
             rx_nums = ocf.ot_ping(br, cli_mleid_addr, 5)[1]
             assert rx_nums == 5
     finally:
-        br.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
         for cli in cli_list:
-            cli.write('factoryreset')
+            ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
 
 
@@ -141,7 +141,7 @@ def formBasicWiFiThreadNetwork(br:IdfDut, cli:IdfDut) -> None:
 @pytest.mark.esp32h2
 @pytest.mark.esp32c6
 @pytest.mark.openthread_br
-@pytest.mark.flaky(reruns=1, reruns_delay=1)
+@pytest.mark.flaky(reruns=0, reruns_delay=1)
 @pytest.mark.parametrize(
     'config, count, app_path, target', [
         ('rcp|cli_h2|br', 3,
@@ -169,18 +169,19 @@ def test_Bidirectional_IPv6_connectivity(Init_interface:bool, dut: Tuple[IdfDut,
         role = re.findall(r' (\d+)%', str(out_str))[0]
         assert role != '100'
         interface_name = ocf.get_host_interface_name()
-        command = 'ifconfig ' + 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')
-        host_global_unicast_addr = re.findall(r'inet6 ((?:\w+:){7}\w+)  prefixlen 64  scopeid 0x0<global>', str(out_str))
+        onlinkprefix = ocf.get_onlinkprefix(br)
+        host_global_unicast_addr = re.findall(r'\W+(%s(?:\w+:){3}\w+)\W+' % onlinkprefix, str(out_str))
         rx_nums = 0
         for ip_addr in host_global_unicast_addr:
             txrx_nums = ocf.ot_ping(cli, str(ip_addr), 5)
             rx_nums = rx_nums + int(txrx_nums[1])
         assert rx_nums != 0
     finally:
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
 
 
@@ -209,7 +210,7 @@ def test_multicast_forwarding_A(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
     formBasicWiFiThreadNetwork(br, cli)
     try:
         assert ocf.is_joined_wifi_network(br)
-        br.write('bbr')
+        ocf.execute_command(br, 'bbr')
         br.expect('server16', timeout=5)
         assert ocf.thread_is_joined_group(cli)
         interface_name = ocf.get_host_interface_name()
@@ -219,8 +220,8 @@ def test_multicast_forwarding_A(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
         role = re.findall(r' (\d+)%', str(out_str))[0]
         assert role != '100'
     finally:
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
 
 
@@ -249,9 +250,9 @@ def test_multicast_forwarding_B(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
     formBasicWiFiThreadNetwork(br, cli)
     try:
         assert ocf.is_joined_wifi_network(br)
-        br.write('bbr')
+        ocf.execute_command(br, 'bbr')
         br.expect('server16', timeout=5)
-        cli.write('udp open')
+        ocf.execute_command(cli, 'udp open')
         cli.expect('Done', timeout=5)
         ocf.wait(cli, 3)
         myudp = ocf.udp_parameter('INET6', '::', 5090, 'ff04::125', False, 15.0, b'')
@@ -263,14 +264,14 @@ def test_multicast_forwarding_B(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
                 assert False
         for num in range(0, 3):
             command = 'udp send ff04::125 5090 hello' + str(num)
-            cli.write(command)
+            ocf.execute_command(cli, command)
             cli.expect('Done', timeout=5)
             ocf.wait(cli, 0.5)
         while udp_mission.is_alive():
             time.sleep(1)
     finally:
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
     assert b'hello' in myudp.udp_bytes
 
@@ -307,18 +308,18 @@ def test_service_discovery_of_Thread_device(Init_interface:bool, Init_avahi:bool
         assert 'myTest' not in str(out_str)
         hostname = 'myTest'
         command = 'srp client host name ' + hostname
-        cli.write(command)
+        ocf.execute_command(cli, command)
         cli.expect('Done', timeout=5)
         cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br)
         print('cli_global_unicast_addr', cli_global_unicast_addr)
         command = 'srp client host address ' + str(cli_global_unicast_addr)
-        cli.write(command)
+        ocf.execute_command(cli, command)
         cli.expect('Done', timeout=5)
         port = '12348'
         command = 'srp client service add my-service _testyyy._udp ' + port
-        cli.write(command)
+        ocf.execute_command(cli, command)
         cli.expect('Done', timeout=5)
-        cli.write('srp client autostart enable')
+        ocf.execute_command(cli, 'srp client autostart enable')
         cli.expect('Done', timeout=5)
         ocf.wait(cli, 3)
         command = 'avahi-browse -rt _testyyy._udp'
@@ -326,8 +327,8 @@ def test_service_discovery_of_Thread_device(Init_interface:bool, Init_avahi:bool
         print('avahi-browse:\n', str(out_str))
         assert 'myTest' in str(out_str)
     finally:
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
 
 
@@ -359,41 +360,42 @@ def test_service_discovery_of_WiFi_device(Init_interface:bool, Init_avahi:bool,
         assert ocf.is_joined_wifi_network(br)
         br_global_unicast_addr = ocf.get_global_unicast_addr(br, br)
         command = 'dns config ' + br_global_unicast_addr
-        ocf.clean_buffer(cli)
-        cli.write(command)
+        ocf.execute_command(cli, command)
         cli.expect('Done', timeout=5)
         ocf.wait(cli, 1)
         command = 'dns resolve FA000123.default.service.arpa.'
         ocf.clean_buffer(cli)
-        cli.write(command)
+        ocf.execute_command(cli, command)
         cli.expect('Error', timeout=15)
         domain_name = ocf.get_domain()
         print('domain name is: ', domain_name)
         command = 'dns resolve ' + domain_name + '.default.service.arpa.'
-        ocf.clean_buffer(cli)
-        cli.write(command)
+
+        ocf.execute_command(cli, command)
         cli.expect('TTL', timeout=10)
         cli.expect('Done', timeout=10)
-        ocf.clean_buffer(cli)
-        cli.write('dns browse _testxxx._udp.default.service.arpa')
-        tmp = cli.expect(pexpect.TIMEOUT, timeout=5)
+
+        command = 'dns browse _testxxx._udp.default.service.arpa'
+        tmp = ocf.get_ouput_string(cli, command, 5)
         assert 'Port:12347' not in str(tmp)
         ocf.host_publish_service()
         ocf.wait(cli, 5)
-        ocf.clean_buffer(cli)
-        cli.write('dns browse _testxxx._udp.default.service.arpa')
-        tmp = cli.expect(pexpect.TIMEOUT, timeout=5)
+
+        command = 'dns browse _testxxx._udp.default.service.arpa'
+        tmp = ocf.get_ouput_string(cli, command, 5)
         assert 'response for _testxxx' in str(tmp)
         assert 'Port:12347' in str(tmp)
-        ocf.clean_buffer(cli)
-        cli.write('dns service testxxx _testxxx._udp.default.service.arpa.')
+
+        command = 'dns browse _testxxx._udp.default.service.arpa'
+        tmp = ocf.get_ouput_string(cli, command, 5)
+        ocf.execute_command(cli, 'dns service testxxx _testxxx._udp.default.service.arpa.')
         tmp = cli.expect(pexpect.TIMEOUT, timeout=5)
         assert 'response for testxxx' in str(tmp)
         assert 'Port:12347' in str(tmp)
     finally:
         ocf.host_close_service()
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
 
 
@@ -427,8 +429,8 @@ def test_ICMP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) ->
         rx_nums = ocf.ot_ping(cli, str(host_ipv4_address), 5)[1]
         assert rx_nums != 0
     finally:
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
 
 
@@ -457,9 +459,9 @@ def test_UDP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N
     formBasicWiFiThreadNetwork(br, cli)
     try:
         assert ocf.is_joined_wifi_network(br)
-        br.write('bbr')
+        ocf.execute_command(br, 'bbr')
         br.expect('server16', timeout=5)
-        cli.write('udp open')
+        ocf.execute_command(cli, 'udp open')
         cli.expect('Done', timeout=5)
         ocf.wait(cli, 3)
         host_ipv4_address = ocf.get_host_ipv4_address()
@@ -473,14 +475,14 @@ def test_UDP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N
                 assert False
         for num in range(0, 3):
             command = 'udp send ' + host_ipv4_address + ' 5090 hello' + str(num)
-            cli.write(command)
+            ocf.execute_command(cli, command)
             cli.expect('Done', timeout=5)
             ocf.wait(cli, 0.5)
         while udp_mission.is_alive():
             time.sleep(1)
     finally:
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
     assert b'hello' in myudp.udp_bytes
 
@@ -510,9 +512,9 @@ def test_TCP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N
     formBasicWiFiThreadNetwork(br, cli)
     try:
         assert ocf.is_joined_wifi_network(br)
-        br.write('bbr')
+        ocf.execute_command(br, 'bbr')
         br.expect('server16', timeout=5)
-        cli.write('tcpsockclient open')
+        ocf.execute_command(cli, 'tcpsockclient open')
         cli.expect('Done', timeout=5)
         ocf.wait(cli, 3)
         host_ipv4_address = ocf.get_host_ipv4_address()
@@ -526,19 +528,19 @@ def test_TCP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N
             if (time.time() - start_time) > 10:
                 assert False
         command = 'tcpsockclient connect ' + connect_address + ' 12345'
-        cli.write(command)
+        ocf.execute_command(cli, command)
         cli.expect('Successfully connected', timeout=10)
         start_time = time.time()
         while not mytcp.recv_flag:
             if (time.time() - start_time) > 10:
                 assert False
         command = 'tcpsockclient send hello'
-        cli.write(command)
+        ocf.execute_command(cli, command)
         cli.expect('Done', timeout=5)
         while tcp_mission.is_alive():
             time.sleep(1)
     finally:
-        br.write('factoryreset')
-        cli.write('factoryreset')
+        ocf.execute_command(br, 'factoryreset')
+        ocf.execute_command(cli, 'factoryreset')
         time.sleep(3)
     assert b'hello' in mytcp.tcp_bytes