From e09abaec3147a9a1b6024d5a1853ec6dd135c8e3 Mon Sep 17 00:00:00 2001 From: Yufeng Wang Date: Thu, 14 Nov 2024 19:12:07 -0800 Subject: [PATCH] Do not use reverse commissioning to pair test server (#36490) --- src/python_testing/TC_MCORE_FS_1_3.py | 95 ++++++++++----------------- 1 file changed, 35 insertions(+), 60 deletions(-) diff --git a/src/python_testing/TC_MCORE_FS_1_3.py b/src/python_testing/TC_MCORE_FS_1_3.py index 2ac1556199a0c7..4b732a1b3a8995 100644 --- a/src/python_testing/TC_MCORE_FS_1_3.py +++ b/src/python_testing/TC_MCORE_FS_1_3.py @@ -34,7 +34,7 @@ # --commissioning-method on-network # --discriminator 1234 # --passcode 20202021 -# --string-arg th_server_no_uid_app_path:${LIGHTING_APP_NO_UNIQUE_ID} +# --string-arg th_server_no_uid_app_path:${LIGHTING_APP_NO_UNIQUE_ID} dut_fsa_stdin_pipe:dut-fsa-stdin # --trace-to json:${TRACE_TEST_JSON}.json # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto # factory-reset: true @@ -81,6 +81,13 @@ def setup_class(self): self.storage = tempfile.TemporaryDirectory(prefix=self.__class__.__name__) logging.info("Temporary storage directory: %s", self.storage.name) + if self.is_pics_sdk_ci_only: + # Get the named pipe path for the DUT_FSA app input from the user params. + dut_fsa_stdin_pipe = self.user_params.get("dut_fsa_stdin_pipe") + if not dut_fsa_stdin_pipe: + asserts.fail("CI setup requires --string-arg dut_fsa_stdin_pipe:") + self.dut_fsa_stdin = open(dut_fsa_stdin_pipe, "w") + self.th_server_port = 5544 self.th_server_discriminator = random.randint(0, 4095) self.th_server_passcode = 20202021 @@ -111,60 +118,6 @@ def steps_TC_MCORE_FS_1_3(self) -> list[TestStep]: "TH verifies a value is visible for the UniqueID from the DUT_FSA's Bridged Device Basic Information Cluster."), ] - async def commission_via_commissioner_control(self, controller_node_id: int, device_node_id: int, endpoint_id: int): - """Commission device_node_id to controller_node_id using CommissionerControl cluster.""" - - request_id = random.randint(0, 0xFFFFFFFFFFFFFFFF) - - vendor_id = await self.read_single_attribute_check_success( - node_id=device_node_id, - cluster=Clusters.BasicInformation, - attribute=Clusters.BasicInformation.Attributes.VendorID, - ) - - product_id = await self.read_single_attribute_check_success( - node_id=device_node_id, - cluster=Clusters.BasicInformation, - attribute=Clusters.BasicInformation.Attributes.ProductID, - ) - - await self.send_single_cmd( - node_id=controller_node_id, - endpoint=endpoint_id, - cmd=Clusters.CommissionerControl.Commands.RequestCommissioningApproval( - requestID=request_id, - vendorID=vendor_id, - productID=product_id, - ), - ) - - if not self.is_pics_sdk_ci_only: - self.wait_for_user_input("Approve Commissioning Approval Request on DUT using manufacturer specified mechanism") - - resp = await self.send_single_cmd( - node_id=controller_node_id, - endpoint=endpoint_id, - cmd=Clusters.CommissionerControl.Commands.CommissionNode( - requestID=request_id, - responseTimeoutSeconds=30, - ), - ) - - asserts.assert_equal(type(resp), Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow, - "Incorrect response type") - - await self.send_single_cmd( - node_id=device_node_id, - cmd=Clusters.AdministratorCommissioning.Commands.OpenCommissioningWindow( - commissioningTimeout=3*60, - PAKEPasscodeVerifier=resp.PAKEPasscodeVerifier, - discriminator=resp.discriminator, - iterations=resp.iterations, - salt=resp.salt, - ), - timedRequestTimeoutMs=5000, - ) - @async_test_body async def test_TC_MCORE_FS_1_3(self): @@ -218,13 +171,35 @@ async def test_TC_MCORE_FS_1_3(self): asserts.assert_not_equal(aggregator_endpoint, 0, "Invalid aggregator endpoint. Cannot proceed with commissioning.") - await self.commission_via_commissioner_control( - controller_node_id=self.dut_node_id, - device_node_id=th_server_th_node_id, - endpoint_id=aggregator_endpoint) + # Open commissioning window on TH_SERVER_NO_UID. + discriminator = random.randint(0, 4095) + params = await self.default_controller.OpenCommissioningWindow( + nodeid=th_server_th_node_id, + option=self.default_controller.CommissioningWindowPasscode.kTokenWithRandomPin, + discriminator=discriminator, + iteration=10000, + timeout=600) + + # Commissioning TH_SERVER_NO_UID to DUT_FSA fabric. + if not self.is_pics_sdk_ci_only: + self.wait_for_user_input( + f"Commission TH_SERVER_NO_UID on DUT using manufacturer specified mechanism.\n" + f"Use the following parameters:\n" + f"- discriminator: {discriminator}\n" + f"- setupPinCode: {params.setupPinCode}\n" + f"- setupQRCode: {params.setupQRCode}\n" + f"- setupManualCode: {params.setupManualCode}\n" + f"If using FabricSync Admin, you may type:\n" + f">>> pairing onnetwork {params.setupPinCode}") + else: + self.dut_fsa_stdin.write( + f"pairing onnetwork 10 {params.setupPinCode}\n") + self.dut_fsa_stdin.flush() + # Wait for the commissioning to complete. + await asyncio.sleep(5) # Wait for the device to appear on the DUT_FSA_BRIDGE. - await asyncio.sleep(2 if self.is_pics_sdk_ci_only else 30) + await asyncio.sleep(1) # Get the list of endpoints on the DUT_FSA_BRIDGE after adding the TH_SERVER_NO_UID. dut_fsa_bridge_endpoints_new = set(await self.read_single_attribute_check_success(