Fix key error when using TLV dictionary in DeviceConformance test. (#37268) (#37465)

* Change reference from TLV endpoint

* Change test endpoint function

* Fix issue with attr list

* Add handling for TLV attributes

Co-authored-by: Moises Terrones <moises.tersal@gmail.com>

Author: cecille

src/python_testing/TC_DeviceConformance.py

@@ -60,9 +60,9 @@ def _has_device_type_supporting_macl(self):
         # Currently this is just NIM. We may later be able to pull this from the device type scrape using the ManagedAclAllowed condition,
         # but these are not currently exposed directly by the device.
         allowed_ids = [self._get_device_type_id('network infrastructure manager')]
-        for endpoint in self.endpoints_tlv.values():
+        for endpoint in self.endpoints.values():
             desc = Clusters.Descriptor
-            device_types = [dt.deviceType for dt in endpoint[desc.id][desc.Attributes.DeviceTypeList.attribute_id]]
+            device_types = [dt.deviceType for dt in endpoint[desc][desc.Attributes.DeviceTypeList]]
             if set(allowed_ids).intersection(set(device_types)):
                 # TODO: it's unclear if this needs to be present on every endpoint. Right now, this assumes one is sufficient.
                 return True

src/python_testing/TestConformanceTest.py

@@ -15,7 +15,7 @@
 #    limitations under the License.
 #
 
-from typing import Any
+from typing import Any, Optional
 
 import chip.clusters as Clusters
 from basic_composition_support import arls_populated
@@ -142,6 +142,12 @@ async def test_provisional_cluster(self):
         success, problems = self.check_conformance(ignore_in_progress=False, is_ci=False, allow_provisional=False)
         asserts.assert_true(success, "Unexpected failure parsing endpoint with no clusters marked as provisional")
 
+    def _get_field_by_label(self, cl_object: Clusters.ClusterObjects.ClusterObject, label: str) -> Optional[Clusters.ClusterObjects.ClusterObjectFieldDescriptor]:
+        for field in cl_object.descriptor.Fields:
+            if field.Label == label:
+                return field
+        return None
+
     def _create_minimal_cluster(self, cluster_id: int) -> dict[int, Any]:
         attrs = {}
         attrs[GlobalAttributeIds.FEATURE_MAP_ID] = 0
@@ -160,37 +166,61 @@ def _create_minimal_cluster(self, cluster_id: int) -> dict[int, Any]:
         attrs[GlobalAttributeIds.CLUSTER_REVISION_ID] = self.xml_clusters[cluster_id].revision
         return attrs
 
-    def _create_minimal_dt(self, device_type_id: int) -> dict[int, dict[int, Any]]:
-        ''' Creates the internals of an endpoint_tlv with the minimal set of clusters, with the minimal set of attributes and commands. Global attributes only.
+    def _create_minimal_dt(self, device_type_id: int, is_tlv_endpoint: bool = True) -> dict[int, dict[int, Any]]:
+        ''' Creates the internals of an endpoint with the minimal set of clusters, with the minimal set of attributes and commands. Global attributes only.
             Does NOT take into account overrides yet.
         '''
-        endpoint_tlv = {}
+        endpoint = {}
         required_servers = [id for id, c in self.xml_device_types[device_type_id].server_clusters.items()
                             if is_mandatory(c.conformance)]
         required_clients = [id for id, c in self.xml_device_types[device_type_id].client_clusters.items()
                             if is_mandatory(c.conformance)]
         device_type_revision = self.xml_device_types[device_type_id].revision
 
         for s in required_servers:
-            endpoint_tlv[s] = self._create_minimal_cluster(s)
+            endpoint[s] = self._create_minimal_cluster(s)
 
         # Descriptor
         attr = Clusters.Descriptor.Attributes
+        structs = Clusters.Descriptor.Structs
         attrs = {}
-        attrs[attr.FeatureMap.attribute_id] = 0
-        attrs[attr.AcceptedCommandList.attribute_id] = []
-        attrs[attr.GeneratedCommandList.attribute_id] = []
-        attrs[attr.ClusterRevision.attribute_id] = self.xml_clusters[Clusters.Descriptor.id].revision
-        attrs[attr.DeviceTypeList.attribute_id] = [
-            Clusters.Descriptor.Structs.DeviceTypeStruct(deviceType=device_type_id, revision=device_type_revision)]
-        attrs[attr.ServerList.attribute_id] = required_servers
-        attrs[attr.ClientList.attribute_id] = required_clients
-        attrs[attr.PartsList.attribute_id] = []
-        attrs[attr.AttributeList.attribute_id] = []
-        attrs[attr.AttributeList.attribute_id] = list(attrs.keys())
-
-        endpoint_tlv[Clusters.Descriptor.id] = attrs
-        return endpoint_tlv
+
+        attributes = [
+            attr.FeatureMap,
+            attr.AcceptedCommandList,
+            attr.GeneratedCommandList,
+            attr.ClusterRevision,
+            attr.DeviceTypeList,
+            attr.ServerList,
+            attr.ClientList,
+            attr.PartsList,
+        ]
+
+        attribute_values = [
+            (0, 0),  # FeatureMap
+            ([], []),  # AcceptedCommandList
+            ([], []),  # GeneratedCommandList
+            (self.xml_clusters[Clusters.Descriptor.id].revision,
+             self.xml_clusters[Clusters.Descriptor.id].revision),  # ClusterRevision
+            ([{self._get_field_by_label(structs.DeviceTypeStruct, "deviceType").Tag: device_type_id,
+               self._get_field_by_label(structs.DeviceTypeStruct, "revision").Tag: device_type_revision}],
+             [Clusters.Descriptor.Structs.DeviceTypeStruct(
+                 deviceType=device_type_id, revision=device_type_revision)]),  # DeviceTypeList
+            (required_servers, required_servers),  # ServerList
+            (required_clients, required_clients),  # ClientList
+            ([], []),  # PartsList
+        ]
+
+        for attribute_name, attribute_value in zip(attributes, attribute_values):
+            key = attribute_name.attribute_id if is_tlv_endpoint else attribute_name
+            attrs[key] = attribute_value[0] if is_tlv_endpoint else attribute_value[1]
+
+        # Append the attribute list now that is populated.
+        attrs[attr.AttributeList.attribute_id if is_tlv_endpoint else attr.AttributeList] = list(attrs.keys())
+
+        endpoint[Clusters.Descriptor.id if is_tlv_endpoint else Clusters.Descriptor] = attrs
+
+        return endpoint
 
     def add_macl(self, root_endpoint: dict[int, dict[int, Any]], populate_arl: bool = False, populate_commissioning_arl: bool = False):
         ac = Clusters.AccessControl
@@ -220,6 +250,10 @@ async def test_macl_handling(self):
         root = self._create_minimal_dt(device_type_id=root_node_id)
         nim = self._create_minimal_dt(device_type_id=nim_id)
         self.endpoints_tlv = {0: root, 1: nim}
+
+        root_no_tlv = self._create_minimal_dt(device_type_id=root_node_id, is_tlv_endpoint=False)
+        nim_no_tlv = self._create_minimal_dt(device_type_id=nim_id, is_tlv_endpoint=False)
+        self.endpoints = {0: root_no_tlv, 1: nim_no_tlv}
         asserts.assert_true(self._has_device_type_supporting_macl(), "Did not find supported device in generated device")
 
         success, problems = self.check_conformance(ignore_in_progress=False, is_ci=False, allow_provisional=True)
@@ -233,7 +267,7 @@ async def test_macl_handling(self):
         asserts.assert_true(success, "Unexpected failure with NIM and MACL")
 
         # A MACL is not allowed when there is no NIM
-        self.endpoints_tlv[1] = self._create_minimal_dt(device_type_id=on_off_id)
+        self.endpoints[1] = self._create_minimal_dt(device_type_id=on_off_id, is_tlv_endpoint=False)
         success, problems = self.check_conformance(ignore_in_progress=False, is_ci=False, allow_provisional=True)
         self.problems.extend(problems)
         asserts.assert_false(success, "Unexpected success with On/Off and MACL")