| # |
| # 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 |
| from time import sleep |
| |
| import chip.clusters as Clusters |
| 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): |
| try: |
| return f'{Clusters.RvcOperationalState.Enums.OperationalStateEnum(state_enum).name[1:]}(0x{state_enum:02x})' |
| except AttributeError: |
| return f'{Clusters.OperationalState.Enums.OperationalStateEnum(state_enum).name[1:]}(0x{state_enum:02x})' |
| |
| |
| # Takes an OpState or RvcOpState error enum and returns a string representation |
| def error_enum_to_text(error_enum): |
| try: |
| return f'{Clusters.RvcOperationalState.Enums.ErrorStateEnum(error_enum).name[1:]}(0x{error_enum:02x})' |
| except AttributeError: |
| return f'{Clusters.OperationalState.Enums.ErrorStateEnum(error_enum).name[1:]}(0x{error_enum:02x})' |
| |
| |
| class TC_RVCOPSTATE_2_4(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_go_home_cmd(self) -> Clusters.Objects.RvcOperationalState.Commands.OperationalCommandResponse: |
| ret = await self.send_single_cmd(cmd=Clusters.Objects.RvcOperationalState.Commands.GoHome(), endpoint=self.endpoint) |
| asserts.assert_true(type_matches(ret, Clusters.Objects.RvcOperationalState.Commands.OperationalCommandResponse), |
| "Unexpected return type for GoHome") |
| return ret |
| |
| # Sends the GoHome command and checks that the returned error matches the expected_error |
| async def send_go_home_cmd_with_check(self, step_number, expected_error): |
| self.print_step(step_number, "Send GoHome command") |
| ret = await self.send_go_home_cmd() |
| asserts.assert_equal(ret.commandResponseState.errorStateID, expected_error, |
| "errorStateID(%s) should be %s" % (ret.commandResponseState.errorStateID, |
| error_enum_to_text(expected_error))) |
| |
| # 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 an RvcRunMode Change to mode command |
| async def send_run_change_to_mode_cmd(self, new_mode): |
| await self.send_single_cmd(cmd=Clusters.Objects.RvcRunMode.Commands.ChangeToMode(newMode=new_mode), |
| endpoint=self.endpoint) |
| |
| # Sends an 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) |
| |
| def pics_TC_RVCOPSTATE_2_4(self) -> list[str]: |
| return ["RVCOPSTATE.S"] |
| |
| @async_test_body |
| async def test_TC_RVCOPSTATE_2_4(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.A0004"), "RVCOPSTATE.S.A0004 must be supported") |
| asserts.assert_true(self.check_pics("RVCOPSTATE.S.C04.Tx"), "RVCOPSTATE.S.C04.Tx must be supported") |
| asserts.assert_true(self.check_pics("RVCOPSTATE.S.C128.Rsp"), "RVCOPSTATE.S.C128.Rsp must be supported") |
| |
| op_states = Clusters.OperationalState.Enums.OperationalStateEnum |
| rvc_op_states = Clusters.RvcOperationalState.Enums.OperationalStateEnum |
| op_errors = Clusters.OperationalState.Enums.ErrorStateEnum |
| |
| # These are the mode values used by the RVC example app that is used in CI. |
| rvc_app_run_mode_idle = 0 |
| rvc_app_run_mode_cleaning = 1 |
| |
| 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"}') |
| |
| if self.check_pics("RVCOPSTATE.S.M.ST_ERROR"): |
| step_name = "Manually put the device in the ERROR operational state" |
| self.print_step(2, step_name) |
| if self.is_ci: |
| self.write_to_app_pipe('{"Name": "ErrorEvent", "Error": "UnableToStartOrResume"}') |
| else: |
| self.wait_for_user_input(step_name) |
| |
| await self.read_operational_state_with_check(3, op_states.kError) |
| |
| await self.send_go_home_cmd_with_check(4, op_errors.kCommandInvalidInState) |
| |
| if self.check_pics("RVCOPSTATE.S.M.ST_CHARGING"): |
| step_name = "Manually put the device in the CHARGING operational state" |
| self.print_step(5, step_name) |
| if self.is_ci: |
| self.write_to_app_pipe('{"Name": "Reset"}') |
| self.write_to_app_pipe('{"Name": "Docked"}') |
| self.write_to_app_pipe('{"Name": "Charging"}') |
| else: |
| self.wait_for_user_input(step_name) |
| |
| await self.read_operational_state_with_check(6, rvc_op_states.kCharging) |
| |
| await self.send_go_home_cmd_with_check(7, op_errors.kCommandInvalidInState) |
| |
| if self.check_pics("RVCOPSTATE.S.M.ST_DOCKED"): |
| step_name = "Manually put the device in the DOCKED operational state" |
| self.print_step(8, step_name) |
| if self.is_ci: |
| self.write_to_app_pipe('{"Name": "Charged"}') |
| else: |
| self.wait_for_user_input(step_name) |
| |
| await self.read_operational_state_with_check(9, rvc_op_states.kDocked) |
| |
| await self.send_go_home_cmd_with_check(10, op_errors.kCommandInvalidInState) |
| |
| if self.check_pics("PICS_M_ST_SEEKING_CHARGER"): |
| step_name = "Manually put the device in the SEEKING CHARGER operational state" |
| self.print_step(8, step_name) |
| if self.is_ci: |
| await self.send_run_change_to_mode_cmd(rvc_app_run_mode_cleaning) |
| await self.send_run_change_to_mode_cmd(rvc_app_run_mode_idle) |
| else: |
| self.wait_for_user_input(step_name) |
| |
| await self.read_operational_state_with_check(9, rvc_op_states.kSeekingCharger) |
| |
| await self.send_go_home_cmd_with_check(10, op_errors.kNoError) |
| |
| |
| if __name__ == "__main__": |
| default_matter_test_main() |
