Skip to content

Commit

Permalink
Added steps to me TP easier to write
Browse files Browse the repository at this point in the history
  • Loading branch information
@rbultman
rbultman committed Aug 2, 2024
1 parent b6b8cbc commit 500c024
Showing 1 changed file with 41 additions and 30 deletions.
71 changes: 41 additions & 30 deletions src/python_testing/TC_OpstateCommon.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1231,15 +1231,20 @@ def steps_TC_OPSTATE_BASE_2_6(self) -> list[TestStep]:
TestStep(2, "Subscribe to CountdownTime attribute"),
TestStep(3, "Manually put the DUT into a state where it will use the CountdownTime attribute, "
"the initial value of the CountdownTime is greater than 30, "
"and it will begin counting down the CountdownTime attribute."),
TestStep(4, "Over a period of 30 seconds, TH counts all report transactions with an attribute "
"and it will begin counting down the CountdownTime attribute. "
"Test harness reads the CountdownTime attribute."),
TestStep(4, "Test harness reads the CountdownTime attribute."),
TestStep(5, "Over a period of 30 seconds, TH counts all report transactions with an attribute "
"report for the CountdownTime attribute in numberOfReportsReceived"),
TestStep(5, "Until the current operation finishes, TH counts all report transactions with "
TestStep(6, "Test harness reads the CountdownTime attribute."),
TestStep(7, "Until the current operation finishes, TH counts all report transactions with "
"an attribute report for the CountdownTime attribute in numberOfReportsReceived and saves up to 5 such reports."),
TestStep(6, "Manually put the DUT into a state where it will use the CountdownTime attribute, "
"the initial value of the CountdownTime is greater than 30, and it will begin counting down the CountdownTime attribute."),
TestStep(7, "TH reads from the DUT the OperationalState attribute"),
TestStep(8, "Manually put the device in the Paused(0x02) operational state")
TestStep(8, "Manually put the DUT into a state where it will use the CountdownTime attribute, "
"the initial value of the CountdownTime is greater than 30, and it will begin counting down the CountdownTime attribute."
"Test harness reads the CountdownTime attribute."),
TestStep(9, "TH reads from the DUT the OperationalState attribute"),
TestStep(10, "Test harness reads the CountdownTime attribute."),
TestStep(11, "Manually put the device in the Paused(0x02) operational state")
]
return steps

Expand Down Expand Up @@ -1275,9 +1280,10 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1):
self.skip_step(3)

sub_handler.reset()
self.step(4)
countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
if countdownTime is not NullValue:
self.step(4)
self.step(5)
logging.info('Test will now collect data for 30 seconds')
time.sleep(30)

Expand All @@ -1286,30 +1292,34 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1):
asserts.assert_less_equal(count, 5, "Received more than 5 reports for CountdownTime")
asserts.assert_greater_equal(count, 0, "Did not receive any reports for CountdownTime")
else:
self.skip_step(4)
self.skip_step(5)

attr_value = await self.read_expect_success(
endpoint=endpoint,
attribute=attributes.OperationalState)
self.step(6)
countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
if attr_value == cluster.Enums.OperationalStateEnum.kRunning and countdownTime is not NullValue:
self.step(5)
wait_count = 0
while (attr_value != cluster.Enums.OperationalStateEnum.kStopped) and (wait_count < 20):
time.sleep(1)
wait_count = wait_count + 1
attr_value = await self.read_expect_success(
endpoint=endpoint,
attribute=attributes.OperationalState)
count = sub_handler.attribute_report_counts[attributes.CountdownTime]
asserts.assert_less_equal(count, 5, "Received more than 5 reports for CountdownTime")
asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
if countdownTime is not NullValue:
attr_value = await self.read_expect_success(
endpoint=endpoint,
attribute=attributes.OperationalState)
if attr_value == cluster.Enums.OperationalStateEnum.kRunning:
self.step(7)
wait_count = 0
while (attr_value != cluster.Enums.OperationalStateEnum.kStopped) and (wait_count < 20):
time.sleep(1)
wait_count = wait_count + 1
attr_value = await self.read_expect_success(
endpoint=endpoint,
attribute=attributes.OperationalState)
count = sub_handler.attribute_report_counts[attributes.CountdownTime]
asserts.assert_less_equal(count, 5, "Received more than 5 reports for CountdownTime")
asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
else:
self.skip_step(7)
else:
self.skip_step(5)
self.skip_step(7)

sub_handler.reset()
if self.pics_guard(self.check_pics(f"{self.test_info.pics_code}.S.M.ST_RUNNING")):
self.step(6)
self.step(8)
self.send_manual_or_pipe_command(name="OperationalStateChange",
device=self.device,
operation="Start")
Expand All @@ -1322,22 +1332,23 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1):
count = sub_handler.attribute_report_counts[attributes.CountdownTime]
asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
else:
self.skip_step(6)
self.skip_step(8)

self.step(7)
self.step(9)
await self.read_and_expect_value(endpoint=endpoint,
attribute=attributes.OperationalState,
expected_value=cluster.Enums.OperationalStateEnum.kRunning)

sub_handler.reset()
self.step(10)
countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
if self.pics_guard(self.check_pics(f"{self.test_info.pics_code}.S.M.ST_PAUSED")) and countdownTime is not NullValue:
self.step(8)
self.step(11)
self.send_manual_or_pipe_command(name="OperationalStateChange",
device=self.device,
operation="Pause")
time.sleep(1)
count = sub_handler.attribute_report_counts[attributes.CountdownTime]
asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
else:
self.skip_step(8)
self.skip_step(11)

0 comments on commit 500c024

Please sign in to comment.