Skip to content

Commit

Permalink
opstate - add operational state test scripts (#28373)
Browse files Browse the repository at this point in the history 
* add operational state scripts

* use pics not pixit

* address comments

* address comments

* address comments

* clean up
  • Loading branch information
@duyen-tran
duyen-tran authored Jul 31, 2023
1 parent 55a1857 commit 521eb16
Show file tree
Hide file tree
Showing 2 changed files with 343 additions and 0 deletions.
185 changes: 185 additions & 0 deletions src/python_testing/TC_OPSTATE_2_1.py
View file
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()
158 changes: 158 additions & 0 deletions src/python_testing/TC_OPSTATE_2_3.py
View file
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()

0 comments on commit 521eb16

Please sign in to comment.