diff --git a/src/controller/python/chip/CertificateAuthority.py b/src/controller/python/chip/CertificateAuthority.py index 4c6ff194554c35..d9bd00e86ba261 100644 --- a/src/controller/python/chip/CertificateAuthority.py +++ b/src/controller/python/chip/CertificateAuthority.py @@ -315,6 +315,21 @@ def NewCertificateAuthority(self, caIndex: Optional[int] = None, maximizeCertCha return ca + def create_new_controller(self, vendor_Id: int = 0xFFF1, fabric_Id: int = 1, new_Fabric: bool = True, ca_List: int = 0, node_Id: int = 2): + if new_Fabric: + """ + Create new fabric and controller for commissioning + """ + new_cert_auth = self.NewCertificateAuthority() + new_fabric_admin = new_cert_auth.NewFabricAdmin(vendorId=vendor_Id, fabricId=fabric_Id) + return new_fabric_admin.NewController(nodeId=node_Id) + + else: + """ + Create new controller on an already established fabric + """ + return self.activeCaList[ca_List].adminList[0].NewController(nodeId=node_Id) + def Shutdown(self): ''' Shuts down all active CertificateAuthority instances tracked by this manager, before shutting itself down. diff --git a/src/python_testing/TC_CreateNewController.py b/src/python_testing/TC_CreateNewController.py new file mode 100644 index 00000000000000..28dbd0d0370676 --- /dev/null +++ b/src/python_testing/TC_CreateNewController.py @@ -0,0 +1,128 @@ +# +# Copyright (c) 2023 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. +# + +# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments +# for details about the block below. +# +# === BEGIN CI TEST ARGUMENTS === +# test-runner-runs: +# run1: +# app: ${ALL_CLUSTERS_APP} +# app-args: --discriminator 1234 --passcode 20202021 --KVS kvs1 +# script-args: > +# --storage-path admin_storage.json +# --discriminator 1234 +# --passcode 20202021 +# --endpoint 1 +# --commissioning-method on-network +# factory-reset: true +# quiet: true +# === END CI TEST ARGUMENTS === + +import asyncio + +from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main +from mobly import asserts + + +class TC_CreateNewController(MatterBaseTest): + # Using get_code and a modified version of setup_class_helper functions from chip.testing.basic_composition module + def get_code(self, dev_ctrl): + created_codes = [] + for idx, discriminator in enumerate(self.matter_test_config.discriminators): + created_codes.append(dev_ctrl.CreateManualCode(discriminator, self.matter_test_config.setup_passcodes[idx])) + + setup_codes = self.matter_test_config.qr_code_content + self.matter_test_config.manual_code + created_codes + if not setup_codes: + return None + asserts.assert_equal(len(setup_codes), 1, + "Require exactly one of either --qr-code, --manual-code or (--discriminator and --passcode).") + return setup_codes[0] + + async def setup_class_helper(self, allow_pase: bool = True): + dev_ctrl = self.default_controller + self.problems = [] + + node_id = self.dut_node_id + + task_list = [] + if allow_pase and self.get_code(dev_ctrl): + setup_code = self.get_code(dev_ctrl) + pase_future = dev_ctrl.EstablishPASESession(setup_code, self.dut_node_id) + task_list.append(asyncio.create_task(pase_future)) + + case_future = dev_ctrl.GetConnectedDevice(nodeid=node_id, allowPASE=False) + task_list.append(asyncio.create_task(case_future)) + + for task in task_list: + asyncio.ensure_future(task) + + done, pending = await asyncio.wait(task_list, return_when=asyncio.FIRST_COMPLETED) + + for task in pending: + try: + task.cancel() + await task + except asyncio.CancelledError: + pass + + wildcard_read = (await dev_ctrl.Read(node_id, [()])) + + # ======= State kept for use by all tests ======= + # All endpoints in "full object" indexing format + self.endpoints = wildcard_read.attributes + + def steps_TC_CreateNewController(self) -> list[TestStep]: + return [ + TestStep(1, "Commissioning, already done", is_commissioning=True), + TestStep(2, "Checking create new controller on new fabric"), + TestStep(3, "Remove new fabric from TH to verify it can be removed after creation"), + TestStep(4, "Checking create new controller on existing fabric"), + TestStep(5, "Remove fabric from TH to verify it can be removed after creation") + ] + + def TC_TestAttrAvail(self) -> list[str]: + return ["RVCOPSTATE.S"] + + @async_test_body + async def setup_class(self): + super().setup_class() + await self.setup_class_helper() + + # ======= START OF ACTUAL TESTS ======= + @async_test_body + async def test_TC_CreateNewController(self): + self.step(1) + self.endpoint = self.get_endpoint() + asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.") + self.th1 = self.default_controller + + self.step(2) + self.th2 = self.certificate_authority_manager.create_new_controller(fabric_Id=self.th1.fabricId + 1, ca_List=1, node_Id=2) + + self.step(3) + self.th2.Shutdown() + + self.step(4) + self.th3 = self.certificate_authority_manager.create_new_controller(ca_List=0, new_Fabric=False, node_Id=2) + + self.step(5) + self.th3.Shutdown() + + +if __name__ == "__main__": + default_matter_test_main()