[Python] Adjust tests to use new commissioning error handling

Author: agners

src/controller/python/chip/yaml/runner.py

@@ -664,10 +664,10 @@ async def run_action(self, dev_ctrl: ChipDeviceController) -> _ActionResult:
         if self._command == 'GetCommissionerNodeId':
             return _ActionResult(status=_ActionStatus.SUCCESS, response=_GetCommissionerNodeIdResult(dev_ctrl.nodeId))
 
-        resp = dev_ctrl.CommissionWithCode(self._setup_payload, self._node_id)
-        if resp:
+        try:
+            dev_ctrl.CommissionWithCode(self._setup_payload, self._node_id)
             return _ActionResult(status=_ActionStatus.SUCCESS, response=None)
-        else:
+        except ChipStackError:
             return _ActionResult(status=_ActionStatus.ERROR, response=None)
 
 

src/controller/python/test/test_scripts/base.py

@@ -41,6 +41,7 @@
 from chip import ChipDeviceCtrl
 from chip.ChipStack import ChipStack
 from chip.crypto import p256keypair
+from chip.exceptions import ChipStackException
 from chip.utils import CommissioningBuildingBlocks
 from cirque_restart_remote_device import restartRemoteDevice
 from ecdsa import NIST256p
@@ -268,7 +269,9 @@ def TestPaseOnly(self, ip: str, setuppin: int, nodeid: int, devCtrl=None):
     def TestCommissionOnly(self, nodeid: int):
         self.logger.info(
             "Commissioning device with id {}".format(nodeid))
-        if not self.devCtrl.Commission(nodeid):
+        try:
+            self.devCtrl.Commission(nodeid)
+        except ChipStackException:
             self.logger.info(
                 "Failed to commission device with id {}".format(str(nodeid)))
             return False
@@ -311,17 +314,21 @@ def TestCommissionFailureOnReport(self, nodeid: int, failAfter: int):
 
     def TestCommissioning(self, ip: str, setuppin: int, nodeid: int):
         self.logger.info("Commissioning device {}".format(ip))
