IDM-10.2: Handle MACL feature

.github/workflows/tests.yaml

@@ -522,6 +522,7 @@ jobs:
                   scripts/run_in_python_env.sh out/venv 'python3 ./src/python_testing/TestSpecParsingSupport.py'
                   scripts/run_in_python_env.sh out/venv 'python3 ./src/python_testing/TestConformanceTest.py'
                   scripts/run_in_python_env.sh out/venv 'python3 ./src/python_testing/TestConformanceSupport.py'
+                  scripts/run_in_python_env.sh out/venv 'python3 ./src/python_testing/TestConformanceTest.py'
                   scripts/run_in_python_env.sh out/venv 'python3 ./src/python_testing/TestChoiceConformanceSupport.py'
                   scripts/run_in_python_env.sh out/venv 'python3 ./src/python_testing/test_testing/test_IDM_10_4.py'
                   scripts/run_in_python_env.sh out/venv 'python3 ./src/python_testing/test_testing/test_TC_SC_7_1.py'

src/python_testing/TC_DeviceConformance.py

@@ -50,6 +50,24 @@ async def setup_class_helper(self):
         self.xml_device_types, problems = build_xml_device_types()
         self.problems.extend(problems)
 
+    def _get_device_type_id(self, device_type_name: str) -> int:
+        id = [id for id, dt in self.xml_device_types.items() if dt.name.lower() == device_type_name.lower()]
+        if len(id) != 1:
+            self.fail_current_test(f"Unable to find {device_type_name} device type")
+        return id[0]
+
+    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():
+            desc = Clusters.Descriptor
+            device_types = [dt.deviceType for dt in endpoint[desc.id][desc.Attributes.DeviceTypeList.attribute_id]]
+            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
+        return False
+
     def check_conformance(self, ignore_in_progress: bool, is_ci: bool, allow_provisional: bool):
         problems = []
         success = True
@@ -120,6 +138,13 @@ def record_warning(location, problem):
                 for f in feature_masks:
                     location = AttributePathLocation(endpoint_id=endpoint_id, cluster_id=cluster_id,
                                                      attribute_id=GlobalAttributeIds.FEATURE_MAP_ID)
+                    if cluster_id == Clusters.AccessControl.id and f == Clusters.AccessControl.Bitmaps.Feature.kManagedDevice:
+                        # Managed ACL is treated as a special case because it is only allowed if other endpoints support NIM and disallowed otherwise.
+                        if not self._has_device_type_supporting_macl():
+                            record_error(
+                                location=location, problem="MACL feature is disallowed if the a supported device type is not present")
+                        continue
+
                     if f not in self.xml_clusters[cluster_id].features.keys():
                         record_error(location=location, problem=f'Unknown feature with mask 0x{f:02x}')
                         continue

src/python_testing/TestConformanceTest.py

@@ -18,6 +18,8 @@
 from typing import Any
 
 import chip.clusters as Clusters
+from conformance_support import ConformanceDecision
+from global_attribute_ids import GlobalAttributeIds
 from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
 from mobly import asserts
 from spec_parsing_support import build_xml_clusters, build_xml_device_types
@@ -109,6 +111,10 @@ def create_onoff_endpoint(endpoint: int) -> dict[int, dict[int, dict[int, Any]]]
     return endpoint_tlv
 
 
+def is_mandatory(conformance):
+    return conformance(0, [], []).decision == ConformanceDecision.MANDATORY
+
+
 class TestConformanceSupport(MatterBaseTest, DeviceConformanceTests):
     def setup_class(self):
         self.xml_clusters, self.problems = build_xml_clusters()
@@ -135,6 +141,96 @@ 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 _create_minimal_cluster(self, cluster_id: int) -> dict[int, Any]:
+        attrs = {}
+        attrs[GlobalAttributeIds.FEATURE_MAP_ID] = 0
+
+        mandatory_attributes = [id for id, a in self.xml_clusters[cluster_id].attributes.items() if is_mandatory(a.conformance)]
+        for m in mandatory_attributes:
+            # dummy versions - we're not using the values in this test
+            attrs[m] = 0
+        attrs[GlobalAttributeIds.ATTRIBUTE_LIST_ID] = mandatory_attributes
+        mandatory_accepted_commands = [id for id, a in self.xml_clusters[cluster_id].accepted_commands.items()
+                                       if is_mandatory(a.conformance)]
+        attrs[GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID] = mandatory_accepted_commands
+        mandatory_generated_commands = [id for id, a in self.xml_clusters[cluster_id].generated_commands.items()
+                                        if is_mandatory(a.conformance)]
+        attrs[GlobalAttributeIds.GENERATED_COMMAND_LIST_ID] = mandatory_generated_commands
+        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.
+            Does NOT take into account overrides yet.
+        '''
+        endpoint_tlv = {}
+        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)
+
+        # Descriptor
+        attr = Clusters.Descriptor.Attributes
+        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
+
+    def add_macl(self, root_endpoint: dict[int, dict[int, Any]]):
+        ac = Clusters.AccessControl
+        root_endpoint[ac.id][ac.Attributes.FeatureMap.attribute_id] = ac.Bitmaps.Feature.kManagedDevice
+        root_endpoint[ac.id][ac.Attributes.Arl.attribute_id] = []
+        root_endpoint[ac.id][ac.Attributes.CommissioningARL.attribute_id] = []
+        root_endpoint[ac.id][ac.Attributes.AttributeList.attribute_id].extend([
+            ac.Attributes.Arl.attribute_id, ac.Attributes.CommissioningARL.attribute_id])
+        root_endpoint[ac.id][ac.Attributes.AcceptedCommandList.attribute_id].append(ac.Commands.ReviewFabricRestrictions.command_id)
+        root_endpoint[ac.id][ac.Attributes.GeneratedCommandList.attribute_id].append(
+            ac.Commands.ReviewFabricRestrictionsResponse.command_id)
+
+    @async_test_body
+    async def test_macl_handling(self):
+        nim_id = self._get_device_type_id('network infrastructure manager')
+        root_node_id = self._get_device_type_id('root node')
+        on_off_id = self._get_device_type_id('On/Off Light')
+
+        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}
+        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)
+        self.problems.extend(problems)
+        asserts.assert_true(success, "Unexpected failure parsing minimal dt")
+
+        self.add_macl(root)
+        # A MACL is allowed when there is a NIM, so this should succeed as well
+        success, problems = self.check_conformance(ignore_in_progress=False, is_ci=False, allow_provisional=True)
+        self.problems.extend(problems)
+        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)
+        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")
+
+        # TODO: what happens if there is a NIM and a non-NIM endpoint?
+
 
 if __name__ == "__main__":
     default_matter_test_main()

src/python_testing/execute_python_tests.py

@@ -85,6 +85,7 @@ def main(search_directory, env_file):
         "TestChoiceConformanceSupport.py",
         "TC_DEMTestBase.py",
         "choice_conformance_support.py",
+        "TestConformanceTest.py",  # Unit test of the conformance test (TC_DeviceConformance) - does not run against an app.
         "TestIdChecks.py",
         "TestSpecParsingDeviceType.py",
         "TestConformanceTest.py",