ci: seperate openthread test cases

This commit is contained in:
yiwenxiu
2026-06-23 11:38:19 +08:00
parent 1874d5ba0d
commit aff69acef4
3 changed files with 61 additions and 27 deletions

View File

@@ -814,9 +814,25 @@ def execute_command(dut: IdfDut, command: str, prefix: str = 'ot ') -> None:
def get_output_string(dut: IdfDut, command: str, wait_time: int) -> str:
execute_command(dut, command)
tmp = dut.expect(pexpect.TIMEOUT, timeout=wait_time)
output = dut.expect(pexpect.TIMEOUT, timeout=wait_time)
clean_buffer(dut)
return str(tmp)
if isinstance(output, bytes):
return output.decode('utf-8', errors='replace')
return str(output)
def escape_ot_cli_arg(value: str) -> str:
return value.replace('\\', '\\\\').replace(' ', '\\ ')
def parse_dns_browse_instance(browse_output: str, port: str = '12347') -> str:
normalized = re.sub(r'\r\n?', '\n', browse_output)
match = re.search(
r'DNS browse response for [^\n]+\n+([^\n]+)\n+\s+Port:' + re.escape(port),
normalized,
)
assert match, f'no service instance with Port:{port} in browse output: {browse_output}'
return match.group(1).strip()
def wait_for_host_network(host: str = '8.8.8.8', retries: int = 6, interval: int = 10) -> None:

View File

