Skip to content

Commit

Permalink
[Python] Adjust tests to use new commissioning error handling
Browse files Browse the repository at this point in the history
  • Loading branch information
agners committed Jun 18, 2024
1 parent a90ae1a commit 2e404b4
Show file tree
Hide file tree
Showing 9 changed files with 83 additions and 57 deletions.
6 changes: 3 additions & 3 deletions src/controller/python/chip/yaml/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)


Expand Down
28 changes: 19 additions & 9 deletions src/controller/python/test/test_scripts/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -311,17 +315,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.")
Expand Down Expand Up @@ -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

Expand Down
4 changes: 2 additions & 2 deletions src/python_testing/TC_ACE_1_5.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
16 changes: 9 additions & 7 deletions src/python_testing/TC_CGEN_2_4.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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)
Expand Down
3 changes: 1 addition & 2 deletions src/python_testing/TC_DA_1_5.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
3 changes: 1 addition & 2 deletions src/python_testing/TC_TIMESYNC_2_13.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
3 changes: 1 addition & 2 deletions src/python_testing/TestCommissioningTimeSync.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
72 changes: 46 additions & 26 deletions src/python_testing/matter_testing_support.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
5 changes: 1 addition & 4 deletions src/test_driver/openiotsdk/integration-tests/common/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down

0 comments on commit 2e404b4

Please sign in to comment.