Skip to content

Commit

Permalink
address code review feedback
Browse files Browse the repository at this point in the history
  • Loading branch information
@plauric
plauric committed Jul 30, 2024
1 parent 6ae583d commit 8d6af88
Show file tree
Hide file tree
Showing 5 changed files with 111 additions and 124 deletions.
7 changes: 3 additions & 4 deletions src/python_testing/TC_SEAR_1_2.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -130,13 +130,12 @@ async def read_and_validate_selected_areas(self, step):
f"SelectedAreas(len {len(selected_areas)}) should have at most {len(self.areaid_list)} entries")

asserts.assert_true(len(set(selected_areas)) == len(selected_areas), "SelectedAreas must have unique AreaID values!")
selareaid_list = []
for a in selected_areas:
selareaid_list.append(a)

for a in selected_areas:
asserts.assert_true(a in self.areaid_list,
f"SelectedAreas entry {a} has invalid value")
#save so other methods can use this if neeeded
self.selareaid_list = selareaid_list
self.selareaid_list = selected_areas

async def read_and_validate_current_area(self, step):
self.print_step(step, "Read CurrentArea attribute")
Expand Down
11 changes: 2 additions & 9 deletions src/python_testing/TC_SEAR_1_3.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,22 +53,15 @@ async def read_supported_areas(self, step):
endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SupportedAreas)
logging.info("SupportedAreas: %s" % (supported_areas))

areaid_list = []
for a in supported_areas:
areaid_list.append(a.areaID)
return areaid_list
return [a.areaID for a in supported_areas]

async def read_selected_areas(self, step):
self.print_step(step, "Read SelectedAreas attribute")
selected_areas = await self.read_sear_attribute_expect_success(
endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SelectedAreas)
logging.info(f"SelectedAreas {selected_areas}")

selareaid_list = []
for a in selected_areas:
selareaid_list.append(a.areaID)
return selareaid_list

return [a.areaID for a in selected_areas]

async def send_cmd_select_areas_expect_response(self, step, new_areas, expected_response):
self.print_step(step, f"Send SelectAreas command with NewAreas({new_areas})")
Expand Down
4 changes: 3 additions & 1 deletion src/python_testing/TC_SEAR_1_4.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,9 @@ async def test_TC_SEAR_1_4(self):
endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.AttributeList)
logging.info("AttributeList: %s" % (attribute_list))

if not (Clusters.ServiceArea.Attributes.CurrentArea in attribute_list and Clusters.ServiceArea.Attributes.Progress in attribute_list):
if Clusters.ServiceArea.Attributes.CurrentArea not in attribute_list \
and Clusters.ServiceArea.Attributes.Progress not in attribute_list:

cmd_list = await self.read_sear_attribute_expect_success(
endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.AcceptedCommandList)
logging.info("AcceptedCommandList: %s" % (cmd_list))
Expand Down
199 changes: 99 additions & 100 deletions src/python_testing/TC_SEAR_1_5.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,21 +53,15 @@ async def read_supported_areas(self, step):
endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SupportedAreas)
logging.info("SupportedAreas: %s" % (supported_areas))

areaid_list = []
for a in supported_areas:
areaid_list.append(a.areaID)
return areaid_list
return [a.areaID for a in supported_areas]

async def read_selected_areas(self, step):
self.print_step(step, "Read SelectedAreas attribute")
selected_areas = await self.read_sear_attribute_expect_success(
endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SelectedAreas)
logging.info(f"SelectedAreas {selected_areas}")

selareaid_list = []
for a in selected_areas:
selareaid_list.append(a.areaID)
return selareaid_list
return selected_areas

async def read_progress(self, step):
self.print_step(step, "Read Progress attribute")
Expand Down Expand Up @@ -104,7 +98,7 @@ def write_to_app_pipe(self, command):
sleep(0.001)

def TC_SEAR_1_5(self) -> list[str]:
return ["SEAR.S", "SEAR.S.A0005", "SEAR.S.M.HAS_MANUAL_OPERATING_STATE_CONTROL"]
return ["SEAR.S", "SEAR.S.C02.Rsp"]

@async_test_body
async def test_TC_SEAR_1_5(self):
Expand Down Expand Up @@ -135,7 +129,8 @@ async def test_TC_SEAR_1_5(self):
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

await self.send_cmd_skip_area_expect_response(step=4, skipped_area=valid_area_id, expected_response=Clusters.ServiceArea.SkipAreaStatus.kInvalidInMode)
await self.send_cmd_skip_area_expect_response(step=4, skipped_area=valid_area_id,
expected_response=Clusters.ServiceArea.SkipAreaStatus.kInvalidInMode)

