alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2025-10-30 20:51:41 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
cdcb4936b76b6421a1917d67b683fcdec560809b
esp-idf/examples/openthread/pytest_otbr.py

532 lines
19 KiB
Python
Raw Blame History

# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
# !/usr/bin/env python3
import os.path
import re
import subprocess
import threading
import time
from typing import Tuple
import ot_ci_function as ocf
import pexpect
import pytest
from pytest_embedded_idf.dut import IdfDut
# This file contains the test scripts for Thread:
# Case 1: Thread network formation and attaching
# A Thread Border Router forms a Thread network, a Thread device attaches to it, then test ping connection between them.
# Case 2: Bidirectional IPv6 connectivity
# Test IPv6 ping connection between Thread device and Linux Host (via Thread Border Router).
# Case 3: Multicast forwarding from Wi-Fi to Thread network
# Thread device joins the multicast group, then test group communication from Wi-Fi to Thread network.
# Case 4: Multicast forwarding from Thread to Wi-Fi network
# Linux Host joins the multicast group, test group communication from Thread to Wi-Fi network.
# Case 5: discover Serice published by Thread device
# Thread device publishes the service, Linux Host discovers the service on Wi-Fi network.
# Case 6: discover Serice published by W-Fi device
# Linux Host device publishes the service on Wi-Fi network, Thread device discovers the service.
# Case 7: ICMP communication via NAT64
# Thread device (IPV6) ping the host device (IPV4) via NAT64.
# Case 8: UDP communication via NAT64
# Thread device (IPV6) send udp message to the host device (IPV4) via NAT64.
# Case 9: TCP communication via NAT64
# Thread device (IPV6) send tcp message to the host device (IPV4) via NAT64.
@pytest.fixture(scope='module', name='Init_avahi')
def fixture_Init_avahi() -> bool:
print('Init Avahi')
ocf.start_avahi()
time.sleep(10)
return True
@pytest.fixture(name='Init_interface')
def fixture_Init_interface() -> bool:
print('Init interface')
ocf.init_interface_ipv6_address()
ocf.reset_host_interface()
time.sleep(30)
ocf.set_interface_sysctl_options()
return True
wifi_ssid = 'OTCITE'
wifi_psk = 'otcitest888'
# Case 1: Thread network formation and attaching
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'), # No need to rename beta_target as it is still called h2 in esptool
],
indirect=True,
)
def test_thread_connect(dut:Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, '0000')
flag = False
try:
cli_mleid_addr = ocf.get_mleid_addr(cli)
br_mleid_addr = ocf.get_mleid_addr(br)
rx_nums = ocf.ot_ping(cli, br_mleid_addr, 5)[1]
assert rx_nums != 0
rx_nums = ocf.ot_ping(br, cli_mleid_addr, 5)[1]
assert rx_nums != 0
flag = True
finally:
br.write('factoryreset')
cli.write('factoryreset')
time.sleep(3)
assert flag
# Case 2: Bidirectional IPv6 connectivity
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_Bidirectional_IPv6_connectivity(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
flag = False
try:
assert ocf.is_joined_wifi_network(br)
cli_global_unicast_addr = ocf.get_global_unicast_addr(cli, br)
print('cli_global_unicast_addr', cli_global_unicast_addr)
command = 'ping ' + str(cli_global_unicast_addr) + ' -c 10'
out_str = subprocess.getoutput(command)
print('ping result:\n', str(out_str))
role = re.findall(r' (\d+)%', str(out_str))[0]
assert role != '100'
interface_name = ocf.get_host_interface_name()
command = 'ifconfig ' + interface_name
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))
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
flag = True
finally:
br.write('factoryreset')
cli.write('factoryreset')
time.sleep(3)
assert flag
# Case 3: Multicast forwarding from Wi-Fi to Thread network
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_multicast_forwarding_A(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
flag = False
try:
assert ocf.is_joined_wifi_network(br)
br.write('bbr')
br.expect('server16', timeout=5)
assert ocf.thread_is_joined_group(cli)
interface_name = ocf.get_host_interface_name()
command = 'ping -I ' + str(interface_name) + ' -t 64 ff04::125 -c 10'
out_str = subprocess.getoutput(command)
print('ping result:\n', str(out_str))
role = re.findall(r' (\d+)%', str(out_str))[0]
assert role != '100'
flag = True
finally:
br.write('factoryreset')
cli.write('factoryreset')
time.sleep(3)
assert flag
# Case 4: Multicast forwarding from Thread to Wi-Fi network
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_multicast_forwarding_B(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
try:
assert ocf.is_joined_wifi_network(br)
br.write('bbr')
br.expect('server16', timeout=5)
cli.write('udp open')
cli.expect('Done', timeout=5)
ocf.wait(cli, 3)
myudp = ocf.udp_parameter('INET6', '::', 5090, 'ff04::125', False, 15.0, b'')
udp_mission = threading.Thread(target=ocf.create_host_udp_server, args=(myudp, ))
udp_mission.start()
start_time = time.time()
while not myudp.init_flag:
if (time.time() - start_time) > 10:
assert False
assert ocf.host_joined_group('ff04::125')
for num in range(0, 3):
command = 'udp send ff04::125 5090 hello' + str(num)
cli.write(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')
time.sleep(3)
assert b'hello' in myudp.udp_bytes
# Case 5: discover dervice published by Thread device
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_service_discovery_of_Thread_device(Init_interface:bool, Init_avahi:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
assert Init_avahi
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
ocf.wait(cli, 3)
flag = False
try:
assert ocf.is_joined_wifi_network(br)
command = 'avahi-browse -rt _testyyy._udp'
out_str = subprocess.getoutput(command)
print('avahi-browse:\n', str(out_str))
assert 'myTest' not in str(out_str)
hostname = 'myTest'
command = 'srp client host name ' + hostname
cli.write(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)
cli.expect('Done', timeout=5)
port = '12348'
command = 'srp client service add my-service _testyyy._udp ' + port
cli.write(command)
cli.expect('Done', timeout=5)
cli.write('srp client autostart enable')
cli.expect('Done', timeout=5)
ocf.wait(cli, 3)
command = 'avahi-browse -rt _testyyy._udp'
out_str = subprocess.getoutput(command)
print('avahi-browse:\n', str(out_str))
assert 'myTest' in str(out_str)
flag = True
finally:
br.write('factoryreset')
cli.write('factoryreset')
time.sleep(3)
assert flag
# Case 6: discover dervice published by Wi-Fi device
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_service_discovery_of_WiFi_device(Init_interface:bool, Init_avahi:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
assert Init_avahi
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
ocf.wait(cli, 3)
flag = False
try:
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)
cli.expect('Done', timeout=5)
ocf.wait(cli, 1)
command = 'dns resolve FA000123.default.service.arpa.'
ocf.clean_buffer(cli)
cli.write(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)
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)
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)
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.')
tmp = cli.expect(pexpect.TIMEOUT, timeout=5)
assert 'response for testxxx' in str(tmp)
assert 'Port:12347' in str(tmp)
flag = True
finally:
ocf.host_close_service()
br.write('factoryreset')
cli.write('factoryreset')
time.sleep(3)
assert flag
# Case 7: ICMP communication via NAT64
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_ICMP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
flag = False
try:
assert ocf.is_joined_wifi_network(br)
host_ipv4_address = ocf.get_host_ipv4_address()
print('host_ipv4_address: ', host_ipv4_address)
rx_nums = ocf.ot_ping(cli, str(host_ipv4_address), 5)[1]
assert rx_nums != 0
flag = True
finally:
br.write('factoryreset')
cli.write('factoryreset')
time.sleep(3)
assert flag
# Case 8: UDP communication via NAT64
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_UDP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
try:
assert ocf.is_joined_wifi_network(br)
br.write('bbr')
br.expect('server16', timeout=5)
cli.write('udp open')
cli.expect('Done', timeout=5)
ocf.wait(cli, 3)
host_ipv4_address = ocf.get_host_ipv4_address()
print('host_ipv4_address: ', host_ipv4_address)
myudp = ocf.udp_parameter('INET4', host_ipv4_address, 5090, '', False, 15.0, b'')
udp_mission = threading.Thread(target=ocf.create_host_udp_server, args=(myudp, ))
udp_mission.start()
start_time = time.time()
while not myudp.init_flag:
if (time.time() - start_time) > 10:
assert False
for num in range(0, 3):
command = 'udp send ' + host_ipv4_address + ' 5090 hello' + str(num)
cli.write(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')
time.sleep(3)
assert b'hello' in myudp.udp_bytes
# Case 9: TCP communication via NAT64
@pytest.mark.esp32s3
@pytest.mark.esp32h4
@pytest.mark.i154_multi_dut
@pytest.mark.flaky(reruns=1, reruns_delay=1)
@pytest.mark.parametrize(
'port, config, count, app_path, beta_target, target', [
('/dev/USB_RCP|/dev/USB_CLI|/dev/USB_BR', 'rcp|cli|br', 3,
f'{os.path.join(os.path.dirname(__file__), "ot_rcp")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_cli")}'
f'|{os.path.join(os.path.dirname(__file__), "ot_br")}',
'esp32h2beta2|esp32h2beta2|esp32s3', 'esp32h4|esp32h4|esp32s3'),
],
indirect=True,
)
def test_TCP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> None:
br = dut[2]
cli = dut[1]
assert Init_interface
dut[0].serial.stop_redirect_thread()
dataset = '-1'
ocf.config_network(br, cli, 'br', 'random', dataset, br, wifi_ssid, wifi_psk)
try:
assert ocf.is_joined_wifi_network(br)
br.write('bbr')
br.expect('server16', timeout=5)
cli.write('tcpsockclient open')
cli.expect('Done', timeout=5)
ocf.wait(cli, 3)
host_ipv4_address = ocf.get_host_ipv4_address()
connect_address = ocf.get_ipv6_from_ipv4(host_ipv4_address, br)
print('connect_address is: ', connect_address)
mytcp = ocf.tcp_parameter('INET4', host_ipv4_address, 12345, False, False, 15.0, b'')
tcp_mission = threading.Thread(target=ocf.create_host_tcp_server, args=(mytcp, ))
tcp_mission.start()
start_time = time.time()
while not mytcp.listen_flag:
if (time.time() - start_time) > 10:
assert False
command = 'tcpsockclient connect ' + connect_address + ' 12345'
cli.write(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)
cli.expect('Done', timeout=5)
while tcp_mission.is_alive():
time.sleep(1)
finally:
br.write('factoryreset')
cli.write('factoryreset')
time.sleep(3)
assert b'hello' in mytcp.tcp_bytes
Reference in New Issue View Git Blame Copy Permalink