drlk_2_x_common.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.
#

import logging
import random
import string
import time

import chip.clusters as Clusters
from chip.clusters.Types import NullValue
from chip.interaction_model import InteractionModelError, Status
from matter_testing_support import type_matches
from mobly import asserts


class DRLK_COMMON:
    async def read_drlk_attribute_expect_success(self, attribute):
        cluster = Clusters.Objects.DoorLock
        return await self.read_single_attribute_check_success(endpoint=self.endpoint, cluster=cluster, attribute=attribute)

    async def write_drlk_attribute_expect_success(self, attribute):
        cluster = Clusters.Objects.DoorLock
        result = await self.default_controller.WriteAttribute(self.dut_node_id, [(self.endpoint, attribute)])
        err_msg = "Received error status {} when writing {}:{}".format(str(result[0].Status), str(cluster), str(attribute))
        asserts.assert_equal(result[0].Status, Status.Success, err_msg)

    async def write_drlk_attribute_expect_error(self, attribute, error):
        cluster = Clusters.Objects.DoorLock
        result = await self.default_controller.WriteAttribute(self.dut_node_id, [(self.endpoint, attribute)])
        err_msg = "Did not see expected error {} when writing {}:{}".format(str(error), str(cluster), str(attribute))
        asserts.assert_equal(result[0].Status, error, err_msg)

    async def send_drlk_cmd_expect_success(self, command) -> None:
        await self.send_single_cmd(cmd=command, endpoint=self.endpoint, timedRequestTimeoutMs=1000)

    async def send_drlk_cmd_expect_error(self, command, error: Status) -> None:
        try:
            await self.send_single_cmd(cmd=command, endpoint=self.endpoint, timedRequestTimeoutMs=1000)
            asserts.assert_true(False, "Unexpected command success, command=%s", command)
        except InteractionModelError as e:
            asserts.assert_equal(e.status, error, "Unexpected error returned")
            pass

    async def send_set_credential_cmd(self, operationType, credential, credentialData, userIndex, userStatus, userType) -> Clusters.Objects.DoorLock.Commands.SetCredentialResponse:
        ret = await self.send_single_cmd(cmd=Clusters.Objects.DoorLock.Commands.SetCredential(operationType=operationType,
                                                                                              credential=credential,
                                                                                              credentialData=credentialData,
                                                                                              userIndex=userIndex,
                                                                                              userStatus=userStatus,
                                                                                              userType=userType),
                                         endpoint=self.endpoint,
                                         timedRequestTimeoutMs=1000)
        asserts.assert_true(type_matches(ret, Clusters.Objects.DoorLock.Commands.SetCredentialResponse),
                            "Unexpected return type for SetCredential")
        asserts.assert_true(ret.status == Status.Success, "Error sending SetCredential command, status={}".format(str(ret.status)))
        return ret

    async def send_clear_credential_cmd(self, credential) -> None:
        await self.send_single_cmd(cmd=Clusters.Objects.DoorLock.Commands.ClearCredential(credential=credential),
                                   endpoint=self.endpoint,
                                   timedRequestTimeoutMs=1000)
        ret = await self.send_single_cmd(cmd=Clusters.Objects.DoorLock.Commands.GetCredentialStatus(credential=self.createdCredential),
                                         endpoint=self.endpoint)
        asserts.assert_true(type_matches(ret, Clusters.Objects.DoorLock.Commands.GetCredentialStatusResponse),
                            "Unexpected return type for GetCredentialStatus")
        asserts.assert_false(ret.credentialExists, "Error clearing Credential (credentialExists==True)")

    async def cleanup_users_and_credentials(self):
        self.print_step("Cleanup", "Clear created User and Credential on the DUT")
        if self.createdCredential:
            await self.send_clear_credential_cmd(self.createdCredential)
            logging.info("Credential cleared at CredentialIndex %d" % (self.createdCredential.credentialIndex))
        self.createdCredential = None

    async def generate_pincode(self, maxPincodeLength):
        return ''.join(random.choices(string.digits, k=maxPincodeLength))

    async def teardown(self):
        await self.cleanup_users_and_credentials()

    async def run_drlk_test_common(self, lockUnlockCommand, lockUnlockCmdRspPICS, lockUnlockText, doAutoRelockTest):
        is_ci = self.check_pics('PICS_SDK_CI_ONLY')

        self.createdCredential = None

        self.endpoint = self.user_params.get("endpoint", 1)

        # Allow for user overrides of these values
        credentialIndex = self.user_params.get("credential_index", 1)
        userCodeTemporaryDisableTime = self.user_params.get("user_code_temporary_disable_time", 15)
        wrongCodeEntryLimit = self.user_params.get("wrong_code_entry_limit", 3)
        autoRelockTime = self.user_params.get("auto_relock_time", 60)
        if is_ci:
            autoRelockTime = 10

        cluster = Clusters.Objects.DoorLock
        attributes = Clusters.DoorLock.Attributes

        validPincode = None
        invalidPincode = None

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

        if self.check_pics("DRLK.S.F00") and self.check_pics("DRLK.S.F07"):
            self.print_step("Preconditions.1a",
                            "TH reads MaxPINCodeLength attribute from DUT and generates a valid PINCode")
            maxPincodeLength_dut = await self.read_drlk_attribute_expect_success(attribute=attributes.MaxPINCodeLength)
            logging.info("MaxPINCodeLength value is %s" % (maxPincodeLength_dut))

            validPincodeString = await self.generate_pincode(maxPincodeLength_dut)
            while True:
                invalidPincodeString = await self.generate_pincode(maxPincodeLength_dut)
                if invalidPincodeString != validPincodeString:
                    break
            logging.info("Valid PinCode=%s, Invalid PinCode=%s" % (validPincodeString, invalidPincodeString))

            validPincode = bytes(validPincodeString, 'ascii')
            invalidPincode = bytes(invalidPincodeString, 'ascii')

            self.print_step("Preconditions.1b",
                            "TH sends SetCredential command to DUT to set up User and Credential at CredentialIndex {}".format(str(credentialIndex)))
            credential = cluster.Structs.CredentialStruct(credentialIndex=credentialIndex,
                                                          credentialType=Clusters.DoorLock.Enums.CredentialTypeEnum.kPin)
            ret = await self.send_set_credential_cmd(Clusters.DoorLock.Enums.DataOperationTypeEnum.kAdd,
                                                     credential,
                                                     validPincode,
                                                     NullValue,
                                                     NullValue,
                                                     NullValue)
            logging.info("Credential created at CredentialIndex %d, UserIndex %d." % (credentialIndex, ret.userIndex))
            self.createdCredential = credential

        requirePinForRemoteOperation_dut = False
        if self.check_pics("DRLK.S.F00") and self.check_pics("DRLK.S.F07"):
            self.print_step("1", "TH writes the RequirePINforRemoteOperation attribute value as false on the DUT")
            attribute = attributes.RequirePINforRemoteOperation(False)
            if self.check_pics("DRLK.S.M.RequirePINForRemoteOperationAttributeWritable"):
                print("---------------------- PICS is true")
                await self.write_drlk_attribute_expect_success(attribute=attribute)
            else:
                print("---------------------- PICS is false")
                await self.write_drlk_attribute_expect_error(attribute=attribute, error=Status.UnsupportedWrite)

            if self.check_pics("DRLK.S.A0033"):
                self.print_step("2", "TH reads and saves the value of the RequirePINforRemoteOperation attribute from the DUT")
                requirePinForRemoteOperation_dut = await self.read_drlk_attribute_expect_success(attribute=attributes.RequirePINforRemoteOperation)
                logging.info("Current RequirePINforRemoteOperation value is %s" % (requirePinForRemoteOperation_dut))

                if self.check_pics("DRLK.S.M.RequirePINForRemoteOperationAttributeWritable"):
                    self.print_step("2a", "TH verifies that RequirePINforRemoteOperation is FALSE")
                    asserts.assert_false(requirePinForRemoteOperation_dut, "RequirePINforRemoteOperation is expected to be FALSE")
            else:
                asserts.assert_true(
                    False, "RequirePINforRemoteOperation is a mandatory attribute if DRLK.S.F07(COTA) & DRLK.S.F00(PIN)")

        if self.check_pics(lockUnlockCmdRspPICS):
            self.print_step("3", "TH sends %s Command to the DUT without PINCode" % lockUnlockText)
            command = lockUnlockCommand(PINCode=None)
            if requirePinForRemoteOperation_dut:
                await self.send_drlk_cmd_expect_error(command=command, error=Status.Failure)
            else:
                await self.send_drlk_cmd_expect_success(command=command)

            self.print_step("4", "TH sends %s Command to the DUT with valid PINCode" % lockUnlockText)
            command = lockUnlockCommand(PINCode=validPincode)
            await self.send_drlk_cmd_expect_success(command=command)
        else:
            asserts.assert_true(False, "%sResponse is a mandatory command response and must be supported in PICS" % lockUnlockText)

        if self.check_pics("DRLK.S.F00") and self.check_pics("DRLK.S.F07"):
            self.print_step("5", "TH writes the RequirePINforRemoteOperation attribute value as true on the DUT")
            attribute = attributes.RequirePINforRemoteOperation(True)
            if self.check_pics("DRLK.S.M.RequirePINForRemoteOperationAttributeWritable"):
                await self.write_drlk_attribute_expect_success(attribute=attribute)
            else:
                await self.write_drlk_attribute_expect_error(attribute=attribute, error=Status.UnsupportedWrite)

            if self.check_pics("DRLK.S.A0033"):
                self.print_step("6", "TH reads and saves the value of the RequirePINforRemoteOperation attribute from the DUT")
                requirePinForRemoteOperation_dut = await self.read_drlk_attribute_expect_success(attribute=attributes.RequirePINforRemoteOperation)
                logging.info("Current RequirePINforRemoteOperation value is %s" % (requirePinForRemoteOperation_dut))

                if self.check_pics("DRLK.S.M.RequirePINForRemoteOperationAttributeWritable"):
                    self.print_step("6a", "TH verifies that RequirePINforRemoteOperation is TRUE")
                    asserts.assert_true(requirePinForRemoteOperation_dut, "RequirePINforRemoteOperation is expected to be TRUE")

        if self.check_pics("DRLK.S.F00") and self.check_pics(lockUnlockCmdRspPICS) and self.check_pics("DRLK.S.A0033"):
            self.print_step("7", "TH sends %s Command to the DUT with an invalid PINCode" % lockUnlockText)
            command = lockUnlockCommand(PINCode=invalidPincode)
            await self.send_drlk_cmd_expect_error(command=command, error=Status.Failure)

            self.print_step("8", "TH sends %s Command to the DUT without PINCode" % lockUnlockText)
            command = lockUnlockCommand(PINCode=None)
            if requirePinForRemoteOperation_dut:
                await self.send_drlk_cmd_expect_error(command=command, error=Status.Failure)
            else:
                await self.send_drlk_cmd_expect_success(command=command)

        if self.check_pics(lockUnlockCmdRspPICS) and self.check_pics("DRLK.S.A0033"):
            self.print_step("9", "TH sends %s Command to the DUT with valid PINCode" % lockUnlockText)
            command = lockUnlockCommand(PINCode=validPincode)
            await self.send_drlk_cmd_expect_success(command=command)

        if self.check_pics("DRLK.S.F00") or self.check_pics("DRLK.S.F01"):
            self.print_step("10a", "TH writes the WrongCodeEntryLimit to any value between 1 and 255")
            attribute = attributes.WrongCodeEntryLimit(wrongCodeEntryLimit)
            if self.check_pics("DRLK.S.M.WrongCodeEntryLimitAttributeWritable"):
                await self.write_drlk_attribute_expect_success(attribute=attribute)
            else:
                await self.write_drlk_attribute_expect_error(attribute=attribute, error=Status.UnsupportedWrite)

            self.print_step("10b", "TH reads the value of WrongCodeEntryLimit attribute. Verify a range of 1-255")
            wrongCodeEntryLimit_dut = await self.read_drlk_attribute_expect_success(attribute=attributes.WrongCodeEntryLimit)
            logging.info("WrongCodeEntryLimit value is %s" % (wrongCodeEntryLimit_dut))
            asserts.assert_in(wrongCodeEntryLimit_dut, range(1, 255), "WrongCodeEntryLimit value is out of range")

            self.print_step("11a", "TH writes the UserCodeTemporaryDisableTime to any value between 1 and 255")
            attribute = attributes.UserCodeTemporaryDisableTime(userCodeTemporaryDisableTime)
            if self.check_pics("DRLK.S.M.UserCodedTemporaryDisableTimeAttributeWritable"):
                await self.write_drlk_attribute_expect_success(attribute=attribute)
            else:
                await self.write_drlk_attribute_expect_error(attribute=attribute, error=Status.UnsupportedWrite)

            self.print_step("11b", "TH reads the value of UserCodeTemporaryDisableTime attribute. Verify a range of 1-255")
            userCodeTemporaryDisableTime_dut = await self.read_drlk_attribute_expect_success(attribute=attributes.UserCodeTemporaryDisableTime)
            logging.info("UserCodeTemporaryDisableTime value is %s" % (userCodeTemporaryDisableTime_dut))
            asserts.assert_in(userCodeTemporaryDisableTime_dut, range(1, 255), "UserCodeTemporaryDisableTime value is out of range")

        if self.check_pics(lockUnlockCmdRspPICS) and self.check_pics("DRLK.S.F00"):
            self.print_step("12", "TH sends {} Command to the DUT with an invalid PINCode, repeated {} times".format(
                lockUnlockText, wrongCodeEntryLimit_dut))
            for i in range(wrongCodeEntryLimit_dut):
                command = lockUnlockCommand(PINCode=invalidPincode)
                await self.send_drlk_cmd_expect_error(command=command, error=Status.Failure)

            self.print_step("13", "TH sends %s Command to the DUT with valid PINCode. Verify failure or no response" % lockUnlockText)
            command = lockUnlockCommand(PINCode=validPincode)
            await self.send_drlk_cmd_expect_error(command=command, error=Status.Failure)

            if self.check_pics("DRLK.S.A0031"):
                self.print_step("14", "Wait for UserCodeTemporaryDisableTime seconds")
                time.sleep(userCodeTemporaryDisableTime_dut)

            if not doAutoRelockTest:
                self.print_step("15", "Send %s with valid Pincode and verify success" % lockUnlockText)
                command = lockUnlockCommand(PINCode=validPincode)
                await self.send_drlk_cmd_expect_success(command=command)

        if doAutoRelockTest:
            if self.check_pics("DRLK.S.A0023"):
                self.print_step("15", "TH writes the AutoRelockTime attribute value on the DUT")
                attribute = attributes.AutoRelockTime(autoRelockTime)
                if self.check_pics("DRLK.S.M.AutoRelockTimeAttributeWritable"):
                    await self.write_drlk_attribute_expect_success(attribute=attribute)
                else:
                    await self.write_drlk_attribute_expect_error(attribute=attribute, error=Status.UnsupportedWrite)

                self.print_step("16", "TH reads the value of AutoRelockTime attribute.")
                autoRelockTime_dut = await self.read_drlk_attribute_expect_success(attribute=attributes.AutoRelockTime)
                logging.info("AutoRelockTime value is %s" % (autoRelockTime_dut))

            if self.check_pics(lockUnlockCmdRspPICS):
                self.print_step("17", "Send %s with valid Pincode and verify success" % lockUnlockText)
                command = lockUnlockCommand(PINCode=validPincode)
                await self.send_drlk_cmd_expect_success(command=command)

            if self.check_pics("DRLK.S.A0023"):
                self.print_step("18a", "Wait for AutoRelockTime seconds")
                # Add additional wait time buffer for motor movement, etc.
                time.sleep(autoRelockTime_dut + 5)

                if self.check_pics("DRLK.S.A0000"):
                    self.print_step("18b", "TH reads LockState attribute after AutoRelockTime Expires")
                    lockstate_dut = await self.read_drlk_attribute_expect_success(attribute=attributes.LockState)
                    logging.info("Current LockState is %s" % (lockstate_dut))
                    asserts.assert_equal(lockstate_dut, Clusters.DoorLock.Enums.DlLockState.kLocked,
                                         "LockState expected to be value==Locked")

        await self.cleanup_users_and_credentials()

    async def run_drlk_test_2_2(self):
        await self.run_drlk_test_common(lockUnlockCommand=Clusters.Objects.DoorLock.Commands.LockDoor,
                                        lockUnlockCmdRspPICS="DRLK.S.C00.Rsp",
                                        lockUnlockText="LockDoor",
                                        doAutoRelockTest=False)

    async def run_drlk_test_2_3(self):
        await self.run_drlk_test_common(lockUnlockCommand=Clusters.Objects.DoorLock.Commands.UnlockDoor,
                                        lockUnlockCmdRspPICS="DRLK.S.C01.Rsp",
                                        lockUnlockText="UnlockDoor",
                                        doAutoRelockTest=True)

    async def run_drlk_test_2_12(self):
        await self.run_drlk_test_common(lockUnlockCommand=Clusters.Objects.DoorLock.Commands.UnboltDoor,
                                        lockUnlockCmdRspPICS="DRLK.S.C27.Rsp",
                                        lockUnlockText="UnboltDoor",
                                        doAutoRelockTest=True)