Allow inheritance of MatterBaseTest (project-chip#39397)

cecille · restyled-commits · web-flow · commit b501f3b0477b · 2025-07-21T21:03:20.000Z
* Allow inheritance of MatterBaseTest

Change CADMIN tests to inherit from a base class rather than
using an additional support class.

This just makes the code a bit neater. It also means we can
add cleanup steps in the base class, which is REALLY handy.

Next PR will add a cleanup step for CADMIN.

* remove some unused imports - linter

* fix more lints

* Restyled by isort

* Address review comments

* Restyled by isort

* review comment CADMINSupport -&gt; CADMINBaseTest

* Restyled by isort

---------

Co-authored-by: Restyled.io &lt;commits@restyled.io&gt;
diff --git a/src/python_testing/TC_CADMIN_1_11.py b/src/python_testing/TC_CADMIN_1_11.py
@@ -42,16 +42,12 @@
 from chip.ChipDeviceCtrl import CommissioningParameters
 from chip.exceptions import ChipStackError
 from chip.native import PyChipError
-from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.testing.matter_testing import TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
-from support_modules.cadmin_support import CADMINSupport
+from support_modules.cadmin_support import CADMINBaseTest
 
 
-class TC_CADMIN_1_11(MatterBaseTest):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.support = CADMINSupport(self)
-
+class TC_CADMIN_1_11(CADMINBaseTest):
     async def OpenCommissioningWindow(self, th, expectedErrCode) -> CommissioningParameters:
         if expectedErrCode is None:
             params = await th.OpenCommissioningWindow(
@@ -209,7 +205,7 @@ async def test_TC_CADMIN_1_11(self):
 
         # Read CurrentFabricIndex attribute from the Operational Credentials cluster
         self.step(10)
-        th2_idx = await self.support.read_currentfabricindex(self.th2)
+        th2_idx = await self.read_currentfabricindex(self.th2)
 
         self.step(11)
         removeFabricCmd = Clusters.OperationalCredentials.Commands.RemoveFabric(th2_idx)
diff --git a/src/python_testing/TC_CADMIN_1_19.py b/src/python_testing/TC_CADMIN_1_19.py
@@ -36,16 +36,12 @@
 from chip import ChipDeviceCtrl
 from chip.exceptions import ChipStackError
 from chip.testing.event_attribute_reporting import AttributeSubscriptionHandler
-from chip.testing.matter_testing import AttributeValue, MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.testing.matter_testing import AttributeValue, TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
-from support_modules.cadmin_support import CADMINSupport
+from support_modules.cadmin_support import CADMINBaseTest
 
 
-class TC_CADMIN_1_19(MatterBaseTest):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.support = CADMINSupport(self)
-
+class TC_CADMIN_1_19(CADMINBaseTest):
     def steps_TC_CADMIN_1_19(self) -> list[TestStep]:
         return [
             TestStep(1, "Commissioning, already done", is_commissioning=True),
@@ -96,7 +92,7 @@ async def test_TC_CADMIN_1_19(self):
         self.max_window_duration = duration.maxCumulativeFailsafeSeconds
 
         self.step(3)
-        fabrics = await self.support.get_fabrics(th=self.th1, fabric_filtered=False)
+        fabrics = await self.get_fabrics(th=self.th1, fabric_filtered=False)
         initial_number_of_fabrics = len(fabrics)
 
         self.step(4)
diff --git a/src/python_testing/TC_CADMIN_1_22.py b/src/python_testing/TC_CADMIN_1_22.py
@@ -37,16 +37,12 @@
 import chip.clusters as Clusters
 from chip.ChipDeviceCtrl import CommissioningParameters
 from chip.exceptions import ChipStackError
-from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.testing.matter_testing import TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
-from support_modules.cadmin_support import CADMINSupport
+from support_modules.cadmin_support import CADMINBaseTest
 
 
-class TC_CADMIN_1_22_24(MatterBaseTest):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.support = CADMINSupport(self)
-
+class TC_CADMIN_1_22_24(CADMINBaseTest):
     async def OpenCommissioningWindow(self) -> CommissioningParameters:
         try:
             params = await self.th1.OpenCommissioningWindow(
@@ -96,12 +92,12 @@ async def test_TC_CADMIN_1_22(self):
         revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
         await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=revokeCmd, timedRequestTimeoutMs=6000)
         # The failsafe cleanup is scheduled after the command completes, so give it a bit of time to do that
-        window_status = await self.support.get_window_status(th=self.th1)
+        window_status = await self.get_window_status(th=self.th1)
         asserts.assert_equal(window_status, Clusters.AdministratorCommissioning.Enums.CommissioningWindowStatusEnum.kWindowNotOpen,
                              "Commissioning window is expected to be closed, but was found to be open")
 
         self.step(4)
-        window_status = await self.support.get_window_status(th=self.th1)
+        window_status = await self.get_window_status(th=self.th1)
         asserts.assert_equal(window_status, Clusters.AdministratorCommissioning.Enums.CommissioningWindowStatusEnum.kWindowNotOpen,
                              "Commissioning window is expected to be closed, but was found to be open")
 
@@ -117,7 +113,7 @@ async def test_TC_CADMIN_1_22(self):
                              "Expected to error as we provided failure value for opening commissioning window")
 
         self.step(6)
-        window_status2 = await self.support.get_window_status(th=self.th1)
+        window_status2 = await self.get_window_status(th=self.th1)
         asserts.assert_equal(window_status2, Clusters.AdministratorCommissioning.Enums.CommissioningWindowStatusEnum.kWindowNotOpen,
                              "Commissioning window is expected to be closed, but was found to be open")
 
@@ -129,7 +125,7 @@ async def test_TC_CADMIN_1_22(self):
         revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
         await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=revokeCmd, timedRequestTimeoutMs=6000)
         # The failsafe cleanup is scheduled after the command completes, so give it a bit of time to do that
-        window_status3 = await self.support.get_window_status(th=self.th1)
+        window_status3 = await self.get_window_status(th=self.th1)
         asserts.assert_equal(window_status3, Clusters.AdministratorCommissioning.Enums.CommissioningWindowStatusEnum.kWindowNotOpen,
                              "Commissioning window is expected to be closed, but was found to be open")
 
@@ -146,7 +142,7 @@ async def test_TC_CADMIN_1_22(self):
                              "Expected to error as we provided failure value for opening commissioning window")
 
         self.step(10)
