alex/esp-idf
alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2024-10-05 20:47:46 -04:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

OpenThread CI: add a function for executing commands

Browse Source
This commit is contained in:
Xu Si Yu 2023-04-28 18:56:01 +08:00
parent 2004bf4e11
commit 938bcc0337
2 changed files with 117 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
examples/openthread/ot_ci_function.py
View File

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

100
examples/openthread/pytest_otbr.py
View File

@ -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] rx_nums = ocf.ot_ping(br, cli_mleid_addr, 5)[1]
assert rx_nums == 5 assert rx_nums == 5
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
for cli in cli_list: for cli in cli_list:
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) time.sleep(3)
@ -141,7 +141,7 @@ def formBasicWiFiThreadNetwork(br:IdfDut, cli:IdfDut) -> None:
@pytest.mark.esp32h2 @pytest.mark.esp32h2
@pytest.mark.esp32c6 @pytest.mark.esp32c6
@pytest.mark.openthread_br @pytest.mark.openthread_br
@pytest.mark.flaky(reruns=1, reruns_delay=1) @pytest.mark.flaky(reruns=0, reruns_delay=1)
@pytest.mark.parametrize( @pytest.mark.parametrize(
'config, count, app_path, target', [ 'config, count, app_path, target', [
('rcp|cli_h2|br', 3, ('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] role = re.findall(r' (\d+)%', str(out_str))[0]
assert role != '100' assert role != '100'
interface_name = ocf.get_host_interface_name() 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_bytes = subprocess.check_output(command, shell=True, timeout=5)
out_str = out_bytes.decode('utf-8') 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 rx_nums = 0
for ip_addr in host_global_unicast_addr: for ip_addr in host_global_unicast_addr:
txrx_nums = ocf.ot_ping(cli, str(ip_addr), 5) txrx_nums = ocf.ot_ping(cli, str(ip_addr), 5)
rx_nums = rx_nums + int(txrx_nums[1]) rx_nums = rx_nums + int(txrx_nums[1])
assert rx_nums != 0 assert rx_nums != 0
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) time.sleep(3)
@ -209,7 +210,7 @@ def test_multicast_forwarding_A(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
formBasicWiFiThreadNetwork(br, cli) formBasicWiFiThreadNetwork(br, cli)
try: try:
assert ocf.is_joined_wifi_network(br) assert ocf.is_joined_wifi_network(br)
br.write('bbr') ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5) br.expect('server16', timeout=5)
assert ocf.thread_is_joined_group(cli) assert ocf.thread_is_joined_group(cli)
interface_name = ocf.get_host_interface_name() 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] role = re.findall(r' (\d+)%', str(out_str))[0]
assert role != '100' assert role != '100'
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) time.sleep(3)
@ -249,9 +250,9 @@ def test_multicast_forwarding_B(Init_interface:bool, dut: Tuple[IdfDut, IdfDut,
formBasicWiFiThreadNetwork(br, cli) formBasicWiFiThreadNetwork(br, cli)
try: try:
assert ocf.is_joined_wifi_network(br) assert ocf.is_joined_wifi_network(br)
br.write('bbr') ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5) br.expect('server16', timeout=5)
cli.write('udp open') ocf.execute_command(cli, 'udp open')
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
ocf.wait(cli, 3) ocf.wait(cli, 3)
myudp = ocf.udp_parameter('INET6', '::', 5090, 'ff04::125', False, 15.0, b'') 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 assert False
for num in range(0, 3): for num in range(0, 3):
command = 'udp send ff04::125 5090 hello' + str(num) command = 'udp send ff04::125 5090 hello' + str(num)
cli.write(command) ocf.execute_command(cli, command)
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
ocf.wait(cli, 0.5) ocf.wait(cli, 0.5)
while udp_mission.is_alive(): while udp_mission.is_alive():
time.sleep(1) time.sleep(1)
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) time.sleep(3)
assert b'hello' in myudp.udp_bytes 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) assert 'myTest' not in str(out_str)
hostname = 'myTest' hostname = 'myTest'
command = 'srp client host name ' + hostname command = 'srp client host name ' + hostname
cli.write(command) ocf.execute_command(cli, command)
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br) cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br)
print('cli_global_unicast_addr', cli_global_unicast_addr) print('cli_global_unicast_addr', cli_global_unicast_addr)
command = 'srp client host address ' + str(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) cli.expect('Done', timeout=5)
port = '12348' port = '12348'
command = 'srp client service add my-service _testyyy._udp ' + port command = 'srp client service add my-service _testyyy._udp ' + port
cli.write(command) ocf.execute_command(cli, command)
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
cli.write('srp client autostart enable') ocf.execute_command(cli, 'srp client autostart enable')
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
ocf.wait(cli, 3) ocf.wait(cli, 3)
command = 'avahi-browse -rt _testyyy._udp' 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)) print('avahi-browse:\n', str(out_str))
assert 'myTest' in str(out_str) assert 'myTest' in str(out_str)
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) 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) assert ocf.is_joined_wifi_network(br)
br_global_unicast_addr = ocf.get_global_unicast_addr(br, br) br_global_unicast_addr = ocf.get_global_unicast_addr(br, br)
command = 'dns config ' + br_global_unicast_addr command = 'dns config ' + br_global_unicast_addr
ocf.clean_buffer(cli) ocf.execute_command(cli, command)
cli.write(command)
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
ocf.wait(cli, 1) ocf.wait(cli, 1)
command = 'dns resolve FA000123.default.service.arpa.' command = 'dns resolve FA000123.default.service.arpa.'
ocf.clean_buffer(cli) ocf.clean_buffer(cli)
cli.write(command) ocf.execute_command(cli, command)
cli.expect('Error', timeout=15) cli.expect('Error', timeout=15)
domain_name = ocf.get_domain() domain_name = ocf.get_domain()
print('domain name is: ', domain_name) print('domain name is: ', domain_name)
command = 'dns resolve ' + domain_name + '.default.service.arpa.' 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('TTL', timeout=10)
cli.expect('Done', timeout=10) cli.expect('Done', timeout=10)
ocf.clean_buffer(cli)
cli.write('dns browse _testxxx._udp.default.service.arpa') command = 'dns browse _testxxx._udp.default.service.arpa'
tmp = cli.expect(pexpect.TIMEOUT, timeout=5) tmp = ocf.get_ouput_string(cli, command, 5)
assert 'Port:12347' not in str(tmp) assert 'Port:12347' not in str(tmp)
ocf.host_publish_service() ocf.host_publish_service()
ocf.wait(cli, 5) ocf.wait(cli, 5)
ocf.clean_buffer(cli)
cli.write('dns browse _testxxx._udp.default.service.arpa') command = 'dns browse _testxxx._udp.default.service.arpa'
tmp = cli.expect(pexpect.TIMEOUT, timeout=5) tmp = ocf.get_ouput_string(cli, command, 5)
assert 'response for _testxxx' in str(tmp) assert 'response for _testxxx' in str(tmp)
assert 'Port:12347' 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) tmp = cli.expect(pexpect.TIMEOUT, timeout=5)
assert 'response for testxxx' in str(tmp) assert 'response for testxxx' in str(tmp)
assert 'Port:12347' in str(tmp) assert 'Port:12347' in str(tmp)
finally: finally:
ocf.host_close_service() ocf.host_close_service()
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) 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] rx_nums = ocf.ot_ping(cli, str(host_ipv4_address), 5)[1]
assert rx_nums != 0 assert rx_nums != 0
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) time.sleep(3)
@ -457,9 +459,9 @@ def test_UDP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N
formBasicWiFiThreadNetwork(br, cli) formBasicWiFiThreadNetwork(br, cli)
try: try:
assert ocf.is_joined_wifi_network(br) assert ocf.is_joined_wifi_network(br)
br.write('bbr') ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5) br.expect('server16', timeout=5)
cli.write('udp open') ocf.execute_command(cli, 'udp open')
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
ocf.wait(cli, 3) ocf.wait(cli, 3)
host_ipv4_address = ocf.get_host_ipv4_address() 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 assert False
for num in range(0, 3): for num in range(0, 3):
command = 'udp send ' + host_ipv4_address + ' 5090 hello' + str(num) command = 'udp send ' + host_ipv4_address + ' 5090 hello' + str(num)
cli.write(command) ocf.execute_command(cli, command)
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
ocf.wait(cli, 0.5) ocf.wait(cli, 0.5)
while udp_mission.is_alive(): while udp_mission.is_alive():
time.sleep(1) time.sleep(1)
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) time.sleep(3)
assert b'hello' in myudp.udp_bytes 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) formBasicWiFiThreadNetwork(br, cli)
try: try:
assert ocf.is_joined_wifi_network(br) assert ocf.is_joined_wifi_network(br)
br.write('bbr') ocf.execute_command(br, 'bbr')
br.expect('server16', timeout=5) br.expect('server16', timeout=5)
cli.write('tcpsockclient open') ocf.execute_command(cli, 'tcpsockclient open')
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
ocf.wait(cli, 3) ocf.wait(cli, 3)
host_ipv4_address = ocf.get_host_ipv4_address() 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: if (time.time() - start_time) > 10:
assert False assert False
command = 'tcpsockclient connect ' + connect_address + ' 12345' command = 'tcpsockclient connect ' + connect_address + ' 12345'
cli.write(command) ocf.execute_command(cli, command)
cli.expect('Successfully connected', timeout=10) cli.expect('Successfully connected', timeout=10)
start_time = time.time() start_time = time.time()
while not mytcp.recv_flag: while not mytcp.recv_flag:
if (time.time() - start_time) > 10: if (time.time() - start_time) > 10:
assert False assert False
command = 'tcpsockclient send hello' command = 'tcpsockclient send hello'
cli.write(command) ocf.execute_command(cli, command)
cli.expect('Done', timeout=5) cli.expect('Done', timeout=5)
while tcp_mission.is_alive(): while tcp_mission.is_alive():
time.sleep(1) time.sleep(1)
finally: finally:
br.write('factoryreset') ocf.execute_command(br, 'factoryreset')
cli.write('factoryreset') ocf.execute_command(cli, 'factoryreset')
time.sleep(3) time.sleep(3)
assert b'hello' in mytcp.tcp_bytes assert b'hello' in mytcp.tcp_bytes