@@ -3,7 +3,7 @@
# !/usr/bin/env python3
import copy
import logging
import os.path
import os
import random
import re
import secrets
@@ -77,14 +77,28 @@ def fixture_Init_interface() -> bool:
return True
default_br_ot_para = ocf.thread_parameter('leader', '', '12', '7766554433221100', True)
default_br_wifi_para = ocf.wifi_parameter('OTCITE', 'otcitest888', 10)
default_cli_ot_para = ocf.thread_parameter('router', '', '', '', False)
ESPPORT1 = os.getenv('ESPPORT1')
ESPPORT2 = os.getenv('ESPPORT2')
ESPPORT3 = os.getenv('ESPPORT3')
PORT_MAPPING = {
'ESPPORT1': 'esp32h2',
'ESPPORT2': 'esp32s3',
'ESPPORT3': 'esp32c6',
}
ESPPORT1, ESPPORT2, ESPPORT3 = (os.getenv(name) for name in PORT_MAPPING)
PORT_MAPPING = {'ESPPORT1': 'esp32h2', 'ESPPORT2': 'esp32s3', 'ESPPORT3': 'esp32c6'}
_RUNNER_CHANNEL = os.getenv('CHANNEL', '12')
_RUNNER_NETWORKKEY = os.getenv('NETWORKKEY', '99112233445566778899aabbccddeeff')
_BR_EXTADDR = '7766554433221100'
def _leader_ot_para(bbr: bool = True) -> ocf.thread_parameter:
para = ocf.thread_parameter('leader', '', _RUNNER_CHANNEL, _BR_EXTADDR, bbr)
if _RUNNER_NETWORKKEY:
para.setnetworkkey(_RUNNER_NETWORKKEY)
return para
default_br_ot_para = _leader_ot_para()
default_br_wifi_para = ocf.wifi_parameter(os.getenv('RUNNER_AP_SSID'), os.getenv('RUNNER_AP_PASSWORD'), 10)
default_cli_ot_para = ocf.thread_parameter('router', '', '', '', False)
@pytest.fixture(scope='module', autouse=True)
@@ -107,7 +121,7 @@ def erase_flash_after_all_cases() -> Generator[None, None, None]:
# Case 1: Thread network formation and attaching
@pytest.mark.openthread_br
@pytest.mark.openthread_br_connect
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
@@ -207,7 +221,7 @@ def formBasicWiFiThreadNetwork(br: IdfDut, cli: IdfDut) -> None:
# Case 2: Bidirectional IPv6 connectivity
@pytest.mark.openthread_br
@pytest.mark.openthread_br_connect
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
@@ -401,7 +415,7 @@ def test_multicast_forwarding_B(Init_interface: bool, dut: Tuple[IdfDut, IdfDut,
# Case 5: discover dervice published by Thread device
@pytest.mark.openthread_br
@pytest.mark.openthread_br_service
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
@@ -471,8 +485,8 @@ def test_service_discovery_of_Thread_device(
# Case 6: discover dervice published by Wi-Fi device
@pytest.mark.openthread_br
@pytest.mark.flaky(reruns=3, reruns_delay=5)
@pytest.mark.openthread_br_service
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
[
@@ -543,10 +557,11 @@ def test_service_discovery_of_WiFi_device(
tmp = ocf.get_output_string(cli, command, 10)
assert 'response for _testxxx' in str(tmp)
assert 'Port:12347' in str(tmp)
instance_label = ocf.parse_dns_browse_instance(str(tmp))
command = 'dns service testxxx _testxxx._udp.default.service.arpa.'
command = 'dns service ' + ocf.escape_ot_cli_arg(instance_label) + ' _testxxx._udp.default.service.arpa.'
tmp = ocf.get_output_string(cli, command, 10)
assert 'response for testxxx' in str(tmp)
assert f'response for {instance_label}' in str(tmp)
assert 'Port:12347' in str(tmp)
finally:
ocf.host_close_service()
@@ -558,7 +573,7 @@ def test_service_discovery_of_WiFi_device(
# Case 7: ICMP communication via NAT64
@pytest.mark.openthread_br
@pytest.mark.openthread_br_nat64
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
@@ -595,7 +610,7 @@ def test_ICMP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) ->
# Case 8: UDP communication via NAT64
@pytest.mark.openthread_br
@pytest.mark.openthread_br_nat64
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
@@ -650,7 +665,7 @@ def test_UDP_NAT64(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) ->
# Case 9: TCP communication via NAT64
@pytest.mark.openthread_br
@pytest.mark.openthread_br_nat64
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
@@ -745,7 +760,7 @@ def test_ot_sleepy_device(dut: Tuple[IdfDut, IdfDut]) -> None:
try:
ocf.init_thread(leader)
time.sleep(3)
leader_para = ocf.thread_parameter('leader', '', '12', '7766554433221100', False)
leader_para = _leader_ot_para(bbr=False)
ocf.SetThreadNetworkPara(leader, leader_para)
ocf.StartThreadNetwork(leader, leader_para)
ocf.wait(leader, 5)
@@ -855,7 +870,7 @@ def test_NAT64_DNS(Init_interface: bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) ->
# Case 13: Meshcop discovery of Border Router
@pytest.mark.openthread_br
@pytest.mark.openthread_br_service
@pytest.mark.flaky(reruns=1, reruns_delay=5)
@pytest.mark.parametrize(
'config, count, app_path, target, port',
@@ -1076,7 +1091,7 @@ def test_ot_ssed_device(dut: Tuple[IdfDut, IdfDut]) -> None:
ssed_device.expect('32k XTAL in use', timeout=10)
ocf.init_thread(leader)
time.sleep(3)
leader_para = ocf.thread_parameter('leader', '', '12', '7766554433221100', False)
leader_para = _leader_ot_para(bbr=False)
ocf.SetThreadNetworkPara(leader, leader_para)
ocf.StartThreadNetwork(leader, leader_para)
ocf.wait(leader, 5)
@@ -1088,7 +1103,7 @@ def test_ot_ssed_device(dut: Tuple[IdfDut, IdfDut]) -> None:
ssed_device.expect('Done', timeout=5)
ocf.execute_command(ssed_device, 'csl period 3000000')
ssed_device.expect('Done', timeout=5)
ocf.execute_command(ssed_device, 'csl channel 12')
ocf.execute_command(ssed_device, f'csl channel {_RUNNER_CHANNEL}')
ssed_device.expect('Done', timeout=5)
ocf.execute_command(ssed_device, 'ifconfig up')
ssed_device.expect('Done', timeout=5)

View File

@@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: 2023-2024 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2023-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""
Pytest Related Constants. Don't import third-party packages here.
@@ -108,8 +108,11 @@ ENV_MARKERS = {
# multi-dut markers
'multi_dut_modbus_rs485': 'a pair of runners connected by RS485 bus',
'ieee802154': 'ieee802154 related tests should run on ieee802154 runners.',
'openthread_br': 'tests should be used for openthread border router.',
'openthread_bbr': 'tests should be used for openthread border router linked to Internet.',
'openthread_br': 'openthread BR basic and multicast cases.',
'openthread_br_connect': 'openthread BR attach and connectivity cases.',
'openthread_br_service': 'openthread BR service discovery cases.',
'openthread_br_nat64': 'openthread BR NAT64 cases without internet.',
'openthread_bbr': 'openthread BR cases with internet access.',
'openthread_sleep': 'tests should be used for openthread sleepy device.',
'zigbee_multi_dut': 'zigbee runner which have multiple duts.',
'two_duts': 'tests should be run on runners which has two wifi duts connected.',