MACL: Add checks that ARL / CommissioningARL are populated or not populated as required by the test (project-chip#35413)

* Add check for unpopulated ARLs

* linter

* Add check to TC-ACL-2.11 for populated ARLs

Author: cecille

src/python_testing/TC_ACL_2_11.py

@@ -31,6 +31,7 @@
 import queue
 
 import chip.clusters as Clusters
+from basic_composition_support import arls_populated
 from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction, ValueDecodeFailure
 from chip.clusters.ClusterObjects import ALL_ACCEPTED_COMMANDS, ALL_ATTRIBUTES, ALL_CLUSTERS, ClusterEvent
 from chip.clusters.Objects import AccessControl
@@ -82,7 +83,7 @@ def desc_TC_ACL_2_11(self) -> str:
 
     def steps_TC_ACL_2_11(self) -> list[TestStep]:
         steps = [
-            TestStep(1, "Commissioning, already done"),
+            TestStep(1, "Commissioning (already done) and precondition checks", is_commissioning=True),
             TestStep(2, "TH1 reads DUT Endpoint 0 AccessControl cluster CommissioningARL attribute"),
             TestStep(3, "TH1 reads DUT Endpoint 0 AccessControl cluster ARL attribute"),
             TestStep(4, "For each entry in ARL, iterate over each restriction and attempt access the restriction's ID on the Endpoint and Cluster in the ARL entry.",
@@ -102,6 +103,14 @@ async def test_TC_ACL_2_11(self):
         dev_ctrl = self.default_controller
         dut_node_id = self.dut_node_id
         self.step(1)
+
+        wildcard_read = (await dev_ctrl.Read(self.dut_node_id, [()]))
+        has_arl, has_carl = arls_populated(wildcard_read.tlvAttributes)
+        asserts.assert_true(
+            has_arl, "ARL attribute must contain at least one restriction to run this test. Please follow manufacturer-specific steps to add access restrictions and re-run this test")
+        asserts.assert_true(
+            has_carl, "CommissioningARL attribute must contain at least one restriction to run this test. Please follow manufacturer-specific steps to add access restrictions and re-run this test")
+
         self.step(2)
         await self.read_single_attribute_check_success(
             endpoint=0,

src/python_testing/TestConformanceTest.py

@@ -18,6 +18,7 @@
 from typing import Any
 
 import chip.clusters as Clusters
+from basic_composition_support import arls_populated
 from conformance_support import ConformanceDecision
 from global_attribute_ids import GlobalAttributeIds
 from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
@@ -191,7 +192,7 @@ def _create_minimal_dt(self, device_type_id: int) -> dict[int, dict[int, Any]]:
         endpoint_tlv[Clusters.Descriptor.id] = attrs
         return endpoint_tlv
 
-    def add_macl(self, root_endpoint: dict[int, dict[int, Any]]):
+    def add_macl(self, root_endpoint: dict[int, dict[int, Any]], populate_arl: bool = False, populate_commissioning_arl: bool = False):
         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] = []
@@ -202,6 +203,14 @@ def add_macl(self, root_endpoint: dict[int, dict[int, Any]]):
         root_endpoint[ac.id][ac.Attributes.GeneratedCommandList.attribute_id].append(
             ac.Commands.ReviewFabricRestrictionsResponse.command_id)
 
+        generic_restriction = ac.Structs.AccessRestrictionStruct(
+            type=ac.Enums.AccessRestrictionTypeEnum.kAttributeAccessForbidden, id=1)
+        entry = ac.Structs.CommissioningAccessRestrictionEntryStruct(endpoint=1, cluster=2, restrictions=generic_restriction)
+        if populate_arl:
+            root_endpoint[ac.id][ac.Attributes.Arl.attribute_id] = [entry]
+        if populate_commissioning_arl:
+            root_endpoint[ac.id][ac.Attributes.CommissioningARL.attribute_id] = [entry]
+
     @async_test_body
     async def test_macl_handling(self):
         nim_id = self._get_device_type_id('network infrastructure manager')
@@ -231,6 +240,45 @@ async def test_macl_handling(self):
 
         # TODO: what happens if there is a NIM and a non-NIM endpoint?
 
+    @async_test_body
+    async def test_macl_restrictions(self):
+
+        nim_id = self._get_device_type_id('network infrastructure manager')
+        root_node_id = self._get_device_type_id('root node')
+
+        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}
+
+        # device with no macl
+        have_arl, have_carl = arls_populated(self.endpoints_tlv)
+        asserts.assert_false(have_arl, "Unexpected ARL found")
+        asserts.assert_false(have_carl, "Unexpected CommissioningARL found")
+
+        # device with unpopulated macl
+        self.add_macl(root)
+        have_arl, have_carl = arls_populated(self.endpoints_tlv)
+        asserts.assert_false(have_arl, "Unexpected ARL found")
+        asserts.assert_false(have_carl, "Unexpected CommissioningARL found")
+
+        # device with populated ARL
+        self.add_macl(root, populate_arl=True)
+        have_arl, have_carl = arls_populated(self.endpoints_tlv)
+        asserts.assert_true(have_arl, "Did not find expected ARL")
+        asserts.assert_false(have_carl, "Unexpected CommissioningARL found")
+
+        # device with populated commissioning ARL
+        self.add_macl(root, populate_commissioning_arl=True)
+        have_arl, have_carl = arls_populated(self.endpoints_tlv)
+        asserts.assert_false(have_arl, "Unexpected ARL found")
+        asserts.assert_true(have_carl, "Did not find expected Commissioning ARL")
+
+        # device with both
+        self.add_macl(root, populate_arl=True, populate_commissioning_arl=True)
+        have_arl, have_carl = arls_populated(self.endpoints_tlv)
+        asserts.assert_true(have_arl, "Did not find expected ARL")
+        asserts.assert_true(have_carl, "Did not find expected Commissioning ARL")
+
 
 if __name__ == "__main__":
     default_matter_test_main()

src/python_testing/basic_composition_support.py

@@ -26,12 +26,30 @@
 from pprint import pformat, pprint
 from typing import Any, Optional
 
+import chip.clusters as Clusters
 import chip.clusters.ClusterObjects
 import chip.tlv
 from chip.clusters.Attribute import ValueDecodeFailure
 from mobly import asserts
 
 
+def arls_populated(tlv_data: dict[int, Any]) -> tuple[bool, bool]:
+    """ Returns a tuple indicating if the ARL and CommissioningARL are populated.
+        Requires a wildcard read of the device TLV.
+    """
+    # ACL is always on endpoint 0
+    if 0 not in tlv_data or Clusters.AccessControl.id not in tlv_data[0]:
+        return (False, False)
+    # Both attributes are mandatory for this feature, so if one doesn't exist, neither should the other.
+    if Clusters.AccessControl.Attributes.Arl.attribute_id not in tlv_data[0][Clusters.AccessControl.id][Clusters.AccessControl.Attributes.AttributeList.attribute_id]:
+        return (False, False)
+
+    have_arl = tlv_data[0][Clusters.AccessControl.id][Clusters.AccessControl.Attributes.Arl.attribute_id]
+    have_carl = tlv_data[0][Clusters.AccessControl.id][Clusters.AccessControl.Attributes.CommissioningARL.attribute_id]
+
+    return (have_arl, have_carl)
+
+
 def MatterTlvToJson(tlv_data: dict[int, Any]) -> dict[str, Any]:
     """Given TLV data for a specific cluster instance, convert to the Matter JSON format."""
 
@@ -169,6 +187,12 @@ async def setup_class_helper(self, allow_pase: bool = True):
         logging.info("Start of actual tests")
         logging.info("###########################################################")
 
+        have_arl, have_carl = arls_populated(self.endpoints_tlv)
+        asserts.assert_false(
+            have_arl, "ARL cannot be populated for this test - Please follow manufacturer-specific steps to remove the access restrictions and re-run this test")
+        asserts.assert_false(
+            have_carl, "CommissioningARL cannot be populated for this test - Please follow manufacturer-specific steps to remove the access restrictions and re-run this test")
+
     def get_test_name(self) -> str:
         """Return the function name of the caller. Used to create logging entries."""
         return sys._getframe().f_back.f_code.co_name