-        window_status4 = await self.support.get_window_status(th=self.th1)
+        window_status4 = await self.get_window_status(th=self.th1)
         asserts.assert_equal(window_status4, Clusters.AdministratorCommissioning.Enums.CommissioningWindowStatusEnum.kWindowNotOpen,
                              "Commissioning window is expected to be closed, but was found to be open")
 
diff --git a/src/python_testing/TC_CADMIN_1_3_4.py b/src/python_testing/TC_CADMIN_1_3_4.py
@@ -53,21 +53,16 @@
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.exceptions import ChipStackError
-from chip.testing.matter_testing import (MatterBaseTest, TestStep, default_matter_test_main, has_cluster, has_feature,
-                                         run_if_endpoint_matches)
+from chip.testing.matter_testing import TestStep, default_matter_test_main, has_cluster, has_feature, run_if_endpoint_matches
 from chip.tlv import TLVReader
 from mobly import asserts
-from support_modules.cadmin_support import CADMINSupport
+from support_modules.cadmin_support import CADMINBaseTest
 
 opcreds = Clusters.OperationalCredentials
 nonce = random.randbytes(32)
 
 
-class TC_CADMIN(MatterBaseTest):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.support = CADMINSupport(self)
-
+class TC_CADMIN(CADMINBaseTest):
     async def combined_commission_val_steps(self, commission_type: str):
         """
         Combined test function for commissioning tests.
@@ -108,13 +103,13 @@ async def combined_commission_val_steps(self, commission_type: str):
 
         self.step("3b")
         if commission_type == "ECM":
-            service = await self.support.wait_for_correct_cm_value(
+            service = await self.wait_for_correct_cm_value(
                 expected_cm_value=2,
                 expected_discriminator=1234
             )
             logging.info(f"Successfully found service with CM={service.txt_record.get('CM')}, D={service.txt_record.get('D')}")
         elif commission_type == "BCM":
-            service = await self.support.wait_for_correct_cm_value(
+            service = await self.wait_for_correct_cm_value(
                 expected_cm_value=1,
                 expected_discriminator=setupPayloadInfo[0].filter_value
             )
@@ -123,8 +118,8 @@ async def combined_commission_val_steps(self, commission_type: str):
         self.step("3c")
         BI_cluster = Clusters.BasicInformation
         self.nl_attribute = BI_cluster.Attributes.NodeLabel
-        await self.support.write_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
-        await self.support.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
+        await self.write_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
+        await self.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
 
         self.step(4)
         # Establishing TH2
@@ -154,7 +149,7 @@ async def combined_commission_val_steps(self, commission_type: str):
 
         self.step(5)
         # TH_CR1 reads the Fabrics attribute
-        th1_fabric_info = await self.support.get_fabrics(th=self.th1)
+        th1_fabric_info = await self.get_fabrics(th=self.th1)
         th1_cam_rcac = TLVReader(base64.b64decode(
             self.certificate_authority_manager.activeCaList[0]._persistentStorage._jsonData["sdk-config"]["f/1/r"])).get()["Any"][9]
         if th1_fabric_info[0].rootPublicKey != th1_cam_rcac:
@@ -164,7 +159,7 @@ async def combined_commission_val_steps(self, commission_type: str):
 
         self.step(6)
         # TH_CR2 reads the Fabrics attribute
-        th2_fabric_info = await self.support.get_fabrics(th=self.th2)
+        th2_fabric_info = await self.get_fabrics(th=self.th2)
         if th2_fabric_info[0].rootPublicKey != th2_rcac_decoded:
             asserts.fail("Public keys from fabric and certs for TH2 are not the same.")
         if th2_fabric_info[0].nodeID != self.dut_node_id:
@@ -173,15 +168,15 @@ async def combined_commission_val_steps(self, commission_type: str):
         if commission_type == "ECM":
             self.step(7)
             # TH_CR1 writes and reads the Basic Information Cluster’s NodeLabel mandatory attribute of DUT_CE
-            await self.support.write_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
-            await self.support.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
+            await self.write_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
+            await self.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th1, attr_val=self.nl_attribute)
 
             self.step(8)
             # TH_CR2 writes and reads the Basic Information Cluster’s NodeLabel mandatory attribute of DUT_CE
-            val = await self.support.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th2, attr_val=self.nl_attribute)
+            val = await self.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th2, attr_val=self.nl_attribute)
             self.print_step("basic information cluster node label attr value", val)
-            await self.support.write_nl_attr(dut_node_id=self.dut_node_id, th=self.th2, attr_val=self.nl_attribute)
-            await self.support.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th2, attr_val=self.nl_attribute)
+            await self.write_nl_attr(dut_node_id=self.dut_node_id, th=self.th2, attr_val=self.nl_attribute)
+            await self.read_nl_attr(dut_node_id=self.dut_node_id, th=self.th2, attr_val=self.nl_attribute)
 
             self.step(9)
             # TH_CR2 opens a commissioning window on DUT_CE for 180 seconds using ECM
diff --git a/src/python_testing/TC_CADMIN_1_9.py b/src/python_testing/TC_CADMIN_1_9.py
@@ -32,57 +32,45 @@
 # === END CI TEST ARGUMENTS ===
 
 import logging
-import random
+from copy import deepcopy
 from time import sleep
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
-from chip.ChipDeviceCtrl import CommissioningParameters
 from chip.exceptions import ChipStackError
 from chip.native import PyChipError
-from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.testing.matter_testing import CustomCommissioningParameters, TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
-from support_modules.cadmin_support import CADMINSupport
-
-
-class TC_CADMIN_1_9(MatterBaseTest):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.support = CADMINSupport(self)
-
-    async def OpenCommissioningWindow(self) -> CommissioningParameters:
-        try:
-            cluster = Clusters.GeneralCommissioning
-            attribute = cluster.Attributes.BasicCommissioningInfo
-            duration = await self.read_single_attribute_check_success(endpoint=0, cluster=cluster, attribute=attribute)
-            return await self.support.open_commissioning_window(
-                th=self.th1,
-                timeout=duration.maxCumulativeFailsafeSeconds,
-                node_id=self.dut_node_id,
-                discriminator=self.discriminator
-            )
-        except Exception as e:
-            logging.exception('Error running OpenCommissioningWindow %s', e)
-            asserts.assert_true(False, 'Failed to open commissioning window')
+from support_modules.cadmin_support import CADMINBaseTest
+
+
+class TC_CADMIN_1_9(CADMINBaseTest):
+    async def OpenCommissioningWindowForMaxTime(self) -> CustomCommissioningParameters:
+        cluster = Clusters.GeneralCommissioning
+        attribute = cluster.Attributes.BasicCommissioningInfo
+        duration = await self.read_single_attribute_check_success(endpoint=0, cluster=cluster, attribute=attribute)
+        return await self.open_commissioning_window(dev_ctrl=self.th1, node_id=self.dut_node_id, timeout=duration.maxCumulativeFailsafeSeconds)
 
-    async def CommissionOnNetwork(self, setup_code: int):
+    async def CommissionOnNetwork(self, params: CustomCommissioningParameters):
         ctx = asserts.assert_raises(ChipStackError)
         with ctx:
             await self.th2.CommissionOnNetwork(
                 nodeId=self.dut_node_id,
-                setupPinCode=setup_code,
+                setupPinCode=params.commissioningParameters.setupPinCode,
                 filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR,
-                filter=self.discriminator
+                filter=params.randomDiscriminator
             )
         errcode = PyChipError.from_code(ctx.exception.err)
         return errcode
 
-    async def CommissionAttempt(self, setupPinCode: int, expectedErrCode: int):
+    async def CommissionAttempt(self, params: CustomCommissioningParameters, expectedErrCode: int):
         if expectedErrCode == 3:
             for cycle in range(20):
                 logging.info("-----------------Current Iteration {}-------------------------".format(cycle+1))
-                setup_code = self.support.generate_unique_random_value(setupPinCode)  # Updated to use support module
-                errcode = await self.CommissionOnNetwork(setup_code)
+                new_params = deepcopy(params)
+                new_params.commissioningParameters.setupPinCode = self.generate_unique_random_value(
+                    params.commissioningParameters.setupPinCode)
+                errcode = await self.CommissionOnNetwork(new_params)
                 logging.info('Commissioning complete done. Successful? {}, errorcode = {}, cycle={}'.format(
                     errcode.is_success, errcode, (cycle+1)))
                 asserts.assert_false(errcode.is_success, 'Commissioning complete did not error as expected')
@@ -91,7 +79,7 @@ async def CommissionAttempt(self, setupPinCode: int, expectedErrCode: int):
 
         elif expectedErrCode == 50:
             logging.info("-----------------Attempting connection expecting timeout-------------------------")
-            errcode = await self.CommissionOnNetwork(setupPinCode)
+            errcode = await self.CommissionOnNetwork(params)
             logging.info('Commissioning complete done. Successful? {}, errorcode = {}'.format(errcode.is_success, errcode))
             asserts.assert_false(errcode.is_success, 'Commissioning complete did not error as expected')
             # TODO: Adding try or except clause here as the errcode code be either 50 for timeout or 3 for incorrect state at this time
@@ -118,23 +106,21 @@ async def test_TC_CADMIN_1_9(self):
 
         # Establishing TH1 and TH2
         self.th1 = self.default_controller
-        self.discriminator = random.randint(0, 4095)
         th2_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
         th2_fabric_admin = th2_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=self.th1.fabricId + 1)
         self.th2 = th2_fabric_admin.NewController(nodeId=2, useTestCommissioner=True)
 
         self.step(2)
-        params = await self.OpenCommissioningWindow()
-        setupPinCode = params.setupPinCode
+        params = await self.OpenCommissioningWindowForMaxTime()
 
         self.step(3)
-        await self.CommissionAttempt(setupPinCode, expectedErrCode=0x03)
+        await self.CommissionAttempt(params, expectedErrCode=0x03)
 
         self.step(4)
-        await self.CommissionAttempt(setupPinCode, expectedErrCode=0x32)
+        await self.CommissionAttempt(params, expectedErrCode=0x32)
 
         self.step(5)
-        params = await self.OpenCommissioningWindow()
+        params = await self.OpenCommissioningWindowForMaxTime()
 
         self.step(6)
         revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
diff --git a/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py b/src/python_testing/matter_testing_infrastructure/chip/testing/matter_testing.py
@@ -1141,15 +1141,24 @@ def _find_test_class():
     Raises:
       SystemExit: Raised if the number of test classes is not exactly one.
     """
