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

# test-runner-runs: run1
# test-runner-run/run1/app: ${CHIP_RVC_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
# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values --endpoint 1 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto

import logging
from time import sleep

import chip.clusters as Clusters
from chip.clusters.Types import NullValue
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, type_matches
from mobly import asserts


# Takes an OpState or RvcOpState state enum and returns a string representation
def state_enum_to_text(state_enum):
    if state_enum == Clusters.OperationalState.Enums.OperationalStateEnum.kStopped:
        return "Stopped(0x00)"
    elif state_enum == Clusters.OperationalState.Enums.OperationalStateEnum.kRunning:
        return "Running(0x01)"
    elif state_enum == Clusters.OperationalState.Enums.OperationalStateEnum.kPaused:
        return "Paused(0x02)"
    elif state_enum == Clusters.OperationalState.Enums.OperationalStateEnum.kError:
        return "Error(0x03)"
    elif state_enum == Clusters.RvcOperationalState.Enums.OperationalStateEnum.kSeekingCharger:
        return "SeekingCharger(0x40)"
    elif state_enum == Clusters.RvcOperationalState.Enums.OperationalStateEnum.kCharging:
        return "Charging(0x41)"
    elif state_enum == Clusters.RvcOperationalState.Enums.OperationalStateEnum.kDocked:
        return "Docked(0x42)"
    else:
        return "UnknownEnumValue"


# Takes an OpState or RvcOpState error enum and returns a string representation
def error_enum_to_text(error_enum):
    if error_enum == Clusters.OperationalState.Enums.ErrorStateEnum.kNoError:
        return "NoError(0x00)"
    elif error_enum == Clusters.OperationalState.Enums.ErrorStateEnum.kUnableToStartOrResume:
        return "UnableToStartOrResume(0x01)"
    elif error_enum == Clusters.OperationalState.Enums.ErrorStateEnum.kUnableToCompleteOperation:
        return "UnableToCompleteOperation(0x02)"
    elif error_enum == Clusters.OperationalState.Enums.ErrorStateEnum.kCommandInvalidInState:
        return "CommandInvalidInState(0x03)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kFailedToFindChargingDock:
        return "FailedToFindChargingDock(0x40)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kStuck:
        return "Stuck(0x41)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kDustBinMissing:
        return "DustBinMissing(0x42)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kDustBinFull:
        return "DustBinFull(0x43)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kWaterTankEmpty:
        return "WaterTankEmpty(0x44)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kWaterTankMissing:
        return "WaterTankMissing(0x45)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kWaterTankLidOpen:
        return "WaterTankLidOpen(0x46)"
    elif error_enum == Clusters.RvcOperationalState.Enums.ErrorStateEnum.kMopCleaningPadMissing:
        return "MopCleaningPadMissing(0x47)"