if self.check_pics("SEAR.S.M.NO_SELAREA_FOR_SKIP") and self.check_pics("SEAR.S.M.HAS_MANUAL_SKIP_STATE_CONTROL"):
test_step = "Manually intervene to put the device in a state where the state would allow it to execute the SkipArea command, \
Expand All @@ -144,85 +139,89 @@ async def test_TC_SEAR_1_5(self):
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

await self.send_cmd_skip_area_expect_response(step=6, skipped_area=valid_area_id, expected_response=Clusters.ServiceArea.SkipAreaStatus.kInvalidAreaList)
await self.send_cmd_skip_area_expect_response(step=6, skipped_area=valid_area_id,
expected_response=Clusters.ServiceArea.SkipAreaStatus.kInvalidAreaList)

if self.check_pics("SEAR.S.M.VALID_STATE_FOR_SKIP") and self.check_pics("SEAR.S.M.HAS_MANUAL_SKIP_STATE_CONTROL"):
test_step = "Manually intervene to put the device in a state that allows it to execute the SkipArea command"
self.print_step("7", test_step)
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

await self.send_cmd_skip_area_expect_response(step=8, skipped_area=invalid_area_id, expected_response=Clusters.ServiceArea.SkipAreaStatus.kInvalidSkippedArea)
await self.send_cmd_skip_area_expect_response(step=8, skipped_area=invalid_area_id,
expected_response=Clusters.ServiceArea.SkipAreaStatus.kInvalidSkippedArea)

if self.check_pics("SEAR.S.A0005"):
old_progress_list = await self.read_progress(step=9)
asserts.assert_true(len(old_progress_list) > 0, f"len of Progress({len(old_progress_list)}) should not be zero)")
if not self.check_pics("SEAR.S.M.VALID_STATE_FOR_SKIP"):
return

if self.check_pics("SEAR.S.A0005"):
old_progress_list = await self.read_progress(step=9)
asserts.assert_true(len(old_progress_list) > 0, f"len of Progress({len(old_progress_list)}) should not be zero)")

selected_areas = await self.read_selected_areas(step=10)
asserts.assert_true(len(selected_areas) > 0, "SelectedAreas is empty")
selected_areas = await self.read_selected_areas(step=10)
asserts.assert_true(len(selected_areas) > 0, "SelectedAreas is empty")

old_current_area = NullValue
if self.check_pics("SEAR.S.A0003"):
old_current_area = await self.read_current_area(step=11)
old_current_area = NullValue
if self.check_pics("SEAR.S.A0003"):
old_current_area = await self.read_current_area(step=11)

self.print_step("12", "")
if old_current_area is not NullValue:
await self.send_cmd_skip_area_expect_response(step=13, skipped_area=old_current_area, expected_response=Clusters.ServiceArea.SkipAreaStatus.kSuccess)

test_step = "(Manual operation) wait for the device to skip the current area, and start operating at\
the next one it should process, or stop operating"
self.print_step("14", test_step)
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if self.check_pics("SEAR.S.A0005"):
new_progress_list = await self.read_progress(step=15)
asserts.assert_true(len(new_progress_list) > 0, f"len of Progress({len(new_progress_list)}) should not be zero)")

new_current_area = NullValue
if self.check_pics("SEAR.S.A0003"):
new_current_area = await self.read_current_area(step=16)
for p in new_progress_list:
if p.areaID == old_current_area:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
"Progress for areaID({old_current_area}) should be Skipped")
break
test_step = "Indicate whether the device has stopped operating (y/n)"
ret = self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if ret != "y":
for p in new_progress_list:
if p.areaID == new_current_area:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kOperating,
"Progress for areaID({new_current_area}) should be Operating")
break
else:
was_only_skipped_or_completed = True
for p in old_progress_list:
if p.areaID != old_current_area:
if p.status not in (Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
Clusters.ServiceArea.OperationalStatusEnum.kCompleted):
was_only_skipped_or_completed = False
break
if was_only_skipped_or_completed:
for p in new_progress_list:
if p.areaID == old_current_area:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
"Progress for areaID({old_current_area}) should be Skipped")
break
await self.send_cmd_skip_area_expect_response(step=13, skipped_area=old_current_area,
expected_response=Clusters.ServiceArea.SkipAreaStatus.kSuccess)
if self.check_pics("SEAR.S.M.HAS_MANUAL_SKIP_STATE_CONTROL"):
test_step = "(Manual operation) wait for the device to skip the current area, and start operating at\
the next one it should process, or stop operating"
self.print_step("14", test_step)
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if self.check_pics("SEAR.S.A0005"):
new_progress_list = await self.read_progress(step=15)
asserts.assert_true(len(new_progress_list) > 0, f"len of Progress({len(new_progress_list)}) should not be zero)")