-        if not self.devCtrl.CommissionIP(ip, setuppin, nodeid):
-            self.logger.info(
+        try:
+            self.devCtrl.CommissionIP(ip, setuppin, nodeid)
+        except ChipStackException:
+            self.logger.exception(
                 "Failed to finish commissioning device {}".format(ip))
             return False
         self.logger.info("Commissioning finished.")
         return True
 
     def TestCommissioningWithSetupPayload(self, setupPayload: str, nodeid: int, discoveryType: int = 2):
         self.logger.info("Commissioning device with setup payload {}".format(setupPayload))
-        if not self.devCtrl.CommissionWithCode(setupPayload, nodeid, chip.discovery.DiscoveryType(discoveryType)):
-            self.logger.info(
+        try:
+            self.devCtrl.CommissionWithCode(setupPayload, nodeid, chip.discovery.DiscoveryType(discoveryType))
+        except ChipStackException:
+            self.logger.exception(
                 "Failed to finish commissioning device {}".format(setupPayload))
             return False
         self.logger.info("Commissioning finished.")
@@ -783,8 +790,10 @@ async def TestMultiFabric(self, ip: str, setuppin: int, nodeid: int):
         self.devCtrl2 = self.fabricAdmin2.NewController(
             self.controllerNodeId, self.paaTrustStorePath)
 
-        if not self.devCtrl2.CommissionIP(ip, setuppin, nodeid):
-            self.logger.info(
+        try:
+            self.devCtrl2.CommissionIP(ip, setuppin, nodeid)
+        except ChipStackException:
+            self.logger.exception(
                 "Failed to finish key exchange with device {}".format(ip))
             return False
 

src/python_testing/TC_ACE_1_5.py

@@ -54,10 +54,10 @@ async def test_TC_ACE_1_5(self):
         params = self.openCommissioningWindow(self.th1, self.dut_node_id)
         self.print_step(2, "TH1 opens the commissioning window on the DUT")
 
-        errcode = self.th2.CommissionOnNetwork(
+        self.th2.CommissionOnNetwork(
             nodeId=self.dut_node_id, setupPinCode=params.commissioningParameters.setupPinCode,
             filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=params.randomDiscriminator)
-        logging.info('Commissioning complete done. Successful? {}, errorcode = {}'.format(errcode.is_success, errcode))
+        logging.info('Commissioning complete done. Successful.')
         self.print_step(3, "TH2 commissions DUT using admin node ID N2")
 
         self.print_step(4, "TH2 reads its fabric index from the Operational Credentials cluster CurrentFabricIndex attribute")

src/python_testing/TC_CGEN_2_4.py

@@ -25,6 +25,7 @@
 import chip.FabricAdmin
 from chip import ChipDeviceCtrl
 from chip.ChipDeviceCtrl import CommissioningParameters
+from chip.exception import ChipStackError
 from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
 from mobly import asserts
 
@@ -60,11 +61,12 @@ async def CommissionToStageSendCompleteAndCleanup(
         # This will run the commissioning up to the point where stage x is run and the
         # response is sent before the test commissioner simulates a failure
         self.th2.SetTestCommissionerPrematureCompleteAfter(stage)
-        errcode = self.th2.CommissionOnNetwork(
-            nodeId=self.dut_node_id, setupPinCode=params.setupPinCode,
-            filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.discriminator)
-        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')
+        raises_assert = asserts.assert_raises(ChipStackError)
+        with raises_assert:
+            self.th2.CommissionOnNetwork(
+                nodeId=self.dut_node_id, setupPinCode=params.setupPinCode,
+                filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.discriminator)
+        errcode = raises_assert.exception.chip_error
         asserts.assert_true(errcode.sdk_part == expectedErrorPart, 'Unexpected error type returned from CommissioningComplete')
         asserts.assert_true(errcode.sdk_code == expectedErrCode, 'Unexpected error code returned from CommissioningComplete')
         revokeCmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
@@ -101,10 +103,10 @@ async def test_TC_CGEN_2_4(self):
 
         logging.info('Step 16 - TH2 fully commissions the DUT')
         self.th2.ResetTestCommissioner()
-        errcode = self.th2.CommissionOnNetwork(
+        self.th2.CommissionOnNetwork(
             nodeId=self.dut_node_id, setupPinCode=params.setupPinCode,
             filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.discriminator)
-        logging.info('Commissioning complete done. Successful? {}, errorcode = {}'.format(errcode.is_success, errcode))
+        logging.info('Commissioning complete done. Successful.')
 
         logging.info('Step 17 - TH1 sends an arm failsafe')
         cmd = Clusters.GeneralCommissioning.Commands.ArmFailSafe(expiryLengthSeconds=900, breadcrumb=0)

src/python_testing/TC_DA_1_5.py

@@ -170,10 +170,9 @@ async def test_TC_DA_1_5(self):
         new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
         TH2 = new_fabric_admin.NewController(nodeId=112233)
 
-        errcode = TH2.CommissionOnNetwork(
+        TH2.CommissionOnNetwork(
             nodeId=self.dut_node_id, setupPinCode=params.setupPinCode,
             filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=1234)
-        asserts.assert_true(errcode.is_success, 'Commissioning on TH2 did not complete successfully')
 
         self.print_step(15, "Read NOCs list for TH1")
         temp = await self.read_single_attribute_check_success(

src/python_testing/TC_TIMESYNC_2_13.py

@@ -53,10 +53,9 @@ async def test_TC_TIMESYNC_2_13(self):
         new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
         TH2 = new_fabric_admin.NewController(nodeId=112233)
 
-        errcode = TH2.CommissionOnNetwork(
+        TH2.CommissionOnNetwork(
             nodeId=self.dut_node_id, setupPinCode=params.setupPinCode,
             filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=1234)
-        asserts.assert_true(errcode.is_success, 'Commissioning on TH2 did not complete successfully')
 
         self.print_step(3, "TH2 reads the current fabric")
         th2_fabric_idx = await self.read_single_attribute_check_success(

src/python_testing/TestCommissioningTimeSync.py

@@ -58,10 +58,9 @@ async def teardown_test(self):
     async def commission_and_base_checks(self):
         params = self.default_controller.OpenCommissioningWindow(
             nodeid=self.dut_node_id, timeout=600, iteration=10000, discriminator=1234, option=1)
-        errcode = self.commissioner.CommissionOnNetwork(
+        self.commissioner.CommissionOnNetwork(
             nodeId=self.dut_node_id, setupPinCode=params.setupPinCode,
             filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=1234)
-        asserts.assert_true(errcode.is_success, 'Commissioning did not complete successfully')
         self.commissioned = True
 
         # Check the feature map - if we have a time cluster, we want UTC time to be set

src/python_testing/matter_testing_support.py

@@ -1542,35 +1542,55 @@ def _commission_device(self, i) -> bool:
             info.filter_value = conf.discriminators[i]
 
         if conf.commissioning_method == "on-network":
-            return dev_ctrl.CommissionOnNetwork(
-                nodeId=conf.dut_node_ids[i],
-                setupPinCode=info.passcode,
-                filterType=info.filter_type,
-                filter=info.filter_value
-            )
+            try:
+                dev_ctrl.CommissionOnNetwork(
+                    nodeId=conf.dut_node_ids[i],
+                    setupPinCode=info.passcode,
+                    filterType=info.filter_type,
+                    filter=info.filter_value
+                )
+                return True
+            except ChipStackError as e:
+                logging.error("Commissioning failed: %s" % e)
+                return False
         elif conf.commissioning_method == "ble-wifi":
-            return dev_ctrl.CommissionWiFi(
-                info.filter_value,
-                info.passcode,
-                conf.dut_node_ids[i],
-                conf.wifi_ssid,
-                conf.wifi_passphrase,
-                isShortDiscriminator=(info.filter_type == DiscoveryFilterType.SHORT_DISCRIMINATOR)
-            )
+            try:
+                dev_ctrl.CommissionWiFi(
+                    info.filter_value,
+                    info.passcode,
+                    conf.dut_node_ids[i],
+                    conf.wifi_ssid,
+                    conf.wifi_passphrase,
+                    isShortDiscriminator=(info.filter_type == DiscoveryFilterType.SHORT_DISCRIMINATOR)
+                )
+                return True
+            except ChipStackError as e:
+                logging.error("Commissioning failed: %s" % e)
+                return False
         elif conf.commissioning_method == "ble-thread":
-            return dev_ctrl.CommissionThread(
-                info.filter_value,
-                info.passcode,
-                conf.dut_node_ids[i],
-                conf.thread_operational_dataset,
-                isShortDiscriminator=(info.filter_type == DiscoveryFilterType.SHORT_DISCRIMINATOR)
-            )
+            try:
+                dev_ctrl.CommissionThread(
+                    info.filter_value,
+                    info.passcode,
+                    conf.dut_node_ids[i],
+                    conf.thread_operational_dataset,
+                    isShortDiscriminator=(info.filter_type == DiscoveryFilterType.SHORT_DISCRIMINATOR)
+                )
+                return True
+            except ChipStackError as e:
+                logging.error("Commissioning failed: %s" % e)
+                return False
         elif conf.commissioning_method == "on-network-ip":
-            logging.warning("==== USING A DIRECT IP COMMISSIONING METHOD NOT SUPPORTED IN THE LONG TERM ====")
-            return dev_ctrl.CommissionIP(
-                ipaddr=conf.commissionee_ip_address_just_for_testing,
-                setupPinCode=info.passcode, nodeid=conf.dut_node_ids[i]
-            )
+            try:
+                logging.warning("==== USING A DIRECT IP COMMISSIONING METHOD NOT SUPPORTED IN THE LONG TERM ====")
+                dev_ctrl.CommissionIP(
+                    ipaddr=conf.commissionee_ip_address_just_for_testing,
+                    setupPinCode=info.passcode, nodeid=conf.dut_node_ids[i]
+                )
+                return True
+            except ChipStackError as e:
+                logging.error("Commissioning failed: %s" % e)
+                return False
         else:
             raise ValueError("Invalid commissioning method %s!" % conf.commissioning_method)
 

src/test_driver/openiotsdk/integration-tests/common/utils.py

@@ -92,14 +92,11 @@ def connect_device(devCtrl, setupPayload, commissionableDevice, nodeId=None):
 
     pincode = int(setupPayload.attributes['SetUpPINCode'])
     try:
-        res = devCtrl.CommissionOnNetwork(
+        devCtrl.CommissionOnNetwork(
             nodeId, pincode, filterType=discovery.FilterType.INSTANCE_NAME, filter=commissionableDevice.instanceName)
     except exceptions.ChipStackError as ex:
         log.error("Commission discovered device failed {}".format(str(ex)))
         return None
-    if not res:
-        log.info("Commission discovered device failed: %r" % res)
-        return None
     return nodeId