TC_RR_1_1.py

#
#    Copyright (c) 2022 Project CHIP Authors
#    All rights reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

import asyncio
import logging
import math
import queue
import random
import string
import time
from typing import Any, Dict, List, Set

import chip.clusters as Clusters
from chip.interaction_model import Status as StatusEnum
from chip.utils import CommissioningBuildingBlocks
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
from mobly import asserts
from TC_SC_3_6 import AttributeChangeAccumulator, ResubscriptionCatcher

# TODO: Overall, we need to add validation that session IDs have not changed throughout to be agnostic
#       to some internal behavior assumptions of the SDK we are making relative to the write to
#       the trigger the subscriptions not re-opening a new CASE session
#


def generate_controller_name(fabric_index: int, controller_index: int):
    return f"RD{fabric_index}{string.ascii_uppercase[controller_index]}"


class TC_RR_1_1(MatterBaseTest):
    def setup_class(self):
        self._pseudo_random_generator = random.Random(1234)
        self._subscriptions = []

    def teardown_class(self):
        logging.info("Teardown: shutting down all subscription to avoid racy callbacks")
        for subscription in self._subscriptions:
            subscription.Shutdown()

    @async_test_body
    async def test_TC_RR_1_1(self):
        dev_ctrl = self.default_controller

        # Debug/test arguments

        # Get overrides for debugging the test
        num_fabrics_to_commission = self.user_params.get("num_fabrics_to_commission", 5)
        num_controllers_per_fabric = self.user_params.get("num_controllers_per_fabric", 3)
        # Immediate reporting
        min_report_interval_sec = self.user_params.get("min_report_interval_sec", 0)
        # 10 minutes max reporting interval --> We don't care about keep-alives per-se and
        # want to avoid resubscriptions
        max_report_interval_sec = self.user_params.get("max_report_interval_sec", 10 * 60)
        # Time to wait after changing NodeLabel for subscriptions to all hit. This is dependant
        # on MRP params of subscriber and on actual min_report_interval.
        # TODO: Determine the correct max value depending on target. Test plan doesn't say!
        timeout_delay_sec = self.user_params.get("timeout_delay_sec", max_report_interval_sec * 2)
        # Whether to skip filling the UserLabel clusters
        skip_user_label_cluster_steps = self.user_params.get("skip_user_label_cluster_steps", False)
        # Whether to do the local session ID comparison checks to prove new sessions have not been established.
        check_local_session_id_unchanged = self.user_params.get("check_local_session_id_unchanged", False)
        # Whether to check heap statistics. Add `--bool-arg check_heap_watermarks:true` to command line to enable
        check_heap_watermarks = self.user_params.get("check_heap_watermarks", False)

        BEFORE_LABEL = "Before Subscriptions 12345678912"
        AFTER_LABEL = "After Subscriptions 123456789123"

        # Pre-conditions

        # Do a read-out of heap statistics before the test begins
        if check_heap_watermarks:
            logging.info("Read Heap info before stress test")
            high_watermark_before, current_usage_before = await self.read_heap_statistics(dev_ctrl)

        # Make sure all certificates are installed with maximal size
        dev_ctrl.fabricAdmin.certificateAuthority.maximizeCertChains = True

        # TODO: Do from PICS list. The reflection approach here what a real client would do,
        #       and it respects what the test says: "TH writes 4 entries per endpoint where LabelList is supported"
        logging.info("Pre-condition: determine whether any endpoints have UserLabel cluster (ULABEL.S.A0000(LabelList))")
        endpoints_with_user_label_list = await dev_ctrl.ReadAttribute(self.dut_node_id, [Clusters.UserLabel.Attributes.LabelList])
        has_user_labels = len(endpoints_with_user_label_list) > 0
        if has_user_labels:
            logging.info("--> User label cluster present on endpoints %s" %
                         ", ".join(["%d" % ep for ep in endpoints_with_user_label_list.keys()]))
        else:
            logging.info("--> User label cluster not present on any endpoitns")

        # Generate list of all clients names
        client_list = []

        # TODO: Shall we also verify SupportedFabrics attribute, and the CapabilityMinima attribute?
        logging.info("Pre-conditions: validate CapabilityMinima.CaseSessionsPerFabric >= 3")

        capability_minima = await self.read_single_attribute(dev_ctrl,
                                                             node_id=self.dut_node_id,
                                                             endpoint=0,
                                                             attribute=Clusters.BasicInformation.Attributes.CapabilityMinima)
        asserts.assert_greater_equal(capability_minima.caseSessionsPerFabric, 3)

        # Step 1: Commission 5 fabrics with maximized NOC chains. 1a and 1b have already been completed at this time.
        logging.info(f"Step 1: use existing fabric to configure new fabrics so that total is {num_fabrics_to_commission} fabrics")

        # Generate Node IDs for subsequent controllers start at 200, follow 200, 300, ...
        node_ids = [200 + (i * 100) for i in range(num_controllers_per_fabric - 1)]

        # Prepare clients for first fabric, that includes the default controller
        fabric_index = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.CurrentFabricIndex, dev_ctrl=dev_ctrl)
        dev_ctrl.name = generate_controller_name(fabric_index, 0)
        client_list.append(dev_ctrl)

        if num_controllers_per_fabric > 1:
            new_controllers = await CommissioningBuildingBlocks.CreateControllersOnFabric(
                fabricAdmin=dev_ctrl.fabricAdmin,
                adminDevCtrl=dev_ctrl,
                controllerNodeIds=node_ids,
                privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister,
                targetNodeId=self.dut_node_id, catTags=[0x0001_0001]
            )
            for idx, controller in enumerate(new_controllers):
                controller.name = generate_controller_name(fabric_index, idx+1)
            client_list.extend(new_controllers)

        # Step 1c - Ensure there are no leftover fabrics from another process.
        commissioned_fabric_count: int = await self.read_single_attribute(
            dev_ctrl, node_id=self.dut_node_id,
            endpoint=0, attribute=Clusters.OperationalCredentials.Attributes.CommissionedFabrics)

        # Insert a fabric to self-test the next step.
        # This is not hidden behind a flag to avoid potential undetected bugs.
        if commissioned_fabric_count == 1:
            logging.info("Commissioning fabric for TH test.")
            new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
            new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)

            new_admin_ctrl = new_fabric_admin.NewController(nodeId=dev_ctrl.nodeId, catTags=[0x0001_0001])
            new_admin_ctrl.name = "THTF"
            await CommissioningBuildingBlocks.AddNOCForNewFabricFromExisting(
                commissionerDevCtrl=dev_ctrl, newFabricDevCtrl=new_admin_ctrl,
                existingNodeId=self.dut_node_id, newNodeId=self.dut_node_id)

        commissioned_fabric_count = await self.read_single_attribute(
            dev_ctrl, node_id=self.dut_node_id,
            endpoint=0, attribute=Clusters.OperationalCredentials.Attributes.CommissionedFabrics)
        asserts.assert_not_equal(commissioned_fabric_count, 1, "TH Error: failed to add fabric for testing TH.")

        # Step 1c - perform removal.
        if commissioned_fabric_count > 1:
            logging.info("Removing extra fabrics from device.")
            fabrics: List[Clusters.OperationalCredentials.Structs.FabricDescriptorStruct] = await self.read_single_attribute(
                dev_ctrl, node_id=self.dut_node_id, endpoint=0,
                attribute=Clusters.OperationalCredentials.Attributes.Fabrics, fabricFiltered=False)
            current_fabric_index = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.CurrentFabricIndex)
            for fabric in fabrics:
                if fabric.fabricIndex == current_fabric_index:
                    continue
                # This is not the test client's fabric, so remove it.
                await dev_ctrl.SendCommand(
                    self.dut_node_id, 0, Clusters.OperationalCredentials.Commands.RemoveFabric(fabricIndex=fabric.fabricIndex))

        commissioned_fabric_count = await self.read_single_attribute(
            dev_ctrl, node_id=self.dut_node_id,
            endpoint=0, attribute=Clusters.OperationalCredentials.Attributes.CommissionedFabrics)
        asserts.assert_equal(commissioned_fabric_count, 1, "Failed to remove extra fabrics from DUT.")

        # Prepare clients for subsequent fabrics (step 1d)
        for i in range(num_fabrics_to_commission - 1):
            admin_index = 2 + i
            logging.info("Commissioning fabric %d/%d" % (admin_index, num_fabrics_to_commission))
            new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
            new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=admin_index)

            new_admin_ctrl = new_fabric_admin.NewController(nodeId=dev_ctrl.nodeId, catTags=[0x0001_0001])
            await CommissioningBuildingBlocks.AddNOCForNewFabricFromExisting(commissionerDevCtrl=dev_ctrl,
                                                                             newFabricDevCtrl=new_admin_ctrl,
                                                                             existingNodeId=self.dut_node_id,
                                                                             newNodeId=self.dut_node_id)
            fabric_index = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.CurrentFabricIndex, dev_ctrl=new_admin_ctrl)
            new_admin_ctrl.name = generate_controller_name(fabric_index, 0)
            client_list.append(new_admin_ctrl)
            if num_controllers_per_fabric > 1:
                new_controllers = await CommissioningBuildingBlocks.CreateControllersOnFabric(
                    fabricAdmin=new_fabric_admin,
                    adminDevCtrl=new_admin_ctrl,
                    controllerNodeIds=node_ids,
                    privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister,
                    targetNodeId=self.dut_node_id,
                    catTags=[0x0001_0001]
                )
                for idx, controller in enumerate(new_controllers):
                    controller.name = generate_controller_name(fabric_index, idx+1)

                client_list.extend(new_controllers)

        asserts.assert_equal(len(client_list), num_fabrics_to_commission *
                             num_controllers_per_fabric, "Must have the right number of clients")

        commissioned_fabric_count = await self.read_single_attribute(
            dev_ctrl, node_id=self.dut_node_id,
            endpoint=0, attribute=Clusters.OperationalCredentials.Attributes.CommissionedFabrics)
        asserts.assert_equal(commissioned_fabric_count, num_fabrics_to_commission,
                             "Must have the right number of fabrics commissioned.")
        fabric_table: List[Clusters.OperationalCredentials.Structs.FabricDescriptorStruct] = await self.read_single_attribute(
            dev_ctrl, node_id=self.dut_node_id,
            endpoint=0, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, fabricFiltered=False)

        client_by_name = {client.name: client for client in client_list}
        local_session_id_by_client_name = {client.name: client.GetConnectedDeviceSync(
            self.dut_node_id).localSessionId for client in client_list}

        # Step 2: Set the Label field for each fabric and BasicInformation.NodeLabel to 32 characters
        logging.info("Step 2: Setting the Label field for each fabric and BasicInformation.NodeLabel to 32 characters")

        for fabric in fabric_table:
            client_name = generate_controller_name(fabric.fabricIndex, 0)
            client = client_by_name[client_name]

            # Send the UpdateLabel command
            label = ("%d" % fabric.fabricIndex) * 32
            logging.info("Step 2a: Setting fabric label on fabric %d to '%s' using client %s" %
                         (fabric.fabricIndex, label, client_name))
            await client.SendCommand(self.dut_node_id, 0, Clusters.OperationalCredentials.Commands.UpdateFabricLabel(label))

            # Read back
            fabric_metadata = await self.read_single_attribute(client,
                                                               node_id=self.dut_node_id,
                                                               endpoint=0,
                                                               attribute=Clusters.OperationalCredentials.Attributes.Fabrics)
            print(fabric_metadata)
            asserts.assert_equal(fabric_metadata[0].label, label, "Fabrics[x].label must match what was written")

        # Before subscribing, set the NodeLabel to "Before Subscriptions"
        logging.info(f"Step 2b: Set BasicInformation.NodeLabel to {BEFORE_LABEL}")
        await client_list[0].WriteAttribute(self.dut_node_id,
                                            [(0, Clusters.BasicInformation.Attributes.NodeLabel(value=BEFORE_LABEL))])

        node_label = await self.read_single_attribute(client,
                                                      node_id=self.dut_node_id,
                                                      endpoint=0,
                                                      attribute=Clusters.BasicInformation.Attributes.NodeLabel)
        asserts.assert_equal(node_label, BEFORE_LABEL, "NodeLabel must match what was written")

        # Step 3: Add 4 Access Control entries on DUT with a list of 4 Subjects and 3 Targets with the following parameters (...)
        await self.send_acl(test_step=3, client_by_name=client_by_name, enable_access_to_group_cluster=False, fabric_table=fabric_table)

        # Step 4 and 5 (the operations cannot be separated): establish all CASE sessions and subscriptions

        # Subscribe with all clients to NodeLabel attribute and 2 more paths
        sub_handlers = []
        resub_catchers = []
        output_queue = queue.Queue()
        subscription_contents = [
            (0, Clusters.BasicInformation.Attributes.NodeLabel),  # Single attribute
            (0, Clusters.OperationalCredentials),  # Wildcard all of opcreds attributes on EP0
            Clusters.Descriptor  # All descriptors on all endpoints
        ]

        logging.info("Step 4 and 5 (first part): Establish subscription with all %d clients" % len(client_list))
        for sub_idx, client in enumerate(client_list):
            logging.info("Establishing subscription %d/%d from controller node %s" % (sub_idx + 1, len(client_list), client.name))

            sub = await client.ReadAttribute(
                nodeid=self.dut_node_id,
                attributes=subscription_contents,
                reportInterval=(min_report_interval_sec, max_report_interval_sec),
                keepSubscriptions=False
            )
            self._subscriptions.append(sub)

            attribute_handler = AttributeChangeAccumulator(
                name=client.name, expected_attribute=Clusters.BasicInformation.Attributes.NodeLabel, output=output_queue)
            sub.SetAttributeUpdateCallback(attribute_handler)
            sub_handlers.append(attribute_handler)

            # TODO: Replace resubscription catcher with API to disable re-subscription on failure
            resub_catcher = ResubscriptionCatcher(name=client.name)
            sub.SetResubscriptionAttemptedCallback(resub_catcher)
            resub_catchers.append(resub_catcher)

        asserts.assert_equal(len(self._subscriptions), len(client_list), "Must have the right number of subscriptions")

        # Step 6: Read 9 paths and validate success
        logging.info("Step 6: Read 9 paths (first 9 attributes of Basic Information cluster) and validate success")

        large_read_contents = [
            Clusters.BasicInformation.Attributes.DataModelRevision,
            Clusters.BasicInformation.Attributes.VendorName,
            Clusters.BasicInformation.Attributes.VendorID,
            Clusters.BasicInformation.Attributes.ProductName,
            Clusters.BasicInformation.Attributes.ProductID,
            Clusters.BasicInformation.Attributes.NodeLabel,
            Clusters.BasicInformation.Attributes.Location,
            Clusters.BasicInformation.Attributes.HardwareVersion,
            Clusters.BasicInformation.Attributes.HardwareVersionString,
        ]
        large_read_paths = [(0, attrib) for attrib in large_read_contents]
        basic_info = await dev_ctrl.ReadAttribute(self.dut_node_id, large_read_paths)

        # Make sure everything came back from the read that we expected
        asserts.assert_true(0 in basic_info.keys(), "Must have read endpoint 0 data")
        asserts.assert_true(Clusters.BasicInformation in basic_info[0].keys(), "Must have read Basic Information cluster data")
        for attribute in large_read_contents:
            asserts.assert_true(attribute in basic_info[0][Clusters.BasicInformation],
                                "Must have read back attribute %s" % (attribute.__name__))

        # Step 7: Trigger a change on NodeLabel
        logging.info(
            "Step 7: Change attribute with one client, await all attributes changed successfully without loss of subscriptions")
        await asyncio.sleep(1)
        await client_list[0].WriteAttribute(self.dut_node_id,
                                            [(0, Clusters.BasicInformation.Attributes.NodeLabel(value=AFTER_LABEL))])

        all_changes = {client.name: False for client in client_list}

        # Await a stabilization delay in increments to let the event loops run
        start_time = time.time()
        elapsed = 0
        time_remaining = timeout_delay_sec

        while time_remaining > 0:
            try:
                item = output_queue.get(block=True, timeout=time_remaining)
                client_name, endpoint, attribute, value = item['name'], item['endpoint'], item['attribute'], item['value']

                # Record arrival of an expected subscription change when seen
                if endpoint == 0 and attribute == Clusters.BasicInformation.Attributes.NodeLabel and value == AFTER_LABEL:
                    if not all_changes[client_name]:
                        logging.info("Got expected attribute change for client %s" % client_name)
                        all_changes[client_name] = True

                # We are done waiting when we have accumulated all results
                if all(all_changes.values()):
                    logging.info("All clients have reported, done waiting.")
                    break
            except queue.Empty:
                # No error, we update timeouts and keep going
                pass

            elapsed = time.time() - start_time
            time_remaining = timeout_delay_sec - elapsed

        logging.info("Step 7: Validation of results")
        sub_test_failed = False

        for catcher in resub_catchers:
            if catcher.caught_resubscription:
                logging.error("Client %s saw a resubscription" % catcher.name)
                sub_test_failed = True
            else:
                logging.info("Client %s correctly did not see a resubscription" % catcher.name)

        all_reports_gotten = all(all_changes.values())
        if not all_reports_gotten:
            logging.error("Missing reports from the following clients: %s" %
                          ", ".join([name for name, value in all_changes.items() if value is False]))
            sub_test_failed = True
        else:
            logging.info("Got successful reports from all clients, meaning all concurrent CASE sessions worked")

        # Determine result of Step 7
        if sub_test_failed:
            asserts.fail("Failed step 7 !")

        # Step 8: Validate sessions have not changed by doing a read on NodeLabel from all clients
        logging.info("Step 8a: Read back NodeLabel directly from all clients")
        for sub_idx, client in enumerate(client_list):
            logging.info("Reading NodeLabel (%d/%d) from controller node %s" % (sub_idx + 1, len(client_list), client.name))

            label_readback = await self.read_single_attribute(client,
                                                              node_id=self.dut_node_id,
                                                              endpoint=0,
                                                              attribute=Clusters.BasicInformation.Attributes.NodeLabel)
            asserts.assert_equal(label_readback, AFTER_LABEL)

        # On each client, read back the local session id for the CASE session to the DUT and ensure
        # it's the same as that of the session established right at the beginning of the test.
        # In tandem with checking that the number of sessions to the DUT is exactly one,
        # this ensures we have not established any new CASE
        # sessions in this test.
        if check_local_session_id_unchanged:
            logging.info("Step 8b: Validate that the local CASE session ID hasn't changed")
            num_failed_clients = 0

            for client in client_list:
                beginning_session_id = local_session_id_by_client_name[client.name]
                end_session_id = client.GetConnectedDeviceSync(self.dut_node_id).localSessionId
                total_sessions = client.GetConnectedDeviceSync(self.dut_node_id).numTotalSessions

                if (beginning_session_id != end_session_id):
                    logging.error(
                        f"Test ended with a different session ID created from what we had before for {client.name} "
                        f"(total sessions = {total_sessions})")
                    num_failed_clients = num_failed_clients + 1
                elif (total_sessions != 1):
                    logging.error(f"Test ended with more than 1 session for {client.name}")
                    num_failed_clients = num_failed_clients + 1

            if (num_failed_clients > 0):
                asserts.fail(f"Failed Step 8b: ({num_failed_clients} / {len(client_list)} failed)")

        # Step 9: Fill user label list
        if has_user_labels and not skip_user_label_cluster_steps:
            await self.fill_user_label_list(dev_ctrl, self.dut_node_id)
        else:
            logging.info("Step 9: Skipped due to no UserLabel cluster instances")
        # Step 10: Reconfig ACL to allow test runner access to Groups clusters on all endpoints.
        logging.info("Step 10: Reconfiguring ACL to allow access to Groups Clusters")
        await self.send_acl(test_step=10, client_by_name=client_by_name, enable_access_to_group_cluster=True, fabric_table=fabric_table)
        # Step 11: Count all group cluster instances
        # and ensure MaxGroupsPerFabric >= 4 * counted_groups_clusters.
        logging.info("Step 11: Validating groups support minimums")
        groups_cluster_endpoints: Dict[int, Any] = await dev_ctrl.ReadAttribute(self.dut_node_id, [Clusters.Groups])
        counted_groups_clusters: int = len(groups_cluster_endpoints)

        # The test for Step 11 and all of Steps 12 to 15 are only performed if Groups cluster instances are found.
        if counted_groups_clusters > 0:
            indicated_max_groups_per_fabric: int = await self.read_single_attribute(
                dev_ctrl,
                node_id=self.dut_node_id,
                endpoint=0,
                attribute=Clusters.GroupKeyManagement.Attributes.MaxGroupsPerFabric)
            logging.info(
                f"MaxGroupsPerFabric value: {indicated_max_groups_per_fabric}, number of endpoints with Groups clusters cluster: {counted_groups_clusters}, which are: {list(groups_cluster_endpoints.keys())}")
            if indicated_max_groups_per_fabric < 4 * counted_groups_clusters:
                asserts.fail("Failed Step 11: MaxGroupsPerFabric < 4 * counted_groups_clusters")

            # Step 12: Confirm MaxGroupKeysPerFabric meets the minimum requirement of 3.
            indicated_max_group_keys_per_fabric: int = await self.read_single_attribute(
                dev_ctrl,
                node_id=self.dut_node_id,
                endpoint=0,
                attribute=Clusters.GroupKeyManagement.Attributes.MaxGroupKeysPerFabric)
            if indicated_max_group_keys_per_fabric < 3:
                asserts.fail("Failed Step 12: MaxGroupKeysPerFabric < 3")

            # Create a list of per-fabric clients to use for filling group resources accross all fabrics.
            fabric_unique_clients: List[Any] = []

            for fabric in fabric_table:
                client_name = generate_controller_name(fabric.fabricIndex, 0)
                fabric_unique_clients.append(client_by_name[client_name])

            # Step 13: Write and verify indicated_max_group_keys_per_fabric group keys to all fabrics.
            group_keys: List[List[
                Clusters.GroupKeyManagement.Structs.GroupKeySetStruct]] = await self.fill_and_validate_group_key_sets(
                num_fabrics_to_commission, fabric_unique_clients, indicated_max_group_keys_per_fabric)

            # Step 14: Write and verify indicated_max_groups_per_fabric group/key mappings for all fabrics.
            # First, Generate list of unique group/key mappings
            group_key_map: List[Dict[int, int]] = [{} for _ in range(num_fabrics_to_commission)]
            for fabric_list_idx in range(num_fabrics_to_commission):
                for group_idx in range(indicated_max_groups_per_fabric):
                    group_id: int = fabric_list_idx * indicated_max_groups_per_fabric + group_idx + 1
                    group_key_idx: int = group_idx % len(group_keys[fabric_list_idx])
                    group_key_map[fabric_list_idx][group_id] = group_keys[fabric_list_idx][group_key_idx].groupKeySetID

            await self.fill_and_validate_group_key_map(
                num_fabrics_to_commission, fabric_unique_clients, group_key_map, fabric_table)

            # Step 15: Add all the groups to the discovered groups-supporting endpoints and verify GroupTable
            group_table_written: List[
                Dict[int, Clusters.GroupKeyManagement.Structs.GroupInfoMapStruct]] = await self.add_all_groups(
                num_fabrics_to_commission, fabric_unique_clients, group_key_map,
                groups_cluster_endpoints, indicated_max_groups_per_fabric, fabric_table)
            await self.validate_group_table(num_fabrics_to_commission, fabric_unique_clients, group_table_written, fabric_table)

        # Read heap watermarks after the test
        if check_heap_watermarks:
            logging.info("Read Heap info after stress test")
            high_watermark_after, current_usage_after = await self.read_heap_statistics(dev_ctrl)
            logging.info("=== Heap Usage Diagnostics ===\nHigh watermark: {} (before) / {} (after)\n"
                         "Current usage: {} (before) / {} (after)".format(high_watermark_before, high_watermark_after,
                                                                          current_usage_before, current_usage_after))

    def random_string(self, length) -> str:
        rnd = self._pseudo_random_generator
        return "".join([rnd.choice("abcdef0123456789") for _ in range(length)])[:length]

    async def fill_user_label_list(self, dev_ctrl, target_node_id):
        logging.info("Step 9: Fill UserLabel clusters on each endpoint")
        user_labels = await dev_ctrl.ReadAttribute(target_node_id, [Clusters.UserLabel])

        # Build 4 sets of maximized labels
        random_label = self.random_string(16)
        random_value = self.random_string(16)
        labels = [Clusters.UserLabel.Structs.LabelStruct(label=random_label, value=random_value) for _ in range(4)]

        for endpoint_id in user_labels:
            clusters = user_labels[endpoint_id]
            for cluster in clusters:
                if cluster == Clusters.UserLabel:
                    logging.info("Step 9a: Filling UserLabel cluster on endpoint %d" % endpoint_id)
                    statuses = await dev_ctrl.WriteAttribute(target_node_id,
                                                             [(endpoint_id, Clusters.UserLabel.Attributes.LabelList(labels))])
                    asserts.assert_equal(statuses[0].Status, StatusEnum.Success, "Label write must succeed")

                    logging.info("Step 9b: Validate UserLabel cluster contents after write on endpoint %d" % endpoint_id)
                    read_back_labels = await self.read_single_attribute(dev_ctrl,
                                                                        node_id=target_node_id,
                                                                        endpoint=endpoint_id,
                                                                        attribute=Clusters.UserLabel.Attributes.LabelList)
                    print(read_back_labels)

                    asserts.assert_equal(read_back_labels, labels, "LabelList attribute must match what was written")

    async def fill_and_validate_group_key_sets(self,
                                               fabrics: int,
                                               clients: List[Any],
                                               keys_per_fabric: int) -> List[List[
                                                   Clusters.GroupKeyManagement.Structs.GroupKeySetStruct]]:
        # Step 12: Write indicated_max_group_keys_per_fabric group keys to all fabrics.
        group_keys: List[List[Clusters.GroupKeyManagement.Structs.GroupKeySetStruct]] = [[] for _ in range(fabrics)]
        for client_idx in range(fabrics):
            client: Any = clients[client_idx]

            # Write, skip the IPK key set.
            for group_key_cluster_idx in range(1, keys_per_fabric):
                group_key_list_idx: int = group_key_cluster_idx - 1

                logging.info("Step 13: Setting group key on fabric %d at index '%d'" % (client_idx+1, group_key_cluster_idx))
                group_keys[client_idx].append(self.build_group_key(client_idx, group_key_cluster_idx, keys_per_fabric))
                await client.SendCommand(self.dut_node_id, 0, Clusters.GroupKeyManagement.Commands.KeySetWrite(
                    group_keys[client_idx][group_key_list_idx]))

        # Step 12 verification: After all the key sets were written, read all the information back.
        for client_idx in range(fabrics):
            client: Any = clients[client_idx]

            logging.info("Step 13: Reading back group keys on fabric %d" % (client_idx+1))
            resp = await client.SendCommand(self.dut_node_id, 0,
                                            Clusters.GroupKeyManagement.Commands.KeySetReadAllIndices(),
                                            responseType=Clusters.GroupKeyManagement.Commands.KeySetReadAllIndicesResponse)

            read_group_key_ids: List[int] = resp.groupKeySetIDs
            known_group_key_ids: List[int] = [key_set.groupKeySetID for key_set in group_keys[client_idx]]
            ipk_group_key_id: Set[int] = set(read_group_key_ids) - set(known_group_key_ids)

            asserts.assert_equal(keys_per_fabric, len(read_group_key_ids),
                                 "KeySetReadAllIndicesResponse length does "
                                 "not match the key support indicated: %d." % (keys_per_fabric))

            asserts.assert_equal(len(ipk_group_key_id), 1,
                                 "Read more than 1 key ID that did not match written values after IPK (only expected 1 for IPK).")

        return group_keys

    async def fill_and_validate_group_key_map(self,
                                              fabrics: int,
                                              clients: List[Any],
                                              group_key_map: List[Dict[int, int]],
                                              fabric_table: List[
                                                  Clusters.OperationalCredentials.Structs.FabricDescriptorStruct]) -> None:
        # Step 14: Write and verify indicated_max_groups_per_fabric group/key mappings for all fabrics.
        mapping_structs: List[List[Clusters.GroupKeyManagement.Structs.GroupKeyMapStruct]] = [[] for _ in range(fabrics)]
        for client_idx in range(fabrics):
            client: Any = clients[client_idx]
            fabric_idx: int = fabric_table[client_idx].fabricIndex

            for group in group_key_map[client_idx]:
                mapping_structs[client_idx].append(Clusters.GroupKeyManagement.Structs.GroupKeyMapStruct(
                    groupId=group,
                    groupKeySetID=group_key_map[client_idx][group],
                    fabricIndex=fabric_idx))

            logging.info("Step 14: Setting group key map on fabric %d" % (fabric_idx))
            await client.WriteAttribute(
                self.dut_node_id, [(0, Clusters.GroupKeyManagement.Attributes.GroupKeyMap(mapping_structs[client_idx]))])

        # Step 14 verification: After all the group key maps were written, read all the information back.
        for client_idx in range(fabrics):
            client: Any = clients[client_idx]
            fabric_idx: int = fabric_table[client_idx].fabricIndex

            logging.info("Step 14: Reading group key map on fabric %d" % (fabric_idx))
            group_key_map_readback = await self.read_single_attribute(
                client, node_id=self.dut_node_id, endpoint=0, attribute=Clusters.GroupKeyManagement.Attributes.GroupKeyMap)

            found_entry: int = 0
            for read_entry in group_key_map_readback:
                if read_entry.fabricIndex != fabric_idx:
                    continue

                written_entry = next(entry for entry in mapping_structs[client_idx] if entry.groupId == read_entry.groupId)
                found_entry += 1
                asserts.assert_equal(written_entry.groupId, read_entry.groupId)
                asserts.assert_equal(written_entry.groupKeySetID, read_entry.groupKeySetID)
                asserts.assert_equal(written_entry.fabricIndex, read_entry.fabricIndex)

            asserts.assert_equal(found_entry, len(mapping_structs[client_idx]),
                                 "GroupKeyMap does not match the length of written data.")

    async def add_all_groups(self,
                             fabrics: int,
                             clients: List[Any],
                             group_key_map: List[Dict[int, int]],
                             group_endpoints: Dict[int, Any],
                             groups_per_fabric: int,
                             fabric_table: List[
                                 Clusters.OperationalCredentials.Structs.FabricDescriptorStruct]) -> List[
            Dict[int, Clusters.GroupKeyManagement.Structs.GroupInfoMapStruct]]:
        # Step 14: Add indicated_max_groups_per_fabric to each fabric through the Groups clusters on supporting endpoints.
        written_group_table_map: List[Dict[int, Clusters.GroupKeyManagement.Structs.GroupInfoMapStruct]] = [
            {} for _ in range(fabrics)]
        for client_idx in range(fabrics):
            client: Any = clients[client_idx]
            fabric_idx: int = fabric_table[client_idx].fabricIndex

            base_groups_per_endpoint: int = math.floor(groups_per_fabric / len(group_endpoints))
            groups_remainder: int = groups_per_fabric % len(group_endpoints)

            group_id_list_copy = list(group_key_map[client_idx])
            # This is just a sanity check that the number of IDs provided matches how many group IDs we are told to
            # write to each fabric.
            asserts.assert_equal(len(group_id_list_copy), groups_per_fabric)

            for endpoint_id in group_endpoints:
                groups_to_add: int = base_groups_per_endpoint
                if groups_remainder:
                    groups_to_add += 1
                    groups_remainder -= 1

                feature_map: int = await self.read_single_attribute(client,
                                                                    node_id=self.dut_node_id,
                                                                    endpoint=endpoint_id,
                                                                    attribute=Clusters.Groups.Attributes.FeatureMap)
                name_featrure_bit: int = 0
                name_supported: bool = (feature_map & (1 << name_featrure_bit)) != 0

                # Write groups to cluster
                for _ in range(groups_to_add):
                    group_id = group_id_list_copy.pop()
                    group_name: str = self.random_string(16) if name_supported else ""
                    command: Clusters.Groups.Commands.AddGroup = Clusters.Groups.Commands.AddGroup(
                        groupID=group_id, groupName=group_name)
                    written_group_table_map[client_idx][group_id] = Clusters.GroupKeyManagement.Structs.GroupInfoMapStruct(
                        groupId=group_id,
                        groupName=group_name,
                        fabricIndex=fabric_idx,
                        endpoints=[endpoint_id])
                    add_response: Clusters.Groups.Commands.AddGroupResponse = await client.SendCommand(
                        self.dut_node_id, endpoint_id, command, responseType=Clusters.Groups.Commands.AddGroupResponse)
                    asserts.assert_equal(StatusEnum.Success, add_response.status)
                    asserts.assert_equal(group_id, add_response.groupID)

        return written_group_table_map

    async def validate_group_table(self,
                                   fabrics: int,
                                   clients: List[Any],
                                   group_table_written: List[Dict[int, Clusters.GroupKeyManagement.Structs.GroupInfoMapStruct]],
                                   fabric_table: List[Clusters.OperationalCredentials.Structs.FabricDescriptorStruct]) -> None:
        for client_idx in range(fabrics):
            client: Any = clients[client_idx]
            fabric_idx: int = fabric_table[client_idx].fabricIndex

            group_table_read: List[Clusters.GroupKeyManagement.Attributes.GroupTable] = await self.read_single_attribute(
                client, node_id=self.dut_node_id, endpoint=0, attribute=Clusters.GroupKeyManagement.Attributes.GroupTable)

            found_groups: int = 0
            for read_entry in group_table_read:
                if read_entry.fabricIndex != fabric_idx:
                    continue

                found_groups += 1
                asserts.assert_in(read_entry.groupId, group_table_written[client_idx], "Group missing from group map")
                written_entry: Clusters.GroupKeyManagement.Structs.GroupInfoMapStruct = group_table_written[
                    client_idx][read_entry.groupId]
                asserts.assert_equal(written_entry.groupId, read_entry.groupId)
                asserts.assert_equal(written_entry.endpoints, read_entry.endpoints)
                asserts.assert_equal(written_entry.groupName, read_entry.groupName)
                asserts.assert_equal(written_entry.fabricIndex, read_entry.fabricIndex)

            asserts.assert_equal(found_groups, len(group_table_written[client_idx]),
                                 "Found group count does not match written value.")

    async def send_acl(self,
                       test_step: int,
                       client_by_name,
                       enable_access_to_group_cluster: bool,
                       fabric_table: List[
                           Clusters.OperationalCredentials.Structs.FabricDescriptorStruct]):
        for fabric in fabric_table:
            client_name = generate_controller_name(fabric.fabricIndex, 0)
            client = client_by_name[client_name]

            acl = self.build_acl(enable_access_to_group_cluster)

            logging.info(f"Step {test_step}a: Writing ACL entry for fabric {fabric.fabricIndex}")
            await client.WriteAttribute(self.dut_node_id, [(0, Clusters.AccessControl.Attributes.Acl(acl))])

            logging.info(f"Step {test_step}b: Validating ACL entry for fabric {fabric.fabricIndex}")
            acl_readback = await self.read_single_attribute(
                client, node_id=self.dut_node_id, endpoint=0, attribute=Clusters.AccessControl.Attributes.Acl)
            fabric_index = 9999
            for entry in acl_readback:
                asserts.assert_equal(entry.fabricIndex, fabric.fabricIndex, "Fabric Index of response entries must match")
                fabric_index = entry.fabricIndex

            for entry in acl:
                # Fix-up the original ACL list items (that all had fabricIndex of 0 on write, since ignored)
                # so that they match incoming fabric index. Allows checking by equality of the structs
                entry.fabricIndex = fabric_index
            asserts.assert_equal(acl_readback, acl, "ACL must match what was written")

    def build_acl(self, enable_access_to_group_cluster: bool):
        acl = []

        # Test says:
        #
        # . struct
        # - Privilege field: Administer (5)
        # - AuthMode field: CASE (2)
        # - Subjects field: [0xFFFF_FFFD_0001_0001, 0x2000_0000_0000_0001, 0x2000_0000_0000_0002, 0x2000_0000_0000_0003]
        # - Targets field: [
        #                   {Endpoint: 0},
        #                   {Cluster: 0xFFF1_FC00, DeviceType: 0xFFF1_BC30},
        #                   {Cluster: 0xFFF1_FC00, DeviceType: 0xFFF1_BC31}
        #                 ]
        # . struct
        # - Privilege field: Manage (4)
        # - AuthMode field: CASE (2)
        # - Subjects field: [0x1000_0000_0000_0001, 0x1000_0000_0000_0002, 0x1000_0000_0000_0003, 0x1000_0000_0000_0004]
        # - Targets field: [
        #                   {Cluster: 0xFFF1_FC00, DeviceType: 0xFFF1_BC20},
        #                   {Cluster: 0xFFF1_FC01, DeviceType: 0xFFF1_BC21},
        #                   {Cluster: 0xFFF1_FC02, DeviceType: 0xFFF1_BC22}
        #                 ]
        # . struct
        # - Privilege field: Operate (3)
        # - AuthMode field: CASE (2)
        # - Subjects field: [0x3000_0000_0000_0001, 0x3000_0000_0000_0002, 0x3000_0000_0000_0003, 0x3000_0000_0000_0004]
        # - Targets field: [{Cluster: 0xFFF1_FC40, DeviceType: 0xFFF1_BC20},
        #                  {Cluster: 0xFFF1_FC41, DeviceType: 0xFFF1_BC21},
        #                  {Cluster: 0xFFF1_FC02, DeviceType: 0xFFF1_BC42}]
        # . struct
        # - Privilege field: View (1)
        # - AuthMode field: CASE (2)
        # - Subjects field: [0x4000_0000_0000_0001, 0x4000_0000_0000_0002, 0x4000_0000_0000_0003, 0x4000_0000_0000_0004]
        # - Targets field: [{Cluster: 0xFFF1_FC80, DeviceType: 0xFFF1_BC20},
        #                  {Cluster: 0xFFF1_FC81, DeviceType: 0xFFF1_BC21},
        #                  {Cluster: 0xFFF1_FC82, DeviceType: 0xFFF1_BC22}]

        # Administer ACL entry
        admin_subjects = [0xFFFF_FFFD_0001_0001, 0x2000_0000_0000_0001, 0x2000_0000_0000_0002, 0x2000_0000_0000_0003]

        admin_target_field_2 = Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC00, deviceType=0xFFF1_BC30)
        if enable_access_to_group_cluster:
            admin_target_field_2 = Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0x0000_0004)

        admin_targets = [
            Clusters.AccessControl.Structs.AccessControlTargetStruct(endpoint=0),
            admin_target_field_2,
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC01, deviceType=0xFFF1_BC31)
        ]
        admin_acl_entry = Clusters.AccessControl.Structs.AccessControlEntryStruct(
            privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister,
            authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
            subjects=admin_subjects,
            targets=admin_targets
        )
        acl.append(admin_acl_entry)

        # Manage ACL entry
        manage_subjects = [0x1000_0000_0000_0001, 0x1000_0000_0000_0002, 0x1000_0000_0000_0003, 0x1000_0000_0000_0004]
        manage_targets = [
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC00, deviceType=0xFFF1_BC20),
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC01, deviceType=0xFFF1_BC21),
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC02, deviceType=0xFFF1_BC22)
        ]

        manage_acl_entry = Clusters.AccessControl.Structs.AccessControlEntryStruct(
            privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kManage,
            authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
            subjects=manage_subjects,
            targets=manage_targets
        )
        acl.append(manage_acl_entry)

        # Operate ACL entry
        operate_subjects = [0x3000_0000_0000_0001, 0x3000_0000_0000_0002, 0x3000_0000_0000_0003, 0x3000_0000_0000_0004]
        operate_targets = [
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC40, deviceType=0xFFF1_BC20),
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC41, deviceType=0xFFF1_BC21),
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC42, deviceType=0xFFF1_BC42)
        ]

        operate_acl_entry = Clusters.AccessControl.Structs.AccessControlEntryStruct(
            privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kOperate,
            authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
            subjects=operate_subjects,
            targets=operate_targets
        )
        acl.append(operate_acl_entry)

        # View ACL entry
        view_subjects = [0x4000_0000_0000_0001, 0x4000_0000_0000_0002, 0x4000_0000_0000_0003, 0x4000_0000_0000_0004]
        view_targets = [
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC80, deviceType=0xFFF1_BC20),
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC81, deviceType=0xFFF1_BC21),
            Clusters.AccessControl.Structs.AccessControlTargetStruct(cluster=0xFFF1_FC82, deviceType=0xFFF1_BC22)
        ]

        view_acl_entry = Clusters.AccessControl.Structs.AccessControlEntryStruct(
            privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
            authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
            subjects=view_subjects,
            targets=view_targets)
        acl.append(view_acl_entry)

        return acl

    def build_group_key(self, fabric_index: int,
                        group_key_index: int, keys_per_fabric: int) -> Clusters.GroupKeyManagement.Structs.GroupKeySetStruct:
        asserts.assert_not_equal(group_key_index, 0, "TH Internal Error: IPK key set index (0) should not be re-generated.")

        # groupKeySetID is definted as uint16 in the Matter specification.
        # To easily test that the stored values are unique, unique values are created accross all fabrics.
        # However, it is only required that values be unique within a fabric according to the specifiction.
        # If a device ever provides over 65535 total key sets, then this will need to be updated.
        set_id: int = fabric_index*keys_per_fabric + group_key_index
        asserts.assert_less_equal(
            set_id, 0xFFFF, "Invalid Key Set ID. This may be a limitation of the test harness, not the device under test.")
        return Clusters.GroupKeyManagement.Structs.GroupKeySetStruct(
            groupKeySetID=set_id,
            groupKeySecurityPolicy=Clusters.GroupKeyManagement.Enums.GroupKeySecurityPolicyEnum.kTrustFirst,
            epochKey0=self.random_string(16).encode(),
            epochStartTime0=(set_id * 4),
            epochKey1=self.random_string(16).encode(),
            epochStartTime1=(set_id * 4 + 1),
            epochKey2=self.random_string(16).encode(),
            epochStartTime2=(set_id * 4 + 2))

    async def read_heap_statistics(self, dev_ctrl):
        diagnostics_contents = [
            Clusters.SoftwareDiagnostics.Attributes.CurrentHeapHighWatermark,
            Clusters.SoftwareDiagnostics.Attributes.CurrentHeapUsed,
        ]
        diagnostics_paths = [(0, attrib) for attrib in diagnostics_contents]
        swdiag_info = await dev_ctrl.ReadAttribute(self.dut_node_id, diagnostics_paths)

        # Make sure everything came back from the read that we expected
        asserts.assert_true(0 in swdiag_info.keys(), "Must have read endpoint 0 data")
        asserts.assert_true(Clusters.SoftwareDiagnostics in swdiag_info[0].keys(
        ), "Must have read Software Diagnostics cluster data")
        for attribute in diagnostics_contents:
            asserts.assert_true(attribute in swdiag_info[0][Clusters.SoftwareDiagnostics],
                                "Must have read back attribute %s" % (attribute.__name__))
        high_watermark = swdiag_info[0][Clusters.SoftwareDiagnostics][
            Clusters.SoftwareDiagnostics.Attributes.CurrentHeapHighWatermark]
        current_usage = swdiag_info[0][Clusters.SoftwareDiagnostics][
            Clusters.SoftwareDiagnostics.Attributes.CurrentHeapUsed]
        return high_watermark, current_usage


if __name__ == "__main__":
    default_matter_test_main(maximize_cert_chains=True, controller_cat_tags=[0x0001_0001])