TC_IDM_1_4.py

#
#    Copyright (c) 2023 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.
#

# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
# for details about the block below.
#
# === BEGIN CI TEST ARGUMENTS ===
# test-runner-runs: run1
# test-runner-run/run1/app: ${ALL_CLUSTERS_APP}
# test-runner-run/run1/factoryreset: True
# test-runner-run/run1/quiet: True
# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json --enable-key 000102030405060708090a0b0c0d0e0f
# test-runner-run/run1/script-args: --hex-arg PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
# === END CI TEST ARGUMENTS ===

import logging

import chip.clusters as Clusters
from chip.exceptions import ChipStackError
from chip.interaction_model import InteractionModelError, Status
from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches
from mobly import asserts

# If DUT supports `MaxPathsPerInvoke > 1`, additional command line argument
# run with
# --hex-arg PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY:<key>


class TC_IDM_1_4(MatterBaseTest):

    def steps_TC_IDM_1_4(self) -> list[TestStep]:
        steps = [TestStep(1, "Get remote node's MaxPathsPerInvoke", is_commissioning=True),
                 TestStep(2, "Sending `MaxPathsPerInvoke + 1` InvokeRequest if it fits into single MTU"),
                 TestStep(3, "Sending two InvokeRequests with identical paths"),
                 TestStep(4, "Sending two InvokeRequests with unique paths, but identical CommandRefs"),
                 TestStep(5, "Verify DUT responds to InvokeRequestMessage containing two valid paths"),
                 TestStep(6, "Verify DUT responds to InvokeRequestMessage containing one valid paths, and one InvokeRequest to unsupported endpoint"),
                 TestStep(7, "Verify DUT responds to InvokeRequestMessage containing two valid paths. One of which requires timed invoke, and TimedRequest in InvokeResponseMessage set to true, but never sending preceding Timed Invoke Action"),
                 TestStep(8, "Verify DUT responds to InvokeRequestMessage containing two valid paths. One of which requires timed invoke, and TimedRequest in InvokeResponseMessage set to true"),
                 TestStep(9, "Verify DUT supports extended Data Model Testing feature in General Diagnostics Cluster"),
                 TestStep(10, "Verify DUT has TestEventTriggersEnabled attribute set to true in General Diagnostics Cluster"),
                 TestStep(11, "Verify DUT capable of responding to request with multiple InvokeResponseMessages")]
        return steps

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

        self.print_step(0, "Commissioning - already done")

        self.step(1)
        session_parameters = dev_ctrl.GetRemoteSessionParameters(dut_node_id)
        max_paths_per_invoke = session_parameters.maxPathsPerInvoke

        asserts.assert_greater_equal(max_paths_per_invoke, 1, "Max Paths per Invoke less than spec min")
        asserts.assert_less_equal(max_paths_per_invoke, 65535, "Max Paths per Invoke greater than spec max")

        self.step(2)
        # In practice, it was noticed that we could only fit 57 commands before we hit the MTU limit as a result we
        # conservatively try putting up to 100 commands into an Invoke request. We are expecting one of 2 things to
        # happen if max_paths_per_invoke + 1 is greater than what cap_for_batch_commands is set to:
        # 1. Client (TH) fails to send command, since we cannot fit all the commands single MTU.
        #    When this happens we get ChipStackError with CHIP_ERROR_NO_MEMORY. Test step is skipped
        #    as per test spec instructions
        # 2. Client (TH) able to send command. While unexpected we will hit two different test failure depending on
        #    what the server does.
        #    a. Server successfully handle command and send InvokeResponse with results of all individual commands
        #       being failure. In this case, test already will fail on unexpected successes.
        #    b. Server fails to handle command that is between cap_for_batch_commands and max_paths_per_invoke + 1.
        #       In this case, test fails as device should have actually succeeded and been caught in 2.a.
        cap_for_batch_commands = 100
        number_of_commands_to_send = min(max_paths_per_invoke + 1, cap_for_batch_commands)

        # Use a valid (according to MEI definition) command-id (in this case belonging to Test Vendor MC with prefix 0xfff4)
        # which should never be existing on a prodiction device. We use this decodable id to prevent hitting issues with the
        # specification being not clearly defined in respect to decoding vs processing the invoke requests.
        invalid_command_id = 0xfff4_00ff

        list_of_commands_to_send = []
        for endpoint_index in range(number_of_commands_to_send):
            # Using Toggle command to form the base as it is a command that doesn't take
            # any arguments, this allows us to fit as more requests into single MTU.
            invalid_command = Clusters.OnOff.Commands.Toggle()
            # This is how we make the command invalid
            invalid_command.command_id = invalid_command_id

            list_of_commands_to_send.append(Clusters.Command.InvokeRequestInfo(endpoint_index, invalid_command))

        asserts.assert_greater_equal(len(list_of_commands_to_send), 2,
                                     "Step 2 is always expected to try sending at least 2 command, something wrong with test logic")
        try:
            await dev_ctrl.TestOnlySendBatchCommands(dut_node_id, list_of_commands_to_send, remoteMaxPathsPerInvoke=number_of_commands_to_send)
            # This might happen after TCP is enabled and DUT supports TCP. See comment above `cap_for_batch_commands`
            # for more information.
            asserts.assert_not_equal(number_of_commands_to_send, cap_for_batch_commands,
                                     "Test needs to be updated! Soft cap `cap_for_batch_commands` used in test is no longer correct")
            asserts.fail(
                f"Unexpected success return from sending too many commands, we sent {number_of_commands_to_send}, test capped at {cap_for_batch_commands}")
        except InteractionModelError as e:
            # This check is for 2.b, mentioned above. If this assert occurs, test likely needs updating. Although DUT
            # is still going to fail since it seemingly is failing to process a smaller number then it is reporting
            # that it is capable of processing.
            asserts.assert_equal(number_of_commands_to_send, max_paths_per_invoke + 1,
                                 "Test didn't send as many command as max_paths_per_invoke + 1, likely due to MTU cap_for_batch_commands, but we still got an error from server. This should have been a success from server")
            asserts.assert_equal(e.status, Status.InvalidAction,
                                 "DUT sent back an unexpected error, we were expecting InvalidAction")
            logging.info("DUT successfully failed to process `MaxPathsPerInvoke + 1` InvokeRequests")
        except ChipStackError as e:
            chip_error_no_memory = 0x0b
            asserts.assert_equal(e.err, chip_error_no_memory, "Unexpected error while trying to send InvokeRequest")
            # TODO it is possible we want to confirm DUT can handle up to MTU max. But that is not in test plan as of right now.
            # Additionally CommandSender is not currently set up to enable caller to fill up to MTU. This might be coming soon,
            # just that it is not supported today.
            logging.info("DUTs reported MaxPathsPerInvoke + 1 is larger than what fits into MTU. Test step is skipped")

        if max_paths_per_invoke == 1:
            self.skip_all_remaining_steps(3)
        else:
            asserts.assert_true('PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY' in self.matter_test_config.global_test_params,
                                "PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY must be included on the command line in "
                                "the --hex-arg flag as PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY:<key>, "
                                "e.g. --hex-arg PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f")

            await self.remaining_batch_commands_test_steps(False)

    async def remaining_batch_commands_test_steps(self, dummy_value):
        dev_ctrl = self.default_controller
        dut_node_id = self.dut_node_id

        self.step(3)
        command = Clusters.BasicInformation.Commands.MfgSpecificPing()
        endpoint = 0
        invoke_request_1 = Clusters.Command.InvokeRequestInfo(endpoint, command)
        try:
            result = await dev_ctrl.SendBatchCommands(dut_node_id, [invoke_request_1, invoke_request_1])
            asserts.fail("Unexpected success return after sending two identical (non-unique) paths in the InvokeRequest")
        except InteractionModelError as e:
            asserts.assert_equal(e.status, Status.InvalidAction,
                                 "DUT sent back an unexpected error, we were expecting InvalidAction")
            logging.info("DUT successfully failed to process two InvokeRequests that contains non-unique paths")

        self.step(4)
        endpoint = 0
        command = Clusters.OperationalCredentials.Commands.CertificateChainRequest(
            Clusters.OperationalCredentials.Enums.CertificateChainTypeEnum.kDACCertificate)
        invoke_request_1 = Clusters.Command.InvokeRequestInfo(endpoint, command)

        command = Clusters.GroupKeyManagement.Commands.KeySetRead(0)
        invoke_request_2 = Clusters.Command.InvokeRequestInfo(endpoint, command)
        commandRefsOverride = [1, 1]
        try:
            await dev_ctrl.TestOnlySendBatchCommands(dut_node_id, [invoke_request_1, invoke_request_2], commandRefsOverride=commandRefsOverride)
            asserts.fail("Unexpected success return after sending two unique commands with identical CommandRef in the InvokeRequest")
        except InteractionModelError as e:
            asserts.assert_equal(e.status, Status.InvalidAction,
                                 "DUT sent back an unexpected error, we were expecting InvalidAction")
            logging.info("DUT successfully failed to process two InvokeRequests that contains non-unique CommandRef")

        self.step(5)
        endpoint = 0
        command = Clusters.OperationalCredentials.Commands.CertificateChainRequest(
            Clusters.OperationalCredentials.Enums.CertificateChainTypeEnum.kDACCertificate)
        invoke_request_1 = Clusters.Command.InvokeRequestInfo(endpoint, command)

        command = Clusters.GroupKeyManagement.Commands.KeySetRead(0)
        invoke_request_2 = Clusters.Command.InvokeRequestInfo(endpoint, command)
        try:
            result = await dev_ctrl.SendBatchCommands(dut_node_id, [invoke_request_1, invoke_request_2])
            asserts.assert_true(type_matches(result, list), "Unexpected return from SendBatchCommands")
            asserts.assert_equal(len(result), 2, "Unexpected number of InvokeResponses sent back from DUT")
            asserts.assert_true(type_matches(
                result[0], Clusters.OperationalCredentials.Commands.CertificateChainResponse), "Unexpected return type for first InvokeRequest")
            asserts.assert_true(type_matches(
                result[1], Clusters.GroupKeyManagement.Commands.KeySetReadResponse), "Unexpected return type for second InvokeRequest")
            logging.info("DUT successfully responded to a InvokeRequest action with two valid commands")
        except InteractionModelError:
            asserts.fail("DUT failed to successfully responded to a InvokeRequest action with two valid commands")

        self.step(6)
        # First finding non-existent endpoint
        wildcard_descriptor = await dev_ctrl.ReadAttribute(dut_node_id, [(Clusters.Descriptor)])
        endpoints = list(wildcard_descriptor.keys())
        endpoints.sort()
        non_existent_endpoint = next(i for i, e in enumerate(endpoints + [None]) if i != e)

        endpoint = 0
        command = Clusters.OperationalCredentials.Commands.CertificateChainRequest(
            Clusters.OperationalCredentials.Enums.CertificateChainTypeEnum.kDACCertificate)
        invoke_request_1 = Clusters.Command.InvokeRequestInfo(endpoint, command)

        endpoint = non_existent_endpoint
        command = Clusters.GroupKeyManagement.Commands.KeySetRead(0)
        invoke_request_2 = Clusters.Command.InvokeRequestInfo(endpoint, command)
        try:
            result = await dev_ctrl.SendBatchCommands(dut_node_id, [invoke_request_1, invoke_request_2])
            asserts.assert_true(type_matches(result, list), "Unexpected return from SendBatchCommands")
            asserts.assert_equal(len(result), 2, "Unexpected number of InvokeResponses sent back from DUT")
            asserts.assert_true(type_matches(
                result[0], Clusters.OperationalCredentials.Commands.CertificateChainResponse), "Unexpected return type for first InvokeRequest")
            asserts.assert_true(type_matches(
                result[1], InteractionModelError), "Unexpected return type for second InvokeRequest")
            asserts.assert_equal(result[1].status, Status.UnsupportedEndpoint,
                                 "Unexpected Interaction model error, was expecting UnsupportedEndpoint")
            logging.info(
                "DUT successfully responded to first valid InvokeRequest, and successfully errored with UnsupportedEndpoint for the second")
        except InteractionModelError:
            asserts.fail("DUT failed to successfully responded to a InvokeRequest action with two valid commands")

        self.step(7)
        endpoint = 0
        command = Clusters.GroupKeyManagement.Commands.KeySetRead(0)
        invoke_request_1 = Clusters.Command.InvokeRequestInfo(endpoint, command)

        command = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
        invoke_request_2 = Clusters.Command.InvokeRequestInfo(endpoint, command)
        # It is safe to use RevokeCommissioning in this test without opening the commissioning window because
        # we expect a non-path-specific error. As per the specification, non-path-specific errors of this
        # nature is generated before command dispatch to cluster. In the next test step, we anticipate
        # receiving a path-specific response to the same command, with the TimedRequestMessage sent before
        # the InvokeRequestMessage.
        try:
            await dev_ctrl.TestOnlySendBatchCommands(dut_node_id, [invoke_request_1, invoke_request_2], suppressTimedRequestMessage=True)
            asserts.fail("Unexpected success call to sending Batch command when non-path specific error expected")
        except InteractionModelError as e:
            asserts.assert_equal(e.status, Status.TimedRequestMismatch,
                                 "Unexpected error response from Invoke with TimedRequest flag and no TimedInvoke")

        self.step(8)
        endpoint = 0
        command = Clusters.GroupKeyManagement.Commands.KeySetRead(0)
        invoke_request_1 = Clusters.Command.InvokeRequestInfo(endpoint, command)

        command = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
        invoke_request_2 = Clusters.Command.InvokeRequestInfo(endpoint, command)
        try:
            result = await dev_ctrl.SendBatchCommands(dut_node_id, [invoke_request_1, invoke_request_2], timedRequestTimeoutMs=5000)
            asserts.assert_true(type_matches(result, list), "Unexpected return from SendBatchCommands")
            asserts.assert_equal(len(result), 2, "Unexpected number of InvokeResponses sent back from DUT")
            asserts.assert_true(type_matches(
                result[0], Clusters.GroupKeyManagement.Commands.KeySetReadResponse), "Unexpected return type for first InvokeRequest")
            asserts.assert_true(type_matches(result[1], InteractionModelError), "Unexpected return type for second InvokeRequest")

            # We sent out RevokeCommissioning without an ArmSafe intentionally, confirm that it failed for that reason.
            asserts.assert_equal(result[1].status, Status.Failure,
                                 "Timed command, RevokeCommissioning, didn't fail in manner expected by test")
            window_not_open_cluster_error = 4
            asserts.assert_equal(result[1].clusterStatus, window_not_open_cluster_error,
                                 "Timed command, RevokeCommissioning, failed with incorrect cluster code")
            logging.info("DUT successfully responded to a InvokeRequest action with two valid commands. One of which required timed invoke, and TimedRequest in InvokeResponseMessage was set to true")
        except InteractionModelError:
            asserts.fail("DUT failed with non-path specific error when path specific error was expected")

        self.step(9)
        try:
            feature_map = await self.read_single_attribute(
                dev_ctrl,
                dut_node_id,
                endpoint=0,
                attribute=Clusters.GeneralDiagnostics.Attributes.FeatureMap
            )
        except InteractionModelError:
            asserts.fail("DUT failed to respond reading FeatureMap attribute")
        has_data_model_test_feature = (feature_map & Clusters.GeneralDiagnostics.Bitmaps.Feature.kDataModelTest) != 0
        asserts.assert_true(has_data_model_test_feature, "DataModelTest Feature is not supported by DUT")

        self.step(10)
        try:
            test_event_triggers_enabled = await self.read_single_attribute(
                dev_ctrl,
                dut_node_id,
                endpoint=0,
                attribute=Clusters.GeneralDiagnostics.Attributes.TestEventTriggersEnabled
            )
        except InteractionModelError:
            asserts.fail("DUT failed to respond reading TestEventTriggersEnabled attribute")
        asserts.assert_true(test_event_triggers_enabled, "Test Event Triggers must be enabled on DUT")

        self.step(11)
        enable_key = self.matter_test_config.global_test_params['PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY']
        endpoint = 0
        command = Clusters.GeneralDiagnostics.Commands.PayloadTestRequest(
            enableKey=enable_key,
            value=ord('A'),
            count=800
        )
        invoke_request_1 = Clusters.Command.InvokeRequestInfo(endpoint, command)

        command = Clusters.OperationalCredentials.Commands.CertificateChainRequest(
            Clusters.OperationalCredentials.Enums.CertificateChainTypeEnum.kDACCertificate)
        invoke_request_2 = Clusters.Command.InvokeRequestInfo(endpoint, command)

        try:
            test_only_result = await dev_ctrl.TestOnlySendBatchCommands(dut_node_id, [invoke_request_1, invoke_request_2])
        except InteractionModelError:
            asserts.fail("DUT failed to respond to batch commands, where response is expected to be too large to fit in a single ResponseMessage")

        responses = test_only_result.Responses
        # This check is validating the number of InvokeResponses we got
        asserts.assert_equal(len(responses), 2, "Unexpected number of InvokeResponses sent back from DUT")
        asserts.assert_true(type_matches(
            responses[0], Clusters.GeneralDiagnostics.Commands.PayloadTestResponse), "Unexpected return type for first InvokeRequest")
        asserts.assert_true(type_matches(
            responses[1], Clusters.OperationalCredentials.Commands.CertificateChainResponse), "Unexpected return type for second InvokeRequest")
        logging.info("DUT successfully responded to a InvokeRequest action with two valid commands")

        asserts.assert_equal(responses[0].payload, b'A' * 800, "Expect response to match for count == 800")
        # If this assert below fails then some assumptions we were relying on are now no longer true.
        # This check is validating the number of InvokeResponsesMessages we got. This is different then the earlier
        # `len(responses)` check as you can have multiple InvokeResponses in a single message. But this test step
        # is explicitly making sure that we recieved multiple ResponseMessages.
        asserts.assert_greater_equal(test_only_result.ResponseMessageCount, 2,
                                     "DUT was expected to send multiple InvokeResponseMessages")


if __name__ == "__main__":
    default_matter_test_main()