-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
opstate - add operational state test scripts (#28373)
* add operational state scripts * use pics not pixit * address comments * address comments * address comments * clean up
- Loading branch information
1 parent
9f13eac
commit 9959e3e
Showing
2 changed files
with
343 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,185 @@ | ||
| # | ||
| # 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 chip.clusters as Clusters | ||
| from chip.clusters.Types import NullValue | ||
| from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main | ||
| from mobly import asserts | ||
|
|
||
| # This test requires several additional command line arguments | ||
| # run with | ||
| # --int-arg PIXIT_ENDPOINT:<endpoint> | ||
|
|
||
|
|
||
| class TC_OPSTATE_2_1(MatterBaseTest): | ||
|
|
||
| async def read_mod_attribute_expect_success(self, endpoint, attribute): | ||
| cluster = Clusters.Objects.OperationalState | ||
| return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute) | ||
|
|
||
| async def read_and_validate_opstate(self, step, expected_state): | ||
| self.print_step(step, "Read OperationalState attribute") | ||
| operational_state = await self.read_mod_attribute_expect_success( | ||
| endpoint=self.endpoint, attribute=Clusters.OperationalState.Attributes.OperationalState) | ||
| logging.info("OperationalState: %s" % (operational_state)) | ||
| asserts.assert_equal(operational_state, expected_state, | ||
| "OperationalState(%s) should equal %s" % (operational_state, expected_state)) | ||
|
|
||
| async def read_and_validate_operror(self, step, expected_error): | ||
| self.print_step(step, "Read OperationalError attribute") | ||
| operational_error = await self.read_mod_attribute_expect_success( | ||
| endpoint=self.endpoint, attribute=Clusters.OperationalState.Attributes.OperationalError) | ||
| logging.info("OperationalError: %s" % (operational_error)) | ||
| asserts.assert_equal(operational_error.errorStateID, expected_error, | ||
| "errorStateID(%s) should equal %s" % (operational_error.errorStateID, expected_error)) | ||
|
|
||
| @async_test_body | ||
| async def test_TC_OPSTATE_2_1(self): | ||
|
|
||
| asserts.assert_true('PIXIT_ENDPOINT' in self.matter_test_config.global_test_params, | ||
| "PIXIT_ENDPOINT must be included on the command line in " | ||
| "the --int-arg flag as PIXIT_ENDPOINT:<endpoint>") | ||
|
|
||
| self.endpoint = self.matter_test_config.global_test_params['PIXIT_ENDPOINT'] | ||
|
|
||
| attributes = Clusters.OperationalState.Attributes | ||
|
|
||
| self.print_step(1, "Commissioning, already done") | ||
|
|
||
| if self.check_pics("OPSTATE.S.A0000"): | ||
| self.print_step(2, "Read PhaseList attribute") | ||
| phase_list = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, attribute=attributes.PhaseList) | ||
|
|
||
| if phase_list == NullValue: | ||
| logging.info("PhaseList is null") | ||
| else: | ||
| logging.info("PhaseList: %s" % (phase_list)) | ||
|
|
||
| phase_list_len = len(phase_list) | ||
|
|
||
| asserts.assert_less_equal(phase_list_len, 32, | ||
| "PhaseList length(%d) must be less than 32!" % phase_list_len) | ||
|
|
||
| if self.check_pics("OPSTATE.S.A0001"): | ||
| self.print_step(3, "Read CurrentPhase attribute") | ||
| current_phase = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, attribute=attributes.CurrentPhase) | ||
| logging.info("CurrentPhase: %s" % (current_phase)) | ||
|
|
||
| if phase_list == NullValue: | ||
| asserts.assert_true(current_phase == NullValue, "CurrentPhase(%s) should be null" % current_phase) | ||
| else: | ||
| asserts.assert_true(0 <= current_phase and current_phase < phase_list_len, | ||
| "CurrentPhase(%s) must be between 0 and %s" % (current_phase, (phase_list_len - 1))) | ||
|
|
||
| if self.check_pics("OPSTATE.S.A0002"): | ||
| self.print_step(4, "Read CountdownTime attribute") | ||
| countdown_time = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, | ||
| attribute=attributes.CountdownTime) | ||
|
|
||
| logging.info("CountdownTime: %s" % (countdown_time)) | ||
| if countdown_time is not NullValue: | ||
| asserts.assert_true(0 <= countdown_time <= 259200, | ||
| "CountdownTime(%s) must be between 0 and 259200" % countdown_time) | ||
|
|
||
| if self.check_pics("OPSTATE.S.A0003"): | ||
| self.print_step(5, "Read OperationalStateList attribute") | ||
| operational_state_list = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, | ||
| attribute=attributes.OperationalStateList) | ||
|
|
||
| logging.info("OperationalStateList: %s" % (operational_state_list)) | ||
|
|
||
| defined_states = [state.value for state in Clusters.OperationalState.Enums.OperationalStateEnum | ||
| if state is not Clusters.OperationalState.Enums.OperationalStateEnum.kUnknownEnumValue] | ||
|
|
||
| for state in operational_state_list: | ||
| in_range = (0x80 <= state.operationalStateID <= 0xBF) | ||
| asserts.assert_true(state.operationalStateID in defined_states or in_range, | ||
| "Found a OperationalStateList entry with invalid ID value!") | ||
| if in_range: | ||
| asserts.assert_true(state.operationalStateLabel is not None, | ||
| "The OperationalStateLabel should be populated") | ||
| if state.operationalStateID == Clusters.OperationalState.Enums.OperationalStateEnum.kError: | ||
| error_state_present = True | ||
|
|
||
| asserts.assert_true(error_state_present, "The OperationalStateList does not have an ID entry of Error(0x03)") | ||
|
|
||
| if self.check_pics("OPSTATE.S.A0004"): | ||
| self.print_step(6, "Read OperationalState attribute") | ||
| operational_state = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, | ||
| attribute=attributes.OperationalState) | ||
|
|
||
| logging.info("OperationalState: %s" % (operational_state)) | ||
|
|
||
| in_range = (0x80 <= operational_state <= 0xBF) | ||
| asserts.assert_true(operational_state in defined_states or in_range, "OperationalState has an invalid ID value!") | ||
|
|
||
| if self.check_pics("OPSTATE.S.M.ST_STOPPED"): | ||
| self.print_step("6a", "Manually put the device in the stopped state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_opstate(step="6b", expected_state=Clusters.OperationalState.Enums.OperationalStateEnum.kStopped) | ||
| if self.check_pics("OPSTATE.S.M.ST_RUNNING"): | ||
| self.print_step("6c", "Manually put the device in the running state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_opstate(step="6d", expected_state=Clusters.OperationalState.Enums.OperationalStateEnum.kRunning) | ||
| if self.check_pics("OPSTATE.S.M.ST_PAUSED"): | ||
| self.print_step("6e", "Manually put the device in the paused state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_opstate(step="6f", expected_state=Clusters.OperationalState.Enums.OperationalStateEnum.kPaused) | ||
| if self.check_pics("OPSTATE.S.M.ST_ERROR"): | ||
| self.print_step("6g", "Manually put the device in the error state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_opstate(step="6h", expected_state=Clusters.OperationalState.Enums.OperationalStateEnum.kError) | ||
|
|
||
| if self.check_pics("OPSTATE.S.A0005"): | ||
| self.print_step(7, "Read OperationalError attribute") | ||
| operational_error = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, | ||
| attribute=attributes.OperationalError) | ||
|
|
||
| logging.info("OperationalError: %s" % (operational_error)) | ||
|
|
||
| # Defined Errors | ||
| defined_errors = [error.value for error in Clusters.OperationalState.Enums.ErrorStateEnum | ||
| if error is not Clusters.OperationalState.Enums.ErrorStateEnum.kUnknownEnumValue] | ||
|
|
||
| in_range = (0x80 <= operational_error.errorStateID <= 0xBF) | ||
| asserts.assert_true(operational_error.errorStateID in defined_errors | ||
| or in_range, "OperationalError has an invalid ID value!") | ||
| if in_range: | ||
| asserts.assert_true(operational_error.errorStateLabel is not None, "ErrorStateLabel should be populated") | ||
|
|
||
| if self.check_pics("OPSTATE.S.M.ERR_NO_ERROR"): | ||
| self.print_step("7a", "Manually put the device in the no error state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_operror(step="7b", expected_error=Clusters.OperationalState.Enums.ErrorStateEnum.kNoError) | ||
| if self.check_pics("OPSTATE.S.M.ERR_UNABLE_TO_START_OR_RESUME"): | ||
| self.print_step("7c", "Manually put the device in the unable to start or resume error state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_operror(step="7d", expected_error=Clusters.OperationalState.Enums.ErrorStateEnum.kUnableToStartOrResume) | ||
| if self.check_pics("OPSTATE.S.M.ERR_UNABLE_TO_COMPLETE_OPERATION"): | ||
| self.print_step("7e", "Manually put the device in the unable to complete operation error state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_operror(step="7f", expected_error=Clusters.OperationalState.Enums.ErrorStateEnum.kUnableToCompleteOperation) | ||
| if self.check_pics("OPSTATE.S.M.ERR_COMMAND_INVALID_STATE"): | ||
| self.print_step("7g", "Manually put the device in the command invalid error state") | ||
| input("Press Enter when done.\n") | ||
| await self.read_and_validate_operror(step="7h", expected_error=Clusters.OperationalState.Enums.ErrorStateEnum.kCommandInvalidInState) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| default_matter_test_main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,158 @@ | ||
| # | ||
| # 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 time | ||
|
|
||
| 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 | ||
|
|
||
| # This test requires several additional command line arguments | ||
| # run with | ||
| # --int-arg PIXIT_ENDPOINT:<endpoint> | ||
|
|
||
|
|
||
| class TC_OPSTATE_2_3(MatterBaseTest): | ||
|
|
||
| async def read_mod_attribute_expect_success(self, endpoint, attribute): | ||
| cluster = Clusters.Objects.OperationalState | ||
| return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute) | ||
|
|
||
| async def send_pause_cmd(self) -> Clusters.Objects.OperationalState.Commands.Pause: | ||
| ret = await self.send_single_cmd(cmd=Clusters.Objects.OperationalState.Commands.Pause(), endpoint=self.endpoint) | ||
| asserts.assert_true(type_matches(ret, Clusters.Objects.OperationalState.Commands.OperationalCommandResponse), | ||
| "Unexpected return type for Pause") | ||
| return ret | ||
|
|
||
| async def send_resume_cmd(self) -> Clusters.Objects.OperationalState.Commands.Resume: | ||
| ret = await self.send_single_cmd(cmd=Clusters.Objects.OperationalState.Commands.Resume(), endpoint=self.endpoint) | ||
| asserts.assert_true(type_matches(ret, Clusters.Objects.OperationalState.Commands.OperationalCommandResponse), | ||
| "Unexpected return type for Resume") | ||
| return ret | ||
|
|
||
| @async_test_body | ||
| async def test_TC_OPSTATE_2_3(self): | ||
|
|
||
| asserts.assert_true('PIXIT_ENDPOINT' in self.matter_test_config.global_test_params, | ||
| "PIXIT_ENDPOINT must be included on the command line in " | ||
| "the --int-arg flag as PIXIT_ENDPOINT:<endpoint>") | ||
|
|
||
| self.endpoint = self.matter_test_config.global_test_params['PIXIT_ENDPOINT'] | ||
|
|
||
| asserts.assert_true(self.check_pics("OPSTATE.S.A0002"), "OPSTATE.S.A0002 must be supported") | ||
| asserts.assert_true(self.check_pics("OPSTATE.S.A0003"), "OPSTATE.S.A0003 must be supported") | ||
| asserts.assert_true(self.check_pics("OPSTATE.S.A0004"), "OPSTATE.S.A0004 must be supported") | ||
| asserts.assert_true(self.check_pics("OPSTATE.S.C00.Rsp"), "OPSTATE.S.C00.Rsp must be supported") | ||
| asserts.assert_true(self.check_pics("OPSTATE.S.C03.Rsp"), "OPSTATE.S.C03.Rsp must be supported") | ||
|
|
||
| attributes = Clusters.OperationalState.Attributes | ||
|
|
||
| self.print_step(1, "Commissioning, already done") | ||
|
|
||
| self.print_step(2, "Manually put the device in a state where it can receive a Pause command") | ||
| input("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, "Send Pause command") | ||
| ret = await self.send_pause_cmd() | ||
| asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OperationalState.Enums.ErrorStateEnum.kNoError, | ||
| "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) | ||
|
|
||
| self.print_step(5, "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, Clusters.OperationalState.Enums.OperationalStateEnum.kPaused, | ||
| "OperationalState ID should be Paused(0x02)") | ||
|
|
||
| self.print_step(6, "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(7, "Waiting for 5 seconds") | ||
| time.sleep(5) | ||
|
|
||
| self.print_step(8, "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) does not equal to the intial CountdownTime(%s)" | ||
| % (countdown_time, initial_countdown_time)) | ||
|
|
||
| self.print_step(9, "Send Pause command") | ||
| ret = await self.send_pause_cmd() | ||
| asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OperationalState.Enums.ErrorStateEnum.kNoError, | ||
| "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) | ||
|
|
||
| self.print_step(10, "Send Resume command") | ||
| ret = await self.send_resume_cmd() | ||
| asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OperationalState.Enums.ErrorStateEnum.kNoError, | ||
| "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) | ||
|
|
||
| self.print_step(11, "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, Clusters.OperationalState.Enums.OperationalStateEnum.kRunning, | ||
| "OperationalState(%s) should be Running(0x01)" % operational_state) | ||
|
|
||
| self.print_step(12, "Send Resume command") | ||
| ret = await self.send_resume_cmd() | ||
| asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OperationalState.Enums.ErrorStateEnum.kNoError, | ||
| "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) | ||
|
|
||
| self.print_step(13, "Manually put the device in a state where it cannot receive a Pause command") | ||
| input("Press Enter when done.\n") | ||
|
|
||
| self.print_step(14, "Send Pause command") | ||
| ret = await self.send_pause_cmd() | ||
| asserts.assert_equal(ret.commandResponseState.errorStateID, | ||
| Clusters.OperationalState.Enums.ErrorStateEnum.kCommandInvalidInState, | ||
| "errorStateID(%s) should be CommandInvalidInState(0x03)" % ret.commandResponseState.errorStateID) | ||
|
|
||
| self.print_step(15, "Manually put the device in a state where it cannot receive a Resume command") | ||
| input("Press Enter when done.\n") | ||
|
|
||
| self.print_step(16, "Send Resume command") | ||
| ret = await self.send_resume_cmd() | ||
| asserts.assert_equal(ret.commandResponseState.errorStateID, | ||
| Clusters.OperationalState.Enums.ErrorStateEnum.kCommandInvalidInState, | ||
| "errorStateID(%s) should be CommandInvalidInState(0x03)" % ret.commandResponseState.errorStateID) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| default_matter_test_main() |