From 938bcc0337bfe0da75931fb12e1d59fedf04920b Mon Sep 17 00:00:00 2001 From: Xu Si Yu Date: Fri, 28 Apr 2023 18:56:01 +0800 Subject: [PATCH] OpenThread CI: add a function for executing commands --- examples/openthread/ot_ci_function.py | 106 ++++++++++++++++---------- examples/openthread/pytest_otbr.py | 100 ++++++++++++------------ 2 files changed, 117 insertions(+), 89 deletions(-) diff --git a/examples/openthread/ot_ci_function.py b/examples/openthread/ot_ci_function.py index d8f634fdab..6cd68793d1 100644 --- a/examples/openthread/ot_ci_function.py +++ b/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) diff --git a/examples/openthread/pytest_otbr.py b/examples/openthread/pytest_otbr.py index 3e8a1eba13..7af4661aac 100644 --- a/examples/openthread/pytest_otbr.py +++ b/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', 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