new_current_area = NullValue
if self.check_pics("SEAR.S.A0003"):
new_current_area = await self.read_current_area(step=16)
for p in new_progress_list:
if p.areaID == old_current_area:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
"Progress for areaID({old_current_area}) should be Skipped")
break
test_step = "Indicate whether the device has stopped operating (y/n)"
ret = self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if ret != "y":
for p in new_progress_list:
if p.areaID == new_current_area:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kOperating,
"Progress for areaID({new_current_area}) should be Operating")
break

was_only_skipped_or_completed = True
for p in old_progress_list:
if p.areaID != old_current_area:
if p.status not in (Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
Clusters.ServiceArea.OperationalStatusEnum.kCompleted):
was_only_skipped_or_completed = False
break
if was_only_skipped_or_completed:
asserts.assert_true(ret == "y", "The device should not be operating")

self.print_step("17", "")
return

if self.check_pics("SEAR.S.M.HAS_MANUAL_SKIP_STATE_CONTROL"):
test_step = "Manually intervene to put the device in a state that allows it to execute the SkipArea command"
self.print_step("18", test_step)
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if self.check_pics("SEAR.S.M.VALID_STATE_FOR_SKIP") and self.check_pics("SEAR.S.M.HAS_MANUAL_SKIP_STATE_CONTROL"):
test_step = "Manually intervene to put the device in a state that allows it to execute the SkipArea command"
self.print_step("18", test_step)
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if self.check_pics("SEAR.S.A0005"):
self.print_step("19", "")
if len(old_progress_list) == 0:
return

area_to_skip = NullValue
self.print_step("20", "")
for p in old_progress_list:
Expand All @@ -234,41 +233,41 @@ async def test_TC_SEAR_1_5(self):
if area_to_skip is NullValue:
return

await self.send_cmd_skip_area_expect_response(step=21, skipped_area=area_to_skip, expected_response=Clusters.ServiceArea.SkipAreaStatus.kSuccess)
await self.send_cmd_skip_area_expect_response(step=21, skipped_area=area_to_skip,
expected_response=Clusters.ServiceArea.SkipAreaStatus.kSuccess)

test_step = "(Manual operation) wait for the device to update Progress or to stop operating"
self.print_step("22", test_step)
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
test_step = "(Manual operation) wait for the device to update Progress or to stop operating"
self.print_step("22", test_step)
if not self.is_ci:
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if self.check_pics("SEAR.S.A0005"):
new_progress_list = await self.read_progress(step=23)
asserts.assert_true(len(new_progress_list) > 0, f"len of Progress({len(new_progress_list)}) should not be zero)")
if self.check_pics("SEAR.S.A0005"):
new_progress_list = await self.read_progress(step=23)
asserts.assert_true(len(new_progress_list) > 0, f"len of Progress({len(new_progress_list)}) should not be zero)")

for p in new_progress_list:
if p.areaID == area_to_skip:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
"Progress for areaID({new_current_area}) should be Skipped")
break
for p in new_progress_list:
if p.areaID == area_to_skip:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
"Progress for areaID({new_current_area}) should be Skipped")
break

test_step = "Indicate whether the device has stopped operating (y/n)"
ret = self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

if ret != "y":
was_only_skipped_or_completed = True
for p in old_progress_list:
if p.areaID != area_to_skip:
if p.status not in (Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
Clusters.ServiceArea.OperationalStatusEnum.kCompleted):
was_only_skipped_or_completed = False
break
if was_only_skipped_or_completed:
for p in new_progress_list:
if p.areaID == old_current_area:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
"Progress for areaID({old_current_area}) should be Skipped")
break
test_step = "Indicate whether the device has stopped operating (y/n)"
ret = self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")

was_only_skipped_or_completed = True
for p in old_progress_list:
if p.areaID != area_to_skip:
if p.status not in (Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
Clusters.ServiceArea.OperationalStatusEnum.kCompleted):
was_only_skipped_or_completed = False
break
if was_only_skipped_or_completed:
asserts.assert_true(ret == "y", "The device should not be operating")
for p in new_progress_list:
if p.areaID == old_current_area:
asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
"Progress for areaID({old_current_area}) should be Skipped")
break

if __name__ == "__main__":
default_matter_test_main()
Loading

0 comments on commit 8d6af88

Please sign in to comment.