-    subclasses = utils.find_subclasses_in_module([MatterBaseTest], sys.modules['__main__'])
-    subclasses = [c for c in subclasses if c.__name__ != "MatterBaseTest"]
-    if len(subclasses) != 1:
+    def get_subclasses(cls: Any):
+        subclasses = utils.find_subclasses_in_module([cls], sys.modules['__main__'])
+        subclasses = [c for c in subclasses if c.__name__ != cls.__name__]
+        return subclasses
+
+    def has_subclasses(cls: Any):
+        return get_subclasses(cls) != []
+
+    subclasses_matter_test_base = get_subclasses(MatterBaseTest)
+    leaf_subclasses = [s for s in subclasses_matter_test_base if not has_subclasses(s)]
+
+    if len(leaf_subclasses) != 1:
         print(
             'Exactly one subclass of `MatterBaseTest` should be in the main file. Found %s.' %
-            str([subclass.__name__ for subclass in subclasses]))
+            str([subclass.__name__ for subclass in leaf_subclasses]))
         sys.exit(1)
 
-    return subclasses[0]
+    return leaf_subclasses[0]
 
 
 def int_decimal_or_hex(s: str) -> int:
diff --git a/src/python_testing/support_modules/cadmin_support.py b/src/python_testing/support_modules/cadmin_support.py
