| # | |
| # Copyright (c) 2026 Project CHIP Authors | |
| # All rights reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| # | |
| from mobly import asserts | |
| from matter import ChipDeviceCtrl, discovery | |
| from matter.testing.matter_testing import MatterBaseTest, SetupParameters | |
| THREAD_RENDEZVOUS_INFORMATION = 1 << 5 | |
| def _first_setup_payload_info(test: MatterBaseTest): | |
| setup_payloads = test.get_setup_payload_info() | |
| asserts.assert_true(setup_payloads, | |
| "Thread MeshCoP provisional tests require --qr-code, --manual-code, or --discriminator/--passcode.") | |
| return setup_payloads[0] | |
| def _long_discriminator_from_config_or_payload(test: MatterBaseTest) -> int: | |
| if test.matter_test_config.discriminators: | |
| return test.matter_test_config.discriminators[0] | |
| setup_payload = _first_setup_payload_info(test) | |
| asserts.assert_equal(setup_payload.filter_type, discovery.FilterType.LONG_DISCRIMINATOR, | |
| "Long-discriminator MeshCoP test steps require --qr-code or --discriminator/--passcode.") | |
| return setup_payload.filter_value | |
| def get_setup_code(test: MatterBaseTest, use_short_discriminator: bool) -> str: | |
| if not use_short_discriminator and test.matter_test_config.qr_code_content: | |
| return test.matter_test_config.qr_code_content[0] | |
| if use_short_discriminator and test.matter_test_config.manual_code: | |
| return test.matter_test_config.manual_code[0] | |
| if test.matter_test_config.setup_passcodes and test.matter_test_config.discriminators: | |
| setup_params = SetupParameters(passcode=test.matter_test_config.setup_passcodes[0], | |
| discriminator=test.matter_test_config.discriminators[0], | |
| capabilities=THREAD_RENDEZVOUS_INFORMATION) | |
| return setup_params.manual_code if use_short_discriminator else setup_params.qr_code | |
| setup_payload = _first_setup_payload_info(test) | |
| if not use_short_discriminator: | |
| asserts.assert_is_not_none(setup_payload.setup_code, | |
| "Long-discriminator MeshCoP test steps require --qr-code or --discriminator/--passcode.") | |
| return setup_payload.setup_code | |
| if setup_payload.filter_type == discovery.FilterType.SHORT_DISCRIMINATOR: | |
| manual_code_discriminator = setup_payload.filter_value << 8 | |
| else: | |
| manual_code_discriminator = setup_payload.filter_value | |
| return test.default_controller.CreateManualCode(manual_code_discriminator, setup_payload.passcode) | |
| def discriminator_from_config(test: MatterBaseTest, use_short_discriminator: bool) -> int: | |
| long_discriminator = _long_discriminator_from_config_or_payload(test) | |
| return (long_discriminator >> 8) & 0x0F if use_short_discriminator else long_discriminator | |
| def verify_meshcop_config(test: MatterBaseTest) -> None: | |
| asserts.assert_is_not_none(test.matter_test_config.thread_operational_dataset, | |
| "Thread MeshCoP provisional tests require --thread-dataset-hex.") | |
| asserts.assert_is_not_none(test.matter_test_config.thread_ba_host, | |
| "Thread MeshCoP provisional tests require --thread-ba-host.") | |
| asserts.assert_is_not_none(test.matter_test_config.thread_ba_port, | |
| "Thread MeshCoP provisional tests require --thread-ba-port.") | |
| async def establish_pase_over_thread_meshcop(test: MatterBaseTest, setup_code: str) -> dict: | |
| verify_meshcop_config(test) | |
| commissioner: ChipDeviceCtrl.ChipDeviceController = test.default_controller | |
| commissioner.SetThreadOperationalDataset(test.matter_test_config.thread_operational_dataset) | |
| await commissioner.EstablishPASESessionThreadMeshcop(baAddr=test.matter_test_config.thread_ba_host, | |
| setupCode=setup_code, | |
| nodeId=test.matter_test_config.dut_node_ids[0], | |
| baPort=test.matter_test_config.thread_ba_port) | |
| return commissioner.GetLastThreadMeshcopDiscoveryDiagnostic() | |
| def assert_common_diagnostic_fields(diagnostic: dict, expected_discriminator: int, use_short_discriminator: bool) -> None: | |
| asserts.assert_true(diagnostic["valid"], "Thread MeshCoP discovery diagnostic was not captured.") | |
| asserts.assert_equal(diagnostic["requested_discriminator_type"], | |
| "short" if use_short_discriminator else "long") | |
| asserts.assert_equal(diagnostic["requested_discriminator"], expected_discriminator) | |
| asserts.assert_true(diagnostic["steering_data_hex"], "Steering data must be present.") | |
| asserts.assert_greater(diagnostic["joiner_id"], 0, "Joiner/source identifier must be captured.") | |
| asserts.assert_greater(diagnostic["joiner_udp_port"], 0, "Joiner UDP port must be captured.") | |
| announcement = diagnostic["dns_announcement"] | |
| asserts.assert_true(announcement["thread_meshcop"], "DNS announcement must be from Thread MeshCoP discovery.") | |
| asserts.assert_greater(announcement["service_port"], 0, "DNS announcement must advertise a Matter joiner UDP port.") | |
| asserts.assert_true(0 <= announcement["long_discriminator"] <= 4095, | |
| "DNS announcement must contain a valid commissionable node long discriminator.") | |
| if use_short_discriminator: | |
| actual_short_discriminator = (announcement["long_discriminator"] >> 8) & 0x0F | |
| asserts.assert_equal(actual_short_discriminator, expected_discriminator, | |
| "DNS announcement long discriminator must match the requested short discriminator.") | |
| else: | |
| asserts.assert_equal(announcement["long_discriminator"], expected_discriminator, | |
| "DNS announcement long discriminator must match the requested long discriminator.") |