Skip to content

Commit

Permalink
[Python] Fix src/python_testing tests
Browse files Browse the repository at this point in the history
  • Loading branch information
agners committed Jun 16, 2024
1 parent 4df0874 commit 440ca61
Show file tree
Hide file tree
Showing 4 changed files with 28 additions and 12 deletions.
13 changes: 13 additions & 0 deletions src/controller/python/chip/exceptions/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
"UnknownCommand",
]

from ..native import PyChipError


class ChipStackException(Exception):
pass
Expand All @@ -34,8 +36,19 @@ class ChipStackException(Exception):
class ChipStackError(ChipStackException):
def __init__(self, err, msg=None):
self.err = err
self.chip_error = None
self.msg = msg if msg else "Chip Stack Error %d" % err

@classmethod
def from_chip_error(cls, err: PyChipError):
exc = cls(err, err.sdk_code, str(err))
exc.chip_error = err
return exc

@property
def chip_error(self) -> PyChipError | None:
return self.chip_error

def __str__(self):
return self.msg

Expand Down
2 changes: 1 addition & 1 deletion src/controller/python/chip/native/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ def sdk_code(self) -> int:

def to_exception(self) -> typing.Union[None, chip.exceptions.ChipStackError]:
if not self.is_success:
return chip.exceptions.ChipStackError(self.code, str(self))
return chip.exceptions.ChipStackError.from_chip_error(self)

def __str__(self):
buf = ctypes.create_string_buffer(256)
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
21 changes: 12 additions & 9 deletions src/python_testing/TC_CGEN_2_4.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import chip.clusters as Clusters
import chip.clusters.enum
import chip.FabricAdmin
from chip.exception import ChipStackError
from chip import ChipDeviceCtrl
from chip.ChipDeviceCtrl import CommissioningParameters
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
Expand Down Expand Up @@ -60,13 +61,15 @@ 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')
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')
try:
self.th2.CommissionOnNetwork(
nodeId=self.dut_node_id, setupPinCode=params.setupPinCode,
filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=self.discriminator)
asserts.assert_false(True, 'Commissioning complete did not error as expected')
except ChipStackError as e:
logging.info('Commissioning complete done. Successful? {}, errorcode = {}'.format(False, e.chip_error.sdk_code))
asserts.assert_true(e.chip_error.sdk_part == expectedErrorPart, 'Unexpected error type returned from CommissioningComplete')
asserts.assert_true(e.chip_error.sdk_code == expectedErrCode, 'Unexpected error code returned from CommissioningComplete')
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
Expand Down Expand Up @@ -101,10 +104,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

0 comments on commit 440ca61

Please sign in to comment.