TC-IDM-10.1: Support write-only attributes

src/controller/python/ChipDeviceController-ScriptBinding.cpp

@@ -153,6 +153,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,
@@ -620,6 +622,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;

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()
@@ -1605,6 +1614,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]

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',

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,43 @@ 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
+
+        print(attr_ret)
+
+        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 +262,9 @@ 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:
+                            # This would have been caught already in the previous step
+                            continue
 
         self.print_step(4, "Validate the attribute list exactly matches the set of reported attributes")
         if success:
@@ -236,15 +279,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]

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

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})"