Skip to content

Commit

Permalink
Fix key error when using TLV dictionary in DeviceConformance test. (#…
Browse files Browse the repository at this point in the history
…37268)

* Change reference from TLV endpoint

* Change test endpoint function

* Fix issue with attr list

* Add handling for TLV attributes
  • Loading branch information
tersal authored and cecille committed Feb 7, 2025
1 parent f9f7909 commit 32a413c
Show file tree
Hide file tree
Showing 2 changed files with 56 additions and 22 deletions.
4 changes: 2 additions & 2 deletions src/python_testing/TC_DeviceConformance.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
74 changes: 54 additions & 20 deletions src/python_testing/TestConformanceTest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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")
Expand Down

0 comments on commit 32a413c

Please sign in to comment.