Skip to content

Commit

Permalink
address comments
Browse files Browse the repository at this point in the history
  • Loading branch information
duyen-tran committed Aug 9, 2023
1 parent e5fdffe commit dbdfa94
Showing 1 changed file with 56 additions and 49 deletions.
105 changes: 56 additions & 49 deletions src/python_testing/TC_RVCOPSTATE_2_3.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,15 +57,16 @@ async def test_TC_RVCOPSTATE_2_3(self):

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")
asserts.assert_true(self.check_pics("RVCOPSTATE.S.C00.Rsp") or self.check_pics("RVCOPSTATE.S.C03.Rsp"),
"At least one must be supported: RVCOPSTATE.S.C00.Rsp, RVCOPSTATE.S.C03.Rsp")

attributes = Clusters.RvcOperationalState.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")
if self.check_pics("RVCOPSTATE.S.C00.Rsp"):
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,
Expand All @@ -79,13 +80,13 @@ async def test_TC_RVCOPSTATE_2_3(self):
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")
if self.check_pics("RVCOPSTATE.S.C00.Rsp"):
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(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)

if self.check_pics("RVCOPSTATE.S.M.ST_PAUSED"):
if self.check_pics("RVCOPSTATE.S.M.ST_PAUSED") and self.check_pics("RVCOPSTATE.S.C00.Rsp"):
self.print_step(5, "Read OperationalState attribute")
operational_state = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
attribute=attributes.OperationalState)
Expand Down Expand Up @@ -114,45 +115,51 @@ async def test_TC_RVCOPSTATE_2_3(self):
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 self.check_pics("RVCOPSTATE.S.C00.Rsp"):
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)

if self.check_pics("RVCOPSTATE.S.C03.Rsp"):
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)

if self.check_pics("RVCOPSTATE.S.M.ST_RUNNING") and self.check_pics("RVCOPSTATE.S.C03.Rsp"):
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)

if self.check_pics("RVCOPSTATE.S.C03.Rsp"):
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)

if self.check_pics("RVCOPSTATE.S.C00.Rsp"):
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)

if self.check_pics("RVCOPSTATE.S.C03.Rsp"):
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__":
Expand Down

0 comments on commit dbdfa94

Please sign in to comment.