class TC_RVCOPSTATE_2_3(MatterBaseTest):

    def __init__(self, *args):
        super().__init__(*args)
        self.endpoint = None
        self.is_ci = False
        self.app_pipe = "/tmp/chip_rvc_fifo_"

    async def read_mod_attribute_expect_success(self, endpoint, attribute):
        cluster = Clusters.Objects.RvcOperationalState
        return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)

    async def send_pause_cmd(self) -> Clusters.Objects.RvcOperationalState.Commands.OperationalCommandResponse:
        ret = await self.send_single_cmd(cmd=Clusters.Objects.RvcOperationalState.Commands.Pause(), endpoint=self.endpoint)
        asserts.assert_true(type_matches(ret, Clusters.Objects.RvcOperationalState.Commands.OperationalCommandResponse),
                            "Unexpected return type for Pause")
        return ret

    async def send_resume_cmd(self) -> Clusters.Objects.RvcOperationalState.Commands.OperationalCommandResponse:
        ret = await self.send_single_cmd(cmd=Clusters.Objects.RvcOperationalState.Commands.Resume(), endpoint=self.endpoint)
        asserts.assert_true(type_matches(ret, Clusters.Objects.RvcOperationalState.Commands.OperationalCommandResponse),
                            "Unexpected return type for Resume")
        return ret

    # Prints the step number, reads the operational state attribute and checks if it matches with expected_state
    async def read_operational_state_with_check(self, step_number, expected_state):
        self.print_step(step_number, "Read OperationalState")
        operational_state = await self.read_mod_attribute_expect_success(
            endpoint=self.endpoint, attribute=Clusters.RvcOperationalState.Attributes.OperationalState)
        logging.info("OperationalState: %s" % operational_state)
        asserts.assert_equal(operational_state, expected_state,
                             "OperationalState(%s) should be %s" % (operational_state, state_enum_to_text(expected_state)))

    # Sends the Pause command and checks that the returned error matches the expected_error
    async def send_pause_cmd_with_check(self, step_number, expected_error):
        self.print_step(step_number, "Send Pause command")
        ret = await self.send_pause_cmd()
        asserts.assert_equal(ret.commandResponseState.errorStateID, expected_error,
                             "errorStateID(%s) should be %s" % (ret.commandResponseState.errorStateID,
                                                                error_enum_to_text(expected_error)))

    # Sends the Resume command and checks that the returned error matches the expected_error
    async def send_resume_cmd_with_check(self, step_number, expected_error):
        self.print_step(step_number, "Send Pause command")
        ret = await self.send_resume_cmd()
        asserts.assert_equal(ret.commandResponseState.errorStateID, expected_error,
                             "errorStateID(%s) should be %s" % (ret.commandResponseState.errorStateID,
                                                                error_enum_to_text(expected_error)))

    async def send_run_change_to_mode_cmd(self, new_mode) -> Clusters.Objects.RvcRunMode.Commands.ChangeToModeResponse:
        ret = await self.send_single_cmd(cmd=Clusters.Objects.RvcRunMode.Commands.ChangeToMode(newMode=new_mode),
                                         endpoint=self.endpoint)
        return ret

    # Sends and out-of-band command to the rvc-app
    def write_to_app_pipe(self, command):
        with open(self.app_pipe, "w") as app_pipe:
            app_pipe.write(command + "\n")
        # Delay for pipe command to be processed (otherwise tests are flaky)
        # TODO(#31239): centralize pipe write logic and remove the need of sleep
        sleep(0.001)

    # Prints the instruction and waits for a user input to continue
    def print_instruction(self, step_number, instruction):
        self.print_step(step_number, instruction)
        self.wait_for_user_input(prompt_msg=f"{instruction}, and press Enter when done.")

    def pics_TC_RVCOPSTATE_2_3(self) -> list[str]:
        return ["RVCOPSTATE.S"]

    @async_test_body
    async def test_TC_RVCOPSTATE_2_3(self):

        self.endpoint = self.matter_test_config.endpoint
        asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
        self.is_ci = self.check_pics("PICS_SDK_CI_ONLY")
        if self.is_ci:
            app_pid = self.matter_test_config.app_pid
            if app_pid == 0:
                asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set.c")
            self.app_pipe = self.app_pipe + str(app_pid)

        asserts.assert_true(self.check_pics("RVCOPSTATE.S.A0003"), "RVCOPSTATE.S.A0003 must be supported")
        asserts.assert_true(self.check_pics("RVCOPSTATE.S.A0004"), "RVCOPSTATE.S.A0004 must be supported")
        asserts.assert_true(self.check_pics("RVCOPSTATE.S.C00.Rsp"), "RVCOPSTATE.S.C00.Rsp must be supported")
        asserts.assert_true(self.check_pics("RVCOPSTATE.S.C03.Rsp"), "RVCOPSTATE.S.C03.Rsp must be supported")
        # This command SHALL be supported by an implementation if any of the other commands are supported (6.5)
        asserts.assert_true(self.check_pics("RVCOPSTATE.S.C04.Tx"), "RVCOPSTATE.S.C04.Tx must be supported")

        attributes = Clusters.RvcOperationalState.Attributes
        op_states = Clusters.OperationalState.Enums.OperationalStateEnum
        rvc_op_states = Clusters.RvcOperationalState.Enums.OperationalStateEnum
        op_errors = Clusters.OperationalState.Enums.ErrorStateEnum

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

        # Ensure that the device is in the correct state
        if self.is_ci:
            self.write_to_app_pipe('{"Name": "Reset"}')

        test_step = "Manually put the device in a state where it can receive a Pause command"
        self.print_step(2, test_step)
        if self.is_ci:
            await self.send_run_change_to_mode_cmd(1)
        else:
            self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

        self.print_step(3, "Read OperationalStateList attribute")
        op_state_list = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
                                                                     attribute=attributes.OperationalStateList)

        logging.info("OperationalStateList: %s" % (op_state_list))

        defined_states = [state.value for state in Clusters.OperationalState.Enums.OperationalStateEnum
                          if state is not Clusters.OperationalState.Enums.OperationalStateEnum.kUnknownEnumValue]

        state_ids = set([s.operationalStateID for s in op_state_list])

        asserts.assert_true(all(id in state_ids for id in defined_states), "OperationalStateList is missing a required entry")

        self.print_step(4, "Read OperationalState")
        old_opstate_dut = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
                                                                       attribute=attributes.OperationalState)
        logging.info("OperationalState: %s" % old_opstate_dut)

        await self.send_pause_cmd_with_check(5, op_errors.kNoError)

        await self.read_operational_state_with_check(6, op_states.kPaused)

        if self.check_pics("RVCOPSTATE.S.A0002"):
            self.print_step(7, "Read CountdownTime attribute")
            initial_countdown_time = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
                                                                                  attribute=attributes.CountdownTime)
            logging.info("CountdownTime: %s" % initial_countdown_time)
            if initial_countdown_time is not NullValue:
                in_range = (1 <= initial_countdown_time <= 259200)
            asserts.assert_true(initial_countdown_time is NullValue or in_range,
                                "invalid CountdownTime(%s). Must be in between 1 and 259200, or null " % initial_countdown_time)

            self.print_step(8, "Waiting for 5 seconds")
            sleep(5)

            self.print_step(9, "Read CountdownTime attribute")
            countdown_time = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, attribute=attributes.CountdownTime)
            logging.info("CountdownTime: %s" % countdown_time)
            asserts.assert_true(countdown_time != 0 or countdown_time == NullValue,
                                "invalid CountdownTime(%s). Must be a non zero integer, or null" % countdown_time)
            asserts.assert_equal(countdown_time, initial_countdown_time, "CountdownTime(%s) not equal to the initial CountdownTime(%s)"
                                 % (countdown_time, initial_countdown_time))

        await self.send_pause_cmd_with_check(10, op_errors.kNoError)

        await self.send_resume_cmd_with_check(11, op_errors.kNoError)

        self.print_step(12, "Read OperationalState attribute")
        operational_state = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
                                                                         attribute=attributes.OperationalState)
        logging.info("OperationalState: %s" % operational_state)
        asserts.assert_equal(operational_state, old_opstate_dut,
                             "OperationalState(%s) should be the state before pause (%s)" % (operational_state, old_opstate_dut))

        await self.send_resume_cmd_with_check(13, op_errors.kNoError)

        if self.check_pics("RVCOPSTATE.S.M.RESUME_AFTER_ERR"):
            self.print_instruction(16, "Manually put the device in the Running state")

            await self.read_operational_state_with_check(17, op_states.kRunning)

            self.print_instruction(
                18, "Manually cause the device to pause running due to an error, and be able to resume after clearing the error")

            await self.read_operational_state_with_check(19, op_states.kError)

            self.print_instruction(20, "Manually clear the error")

            await self.read_operational_state_with_check(21, op_states.kPaused)

            await self.send_resume_cmd_with_check(22, op_errors.kNoError)

            await self.read_operational_state_with_check(23, op_states.kRunning)

        if self.check_pics("RVCOPSTATE.S.M.ST_STOPPED"):
            test_step = "Manually put the device in the Stopped(0x00) operational state"
            self.print_step(24, test_step)
            if self.is_ci:
                self.write_to_app_pipe('{"Name": "Reset"}')
            else:
                self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

            await self.read_operational_state_with_check(25, op_states.kStopped)

            await self.send_pause_cmd_with_check(26, op_errors.kCommandInvalidInState)

            await self.send_resume_cmd_with_check(27, op_errors.kCommandInvalidInState)

        if self.check_pics("RVCOPSTATE.S.M.ST_ERROR"):
            test_step = "Manually put the device in the Error(0x03) operational state"
            self.print_step(28, test_step)
            if self.is_ci:
                self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "Stuck"}')
            else:
                self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

            await self.read_operational_state_with_check(29, op_states.kError)

            await self.send_pause_cmd_with_check(30, op_errors.kCommandInvalidInState)

            await self.send_resume_cmd_with_check(31, op_errors.kCommandInvalidInState)

        if self.check_pics("RVCOPSTATE.S.M.ST_CHARGING"):
            test_step = "Manually put the device in the Charging(0x41) operational state"
            self.print_step(32, test_step)
            if self.is_ci:
                self.write_to_app_pipe('{"Name": "Reset"}')
                await self.send_run_change_to_mode_cmd(1)
                await self.send_run_change_to_mode_cmd(0)
                self.write_to_app_pipe('{"Name": "ChargerFound"}')
            else:
                self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

            await self.read_operational_state_with_check(33, rvc_op_states.kCharging)

            await self.send_pause_cmd_with_check(34, op_errors.kCommandInvalidInState)
            test_step = "Manually put the device in the Charging(0x41) operational state and RVC Run Mode cluster's CurrentMode attribute set to a mode with the Idle mode tag"
            self.print_step(35, test_step)
            if not self.is_ci:
                self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

            await self.read_operational_state_with_check(36, rvc_op_states.kCharging)

            await self.send_resume_cmd_with_check(37, op_errors.kCommandInvalidInState)

        if self.check_pics("RVCOPSTATE.S.M.ST_DOCKED"):
            test_step = "Manually put the device in the Docked(0x42) operational state"
            self.print_step(38, test_step)
            if self.is_ci:
                self.write_to_app_pipe('{"Name": "Charged"}')
            else:
                self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

            await self.read_operational_state_with_check(39, rvc_op_states.kDocked)

            await self.send_pause_cmd_with_check(40, op_errors.kCommandInvalidInState)

            test_step = "Manually put the device in the Docked(0x42) operational state and RVC Run Mode cluster's CurrentMode attribute set to a mode with the Idle mode tag"
            self.print_step(41, test_step)
            if not self.is_ci:
                self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

            await self.send_resume_cmd_with_check(42, op_errors.kCommandInvalidInState)

        if self.check_pics("RVCOPSTATE.S.M.ST_SEEKING_CHARGER"):
            test_step = "Manually put the device in the SeekingCharger(0x40) operational state"
            self.print_step(43, test_step)
            if self.is_ci:
                await self.send_run_change_to_mode_cmd(1)
                await self.send_run_change_to_mode_cmd(0)
            else:
                self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

            await self.read_operational_state_with_check(44, rvc_op_states.kSeekingCharger)

            await self.send_resume_cmd_with_check(45, op_errors.kCommandInvalidInState)


if __name__ == "__main__":
    default_matter_test_main()