Add TC_MCORE_FS_1_3.py test implementation

src/python_testing/TC_MCORE_FS_1_3.py

@@ -0,0 +1,161 @@
+#
+#    Copyright (c) 2024 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.
+#
+
+# This test requires a TH_SERVER application that returns UnsupportedAttribute when reading UniqueID from BasicInformation Cluster. Please specify with --string-arg th_server_app_path:<path_to_app>
+
+import logging
+import os
+import random
+import signal
+import subprocess
+import time
+import uuid
+
+import chip.clusters as Clusters
+from chip import ChipDeviceCtrl
+from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches
+from mobly import asserts
+
+
+class TC_MCORE_FS_1_3(MatterBaseTest):
+    @async_test_body
+    async def setup_class(self):
+        super().setup_class()
+        self.device_for_th_eco_kvs = None
+        self.device_for_th_eco_port = 5543
+        self.app_process_for_th_eco = None
+
+        self.device_for_dut_eco_kvs = None
+        self.device_for_dut_eco_port = 5544
+        self.app_process_for_dut_eco = None
+
+    def teardown_class(self):
+        if self.app_process_for_dut_eco is not None:
+            logging.warning("Stopping app with SIGTERM")
+            self.app_process_for_dut_eco.send_signal(signal.SIGTERM.value)
+            self.app_process_for_dut_eco.wait()
+        if self.app_process_for_th_eco is not None:
+            logging.warning("Stopping app with SIGTERM")
+            self.app_process_for_th_eco.send_signal(signal.SIGTERM.value)
+            self.app_process_for_th_eco.wait()
+
+        os.remove(self.device_for_dut_eco_kvs)
+        if self.device_for_th_eco_kvs is not None:
+            os.remove(self.device_for_th_eco_kvs)
+        super().teardown_class()
+
+    async def create_device_for_dut_ecosystem(self):
+        # TODO: confirm whether we can open processes like this on the TH
+        app = self.matter_test_config.user_params.get("th_server_app_path", None)
+        if not app:
+            asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
+
+        self.device_for_dut_eco_kvs = f'kvs_{str(uuid.uuid4())}'
+        # This discriminator is fixed so we can rely on pregenerated QR code
+        discriminator = 3840
+        passcode = 20202021
+        app_args = f'--secured-device-port {self.device_for_dut_eco_port} --discriminator {discriminator} --passcode {passcode} --KVS {self.device_for_dut_eco_kvs}'
+        cmd = f'{app} {app_args}'
+        # TODO: Determine if we want these logs cooked or pushed to somewhere else
+        logging.info("Starting TH device for DUT ecosystem")
+        self.app_process_for_dut_eco = subprocess.Popen(cmd, bufsize=0, shell=True)
+        logging.info("Started TH device for DUT ecosystem")
+        time.sleep(3)
+        return discriminator, passcode
+
+    async def create_and_commission_device_for_th_ecosystem(self):
+        # TODO: confirm whether we can open processes like this on the TH
+        app = self.matter_test_config.user_params.get("th_server_app_path", None)
+
+        self.device_for_th_eco_kvs = f'kvs_{str(uuid.uuid4())}'
+        discriminator = random.randint(0, 4095)
+        passcode = 20202021
+        app_args = f'--secured-device-port {self.device_for_th_eco_port} --discriminator {discriminator} --passcode {passcode} --KVS {self.device_for_th_eco_kvs}'
+        cmd = f'{app} {app_args}'
+        # TODO: Determine if we want these logs cooked or pushed to somewhere else
+        logging.info("Starting TH device for TH ecosystem")
+        self.app_process_for_th_eco = subprocess.Popen(cmd, bufsize=0, shell=True)
+        logging.info("Started TH device for TH ecosystem")
+        time.sleep(3)
+
+        logging.info("Commissioning from separate fabric")
+        # Create a second controller on a new fabric to communicate to the server
+        new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
+        new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
+        paa_path = str(self.matter_test_config.paa_trust_store_path)
+        self.TH_server_controller = new_fabric_admin.NewController(nodeId=112233, paaTrustStorePath=paa_path)
+        self.server_nodeid = 1111
+        await self.TH_server_controller.CommissionOnNetwork(nodeId=self.server_nodeid, setupPinCode=passcode, filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
+        logging.info("Commissioning TH device for TH ecosystem")
+
+    def steps_TC_MCORE_FS_1_3(self) -> list[TestStep]:
+        steps = [TestStep(1, "DUT_FSA commissions TH_SED_DUT to DUT_FSAs fabric and generates a UniqueID", is_commissioning=True),
+                 TestStep(2, "TH_FSA commissions TH_SED_TH onto TH_FSAs fabric and generates a UniqueID."),
+                 TestStep(3, "Follow manufacturer provided instructions to enable DUT_FSA to synchronize TH_SED_TH onto DUT_FSAs fabric."),
+                 TestStep(4, "DUT_FSA synchronizes TH_SED_TH onto DUT_FSAs fabric and copies the UniqueID presented by TH_FSAs Bridged Device Basic Information Cluster.")]
+        return steps
+
+    @async_test_body
+    async def test_TC_MCORE_FS_1_3(self):
+        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
+        self.print_step(0, "Commissioning DUT to TH, already done")
+        self.step(1)
+        # These steps are not explicitly in step 1, but they help identify the dynamically added endpoint in step 1.
+        root_node_endpoint = 0
+        root_part_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor, attribute=Clusters.Descriptor.Attributes.PartsList, endpoint=root_node_endpoint)
+        set_of_endpoints_before_adding_device = set(root_part_list)
+
+        discriminator, passcode = await self.create_device_for_dut_ecosystem()
+        # To prevent need for runtime QR code generation test makes some assumptions
+        # about the app that it valiudates here.
+        asserts.assert_equal(discriminator, 3840, "Test implementation assumption incorrect for discriminator")
+        asserts.assert_equal(passcode, 3840, "Test implementation assumption incorrect for passcode")
+        setup_qrcode = "MT:Y.K90-Q000KA0648G00"
+        self.wait_for_user_input(
+            prompt_msg=f"Using the DUT vendor's provided interface, commission the device using the following parameters:\n"
+            f"- discriminator: {discriminator}\n"
+            f"- passcode: {passcode}\n"
+            f"- setupPinCode: {setup_qrcode}\n"
+            f"If using FabricSync Admin, you may type:\n"
+            f">>> pairing onnetwork 111 {setup_qrcode}")
+
+        root_part_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor, attribute=Clusters.Descriptor.Attributes.PartsList, endpoint=root_node_endpoint)
+        set_of_endpoints_after_adding_device = set(root_part_list)
+
+        asserts.assert_true(set_of_endpoints_after_adding_device.issuperset(
+            set_of_endpoints_before_adding_device), "Expected only new endpoints to be added")
+        unique_endpoints_set = set_of_endpoints_after_adding_device - set_of_endpoints_before_adding_device
+        asserts.assert_equal(len(unique_endpoints_set), 1, "Expected only one new endpoint")
+        newly_added_endpoint = list(unique_endpoints_set)[0]
+
+        th_sed_dut_unique_id = await self.read_single_attribute_check_success(cluster=Clusters.BridgedDeviceBasicInformation, attribute=Clusters.BridgedDeviceBasicInformation.Attributes.UniqueID, endpoint=newly_added_endpoint)
+        asserts.assert_true(type_matches(th_sed_dut_unique_id, str), "UniqueID should be a string")
+        asserts.assert_true(th_sed_dut_unique_id, "UniqueID should not be an empty string")
+
+        self.step(2)
+        await self.create_and_commission_device_for_th_ecosystem()
+        # TODO(https://github.com/CHIP-Specifications/chip-test-plans/issues/4375) During setup we need to create the TH_FSA device
+        # where we would commission device created in create_and_commission_device_for_th_ecosystem to be commissioned into TH_FSA.
+
+        # TODO(https://github.com/CHIP-Specifications/chip-test-plans/issues/4375) Because we cannot create a TH_FSA and there is
+        # no way to mock it the following 2 test steps are skipped for now.
+        self.skip_step(3)
+        self.skip_step(4)
+
+
+if __name__ == "__main__":
+    default_matter_test_main()