From 32949129628f169725dc7c55424aa33acf8e2946 Mon Sep 17 00:00:00 2001 From: C Freeman Date: Thu, 15 Feb 2024 20:11:22 -0500 Subject: [PATCH] TC-IDM-10.1: Support write-only attributes (#32049) * TC-IDM-10.1: Support write-only attributes Write only attributes are not returned in the wildcard, but will return the UNSUPPORTED_READ error if we attempt to read them in a concrete path. We can detect their presence by probing for this error via a read. * remove random comment * Establish PASE session from code * fix node id * Catch KeyError on missing global This would have already been validated in the last step, so catch and release this error and move on. * Fix global attribute names * Fix attribute string * Restyled by clang-format * Update src/python_testing/TC_DeviceBasicComposition.py * Restyled by autopep8 * Make comment more verbose. --------- Co-authored-by: Restyled.io --- .../ChipDeviceController-ScriptBinding.cpp | 9 +++ src/controller/python/chip/ChipDeviceCtrl.py | 12 ++++ .../python/chip/clusters/ClusterObjects.py | 6 +- .../TC_DeviceBasicComposition.py | 64 ++++++++++++++++--- .../basic_composition_support.py | 22 ++----- src/python_testing/matter_testing_support.py | 4 ++ 6 files changed, 89 insertions(+), 28 deletions(-) diff --git a/src/controller/python/ChipDeviceController-ScriptBinding.cpp b/src/controller/python/ChipDeviceController-ScriptBinding.cpp index b4f2edb2955005..d78c9da8ed74d3 100644 --- a/src/controller/python/ChipDeviceController-ScriptBinding.cpp +++ b/src/controller/python/ChipDeviceController-ScriptBinding.cpp @@ -152,6 +152,8 @@ PyChipError pychip_DeviceController_EstablishPASESessionIP(chip::Controller::Dev uint32_t setupPINCode, chip::NodeId nodeid, uint16_t port); PyChipError pychip_DeviceController_EstablishPASESessionBLE(chip::Controller::DeviceCommissioner * devCtrl, uint32_t setupPINCode, uint16_t discriminator, chip::NodeId nodeid); +PyChipError pychip_DeviceController_EstablishPASESession(chip::Controller::DeviceCommissioner * devCtrl, const char * setUpCode, + chip::NodeId nodeid); PyChipError pychip_DeviceController_Commission(chip::Controller::DeviceCommissioner * devCtrl, chip::NodeId nodeid); PyChipError pychip_DeviceController_DiscoverCommissionableNodesLongDiscriminator(chip::Controller::DeviceCommissioner * devCtrl, @@ -600,6 +602,13 @@ PyChipError pychip_DeviceController_EstablishPASESessionBLE(chip::Controller::De return ToPyChipError(devCtrl->EstablishPASEConnection(nodeid, params)); } +PyChipError pychip_DeviceController_EstablishPASESession(chip::Controller::DeviceCommissioner * devCtrl, const char * setUpCode, + chip::NodeId nodeid) +{ + sPairingDelegate.SetExpectingPairingComplete(true); + return ToPyChipError(devCtrl->EstablishPASEConnection(nodeid, setUpCode)); +} + PyChipError pychip_DeviceController_Commission(chip::Controller::DeviceCommissioner * devCtrl, chip::NodeId nodeid) { CommissioningParameters params; diff --git a/src/controller/python/chip/ChipDeviceCtrl.py b/src/controller/python/chip/ChipDeviceCtrl.py index 2400d543f6d09e..318c2b56ef3d4d 100644 --- a/src/controller/python/chip/ChipDeviceCtrl.py +++ b/src/controller/python/chip/ChipDeviceCtrl.py @@ -500,6 +500,15 @@ def EstablishPASESessionIP(self, ipaddr: str, setupPinCode: int, nodeid: int, po self.devCtrl, ipaddr.encode("utf-8"), setupPinCode, nodeid, port) ) + def EstablishPASESession(self, setUpCode: str, nodeid: int): + self.CheckIsActive() + + self.state = DCState.RENDEZVOUS_ONGOING + return self._ChipStack.CallAsync( + lambda: self._dmLib.pychip_DeviceController_EstablishPASESession( + self.devCtrl, setUpCode.encode("utf-8"), nodeid) + ) + def GetTestCommissionerUsed(self): return self._ChipStack.Call( lambda: self._dmLib.pychip_TestCommissionerUsed() @@ -1588,6 +1597,9 @@ def _InitLib(self): self._dmLib.pychip_DeviceController_EstablishPASESessionBLE.argtypes = [ c_void_p, c_uint32, c_uint16, c_uint64] self._dmLib.pychip_DeviceController_EstablishPASESessionBLE.restype = PyChipError + self._dmLib.pychip_DeviceController_EstablishPASESession.argtypes = [ + c_void_p, c_char_p, c_uint64] + self._dmLib.pychip_DeviceController_EstablishPASESession.restype = PyChipError self._dmLib.pychip_DeviceController_DiscoverAllCommissionableNodes.argtypes = [ c_void_p] diff --git a/src/controller/python/chip/clusters/ClusterObjects.py b/src/controller/python/chip/clusters/ClusterObjects.py index 0aef317257014f..111274c884a8dd 100644 --- a/src/controller/python/chip/clusters/ClusterObjects.py +++ b/src/controller/python/chip/clusters/ClusterObjects.py @@ -301,7 +301,7 @@ def __init_subclass__(cls, *args, **kwargs) -> None: """Register a subclass.""" super().__init_subclass__(*args, **kwargs) try: - if cls.cluster_id not in ALL_ATTRIBUTES: + if cls.standard_attribute and cls.cluster_id not in ALL_ATTRIBUTES: ALL_ATTRIBUTES[cls.cluster_id] = {} # register this clusterattribute in the ALL_ATTRIBUTES dict for quick lookups ALL_ATTRIBUTES[cls.cluster_id][cls.attribute_id] = cls @@ -345,6 +345,10 @@ def attribute_type(cls) -> ClusterObjectFieldDescriptor: def must_use_timed_write(cls) -> bool: return False + @ChipUtility.classproperty + def standard_attribute(cls) -> bool: + return True + @ChipUtility.classproperty def _cluster_object(cls) -> ClusterObject: return make_dataclass('InternalClass', diff --git a/src/python_testing/TC_DeviceBasicComposition.py b/src/python_testing/TC_DeviceBasicComposition.py index 409ae97c1a4106..5bfbc5c44e839b 100644 --- a/src/python_testing/TC_DeviceBasicComposition.py +++ b/src/python_testing/TC_DeviceBasicComposition.py @@ -23,7 +23,11 @@ import chip.clusters.ClusterObjects import chip.tlv from basic_composition_support import BasicCompositionTests +from chip import ChipUtility from chip.clusters.Attribute import ValueDecodeFailure +from chip.clusters.ClusterObjects import ClusterAttributeDescriptor, ClusterObjectFieldDescriptor +from chip.interaction_model import InteractionModelError, Status +from chip.tlv import uint from global_attribute_ids import GlobalAttributeIds from matter_testing_support import (AttributePathLocation, ClusterPathLocation, CommandPathLocation, MatterBaseTest, async_test_body, default_matter_test_main) @@ -165,7 +169,41 @@ def test_TC_DT_1_1(self): if not success: self.fail_current_test("At least one endpoint was missing the descriptor cluster.") - def test_TC_IDM_10_1(self): + async def _read_non_standard_attribute_check_unsupported_read(self, endpoint_id, cluster_id, attribute_id) -> bool: + @dataclass + class TempAttribute(ClusterAttributeDescriptor): + @ChipUtility.classproperty + def cluster_id(cls) -> int: + return cluster_id + + @ChipUtility.classproperty + def attribute_id(cls) -> int: + return attribute_id + + @ChipUtility.classproperty + def attribute_type(cls) -> ClusterObjectFieldDescriptor: + return ClusterObjectFieldDescriptor(Type=uint) + + @ChipUtility.classproperty + def standard_attribute(cls) -> bool: + return False + + value: 'uint' = 0 + + result = await self.default_controller.Read(nodeid=self.dut_node_id, attributes=[(endpoint_id, TempAttribute)]) + try: + attr_ret = result.tlvAttributes[endpoint_id][cluster_id][attribute_id] + except KeyError: + attr_ret = None + + error_type_ok = attr_ret is not None and isinstance( + attr_ret, Clusters.Attribute.ValueDecodeFailure) and isinstance(attr_ret.Reason, InteractionModelError) + + got_expected_error = error_type_ok and attr_ret.Reason.status == Status.UnsupportedRead + return got_expected_error + + @async_test_body + async def test_TC_IDM_10_1(self): self.print_step(1, "Perform a wildcard read of attributes on all endpoints - already done") @dataclass @@ -222,6 +260,10 @@ class RequiredMandatoryAttribute: problem=f"Failed validation of value on {location.as_string(self.cluster_mapper)}: {str(e)}", spec_location="Global Elements") success = False continue + except KeyError: + # A KeyError here means the attribute does not exist. This problem was already recorded in step 2, + # but we don't assert until the end of the test, so ignore this and don't re-record the error. + continue self.print_step(4, "Validate the attribute list exactly matches the set of reported attributes") if success: @@ -236,15 +278,19 @@ class RequiredMandatoryAttribute: logging.debug( f"Checking presence of claimed supported {attribute_string} on {location.as_cluster_string(self.cluster_mapper)}: {'found' if has_attribute else 'not_found'}") - # Check attribute is actually present. if not has_attribute: - # TODO: Handle detecting write-only attributes from schema. - if "WriteOnly" in attribute_string: - continue - - self.record_error(self.get_test_name(), location=location, - problem=f"Did not find {attribute_string} on {location.as_cluster_string(self.cluster_mapper)} when it was claimed in AttributeList ({attribute_list})", spec_location="AttributeList Attribute") - success = False + # Check if this is a write-only attribute by trying to read it. + # If it's present and write-only it should return an UNSUPPORTED_READ error. All other errors are a failure. + # Because these can be MEI attributes, we need to build the ClusterAttributeDescriptor manually since it's + # not guaranteed to be generated. Since we expect an error back anyway, the type doesn't matter. + + write_only_attribute = await self._read_non_standard_attribute_check_unsupported_read( + endpoint_id=endpoint_id, cluster_id=cluster_id, attribute_id=attribute_id) + + if not write_only_attribute: + self.record_error(self.get_test_name(), location=location, + problem=f"Did not find {attribute_string} on {location.as_cluster_string(self.cluster_mapper)} when it was claimed in AttributeList ({attribute_list})", spec_location="AttributeList Attribute") + success = False continue attribute_value = cluster[attribute_id] diff --git a/src/python_testing/basic_composition_support.py b/src/python_testing/basic_composition_support.py index 8a6494eccea42f..4a516f67ecbc97 100644 --- a/src/python_testing/basic_composition_support.py +++ b/src/python_testing/basic_composition_support.py @@ -105,24 +105,10 @@ async def setup_class_helper(self, default_to_pase: bool = True): dump_device_composition_path: Optional[str] = self.user_params.get("dump_device_composition_path", None) if do_test_over_pase: - info = self.get_setup_payload_info() - - commissionable_nodes = dev_ctrl.DiscoverCommissionableNodes( - info.filter_type, info.filter_value, stopOnFirst=True, timeoutSecond=15) - logging.info(f"Commissionable nodes: {commissionable_nodes}") - # TODO: Support BLE - if commissionable_nodes is not None and len(commissionable_nodes) > 0: - commissionable_node = commissionable_nodes[0] - instance_name = f"{commissionable_node.instanceName}._matterc._udp.local" - vid = f"{commissionable_node.vendorId}" - pid = f"{commissionable_node.productId}" - address = f"{commissionable_node.addresses[0]}" - logging.info(f"Found instance {instance_name}, VID={vid}, PID={pid}, Address={address}") - - node_id = 1 - dev_ctrl.EstablishPASESessionIP(address, info.passcode, node_id) - else: - asserts.fail("Failed to find the DUT according to command line arguments.") + setupCode = self.matter_test_config.qr_code_content if self.matter_test_config.qr_code_content is not None else self.matter_test_config.manual_code + asserts.assert_true(setupCode, "Require either --qr-code or --manual-code.") + node_id = self.dut_node_id + dev_ctrl.EstablishPASESession(setupCode, node_id) else: # Using the already commissioned node node_id = self.dut_node_id diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index 5a5e9db1820845..7efe5ab948966b 100644 --- a/src/python_testing/matter_testing_support.py +++ b/src/python_testing/matter_testing_support.py @@ -60,6 +60,7 @@ from chip.setup_payload import SetupPayload from chip.storage import PersistentStorage from chip.tracing import TracingContext +from global_attribute_ids import GlobalAttributeIds from mobly import asserts, base_test, signals, utils from mobly.config_parser import ENV_MOBLY_LOGPATH, TestRunConfig from mobly.test_runner import TestRunner @@ -412,6 +413,9 @@ def get_cluster_string(self, cluster_id: int) -> str: return f"Cluster {name} ({cluster_id}, 0x{cluster_id:04X})" def get_attribute_string(self, cluster_id: int, attribute_id) -> str: + global_attrs = [item.value for item in GlobalAttributeIds] + if attribute_id in global_attrs: + return f"Attribute {GlobalAttributeIds(attribute_id).to_name()} {attribute_id}, 0x{attribute_id:04X}" mapping = self._mapping._CLUSTER_ID_DICT.get(cluster_id, None) if not mapping: return f"Attribute Unknown ({attribute_id}, 0x{attribute_id:08X})"