diff --git a/examples/openthread/ot_ci_function.py b/examples/openthread/ot_ci_function.py index d4f76c6864b..96075a1adce 100644 --- a/examples/openthread/ot_ci_function.py +++ b/examples/openthread/ot_ci_function.py @@ -809,9 +809,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: diff --git a/examples/openthread/pytest_otbr.py b/examples/openthread/pytest_otbr.py index 83d2a9afdb6..4121f30acce 100644 --- a/examples/openthread/pytest_otbr.py +++ b/examples/openthread/pytest_otbr.py @@ -94,18 +94,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' + 'ESPPORT3': 'esp32c6', } +ESPPORT1, ESPPORT2, ESPPORT3 = (os.getenv(name) for name in PORT_MAPPING) + +_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) @@ -129,7 +139,7 @@ def erase_flash_after_all_cases() -> Generator[None, None, None]: # Case 1: Thread network formation and attaching @pytest.mark.supported_targets -@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', @@ -230,7 +240,7 @@ def formBasicWiFiThreadNetwork(br:IdfDut, cli:IdfDut) -> None: # Case 2: Bidirectional IPv6 connectivity @pytest.mark.supported_targets -@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', @@ -427,7 +437,7 @@ def test_multicast_forwarding_B(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, # Case 5: discover dervice published by Thread device @pytest.mark.supported_targets -@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', @@ -496,8 +506,8 @@ def test_service_discovery_of_Thread_device(Init_interface:bool, Init_avahi:bool # Case 6: discover dervice published by Wi-Fi device @pytest.mark.supported_targets -@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', [ @@ -566,10 +576,11 @@ def test_service_discovery_of_WiFi_device(Init_interface:bool, Init_avahi:bool, 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() @@ -582,7 +593,7 @@ def test_service_discovery_of_WiFi_device(Init_interface:bool, Init_avahi:bool, # Case 7: ICMP communication via NAT64 @pytest.mark.supported_targets -@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', @@ -620,7 +631,7 @@ def test_ICMP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> # Case 8: UDP communication via NAT64 @pytest.mark.supported_targets -@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', @@ -676,7 +687,7 @@ def test_UDP_NAT64(Init_interface:bool, dut: Tuple[IdfDut, IdfDut, IdfDut]) -> N # Case 9: TCP communication via NAT64 @pytest.mark.supported_targets -@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', @@ -773,7 +784,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) @@ -847,7 +858,7 @@ def test_basic_startup(dut: Tuple[IdfDut, IdfDut]) -> None: # Case 13: Meshcop discovery of Border Router @pytest.mark.supported_targets -@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', @@ -1035,7 +1046,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) @@ -1047,7 +1058,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) diff --git a/tools/ci/idf_pytest/constants.py b/tools/ci/idf_pytest/constants.py index e6f01a03a33..6540a4e4a18 100644 --- a/tools/ci/idf_pytest/constants.py +++ b/tools/ci/idf_pytest/constants.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: 2023-2025 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. @@ -97,8 +97,11 @@ ENV_MARKERS = { 'httpbin': 'runner for tests that need to access the httpbin service', # multi-dut markers '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.',