diff --git a/src/controller/python/chip/yaml/runner.py b/src/controller/python/chip/yaml/runner.py index fb7a71bd457715..312940f34bffd5 100644 --- a/src/controller/python/chip/yaml/runner.py +++ b/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) diff --git a/src/controller/python/test/test_scripts/base.py b/src/controller/python/test/test_scripts/base.py index 3f9f76d9101874..1861627c0cbff4 100644 --- a/src/controller/python/test/test_scripts/base.py +++ b/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 @@ -256,8 +257,9 @@ def TestPaseOnly(self, ip: str, setuppin: int, nodeid: int, devCtrl=None): devCtrl = self.devCtrl self.logger.info( "Attempting to establish PASE session with device id: {} addr: {}".format(str(nodeid), ip)) - if devCtrl.EstablishPASESessionIP( - ip, setuppin, nodeid) is not None: + try: + devCtrl.EstablishPASESessionIP(ip, setuppin, nodeid) + except ChipStackException: self.logger.info( "Failed to establish PASE session with device id: {} addr: {}".format(str(nodeid), ip)) return False @@ -268,7 +270,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,8 +315,10 @@ 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.") @@ -320,8 +326,10 @@ def TestCommissioning(self, ip: str, setuppin: int, nodeid: int): 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 +791,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 diff --git a/src/python_testing/TC_ACE_1_5.py b/src/python_testing/TC_ACE_1_5.py index 93c3fce5c58c11..bab70260fc6508 100644 --- a/src/python_testing/TC_ACE_1_5.py +++ b/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") diff --git a/src/python_testing/TC_CGEN_2_4.py b/src/python_testing/TC_CGEN_2_4.py index 7bb2e6d21d3d73..7f2d4f675c6e16 100644 --- a/src/python_testing/TC_CGEN_2_4.py +++ b/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.exceptions 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) diff --git a/src/python_testing/TC_DA_1_5.py b/src/python_testing/TC_DA_1_5.py index 567d7577604288..17c6e3c16da0fb 100644 --- a/src/python_testing/TC_DA_1_5.py +++ b/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( diff --git a/src/python_testing/TC_TIMESYNC_2_13.py b/src/python_testing/TC_TIMESYNC_2_13.py index ceabb23e5df5c7..fa43bbd00cffcb 100644 --- a/src/python_testing/TC_TIMESYNC_2_13.py +++ b/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( diff --git a/src/python_testing/TestCommissioningTimeSync.py b/src/python_testing/TestCommissioningTimeSync.py index 509aabfc6aa500..0fca7063fcc666 100644 --- a/src/python_testing/TestCommissioningTimeSync.py +++ b/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 diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index e0acefd1c61782..e6ea387a5d4d68 100644 --- a/src/python_testing/matter_testing_support.py +++ b/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) diff --git a/src/test_driver/openiotsdk/integration-tests/common/utils.py b/src/test_driver/openiotsdk/integration-tests/common/utils.py index 1865cf6274f6ed..da2dcff787de7a 100644 --- a/src/test_driver/openiotsdk/integration-tests/common/utils.py +++ b/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