ServiceArea test scripts (#34548)
* initial commit
* fix bugs
* fix issues reported by the linter
* fix bug in checking for unique areaDesc
* add TC 1.5
* Update src/python_testing/TC_SEAR_1_2.py
Co-authored-by: William <hicklin@users.noreply.github.com>
* Update src/python_testing/TC_SEAR_1_2.py
Co-authored-by: William <hicklin@users.noreply.github.com>
* address code review comments
* fix issue introduced by the previous commit
* address code review feedback
* Update src/python_testing/TC_SEAR_1_2.py
Co-authored-by: Kiel Oleson <kielo@apple.com>
* address code review feedback
* remove PICS checked by the TC_SEAR_1.6
* more code review updates
* Restyled by autopep8
---------
Co-authored-by: William <hicklin@users.noreply.github.com>
Co-authored-by: Kiel Oleson <kielo@apple.com>
Co-authored-by: Restyled.io <commits@restyled.io>
diff --git a/src/python_testing/TC_SEAR_1_2.py b/src/python_testing/TC_SEAR_1_2.py
new file mode 100644
index 0000000..4ebb334
--- /dev/null
+++ b/src/python_testing/TC_SEAR_1_2.py
@@ -0,0 +1,368 @@
+#
+# Copyright (c) 2024 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.
+#
+
+# TODO - this was copied/pasted from another test, it needs to be reviewed and updated
+# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
+# for details about the block below.
+#
+# === BEGIN CI TEST ARGUMENTS ===
+# test-runner-runs: run1
+# test-runner-run/run1/app: ${CHIP_RVC_APP}
+# test-runner-run/run1/factoryreset: True
+# test-runner-run/run1/quiet: True
+# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
+# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values --endpoint 1 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# === END CI TEST ARGUMENTS ===
+
+import logging
+from time import sleep
+
+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
+
+
+class TC_SEAR_1_2(MatterBaseTest):
+ def __init__(self, *args):
+ super().__init__(*args)
+ self.endpoint = None
+ self.is_ci = False
+ self.app_pipe = "/tmp/chip_rvc_fifo_"
+ self.mapid_list = []
+
+ # this must be kept in sync with the definitions from the Common Landmark Semantic Tag Namespace
+ self.MAX_LANDMARK_ID = 0x33
+
+ # this must be kept in sync with the definitions from the Common Relative Position Semantic Tag Namespace
+ self.MAX_RELPOS_ID = 0x07
+
+ async def read_sear_attribute_expect_success(self, endpoint, attribute):
+ cluster = Clusters.Objects.ServiceArea
+ return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
+
+ async def read_and_validate_supported_maps(self, step):
+ self.print_step(step, "Read SupportedMaps attribute")
+ supported_maps = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SupportedMaps)
+ logging.info("SupportedMaps: %s" % (supported_maps))
+ asserts.assert_less_equal(len(supported_maps), 255,
+ "SupportedMaps should have max 255 entries")
+
+ mapid_list = [m.mapID for m in supported_maps]
+ asserts.assert_true(len(set(mapid_list)) == len(mapid_list), "SupportedMaps must have unique MapID values!")
+
+ name_list = [m.name for m in supported_maps]
+ asserts.assert_true(len(set(name_list)) == len(name_list), "SupportedMaps must have unique Name values!")
+
+ # save so other methods can use this if neeeded
+ self.mapid_list = mapid_list
+
+ async def read_and_validate_supported_areas(self, step):
+ self.print_step(step, "Read SupportedAreas attribute")
+ supported_areas = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SupportedAreas)
+ logging.info("SupportedAreas: %s" % (supported_areas))
+ asserts.assert_less_equal(len(supported_areas), 255,
+ "SupportedAreas should have max 255 entries")
+ areaid_list = []
+ areadesc_s = set()
+ for a in supported_areas:
+ asserts.assert_true(a.areaID not in areaid_list, "SupportedAreas must have unique AreaID values!")
+
+ areaid_list.append(a.areaID)
+
+ if len(self.mapid_list) > 0:
+ asserts.assert_is_not(a.mapID, NullValue,
+ f"SupportedAreas entry with AreaID({a.areaID}) should not have null MapID")
+ asserts.assert_is(a.mapID in self.mapid_list,
+ f"SupportedAreas entry with AreaID({a.areaID}) has unknown MapID({a.mapID})")
+ k = f"mapID:{a.mapID} areaDesc:{a.areaDesc}"
+ asserts.assert_true(k not in areadesc_s,
+ f"SupportedAreas must have unique MapID({a.mapID}) + AreaDesc({a.areaDesc}) values!")
+ areadesc_s.add(k)
+ else:
+ # empty SupportedMaps
+ asserts.assert_is(a.mapID, NullValue,
+ f"SupportedAreas entry with AreaID({a.areaID}) should have null MapID")
+ k = f"areaDesc:{a.areaDesc}"
+ asserts.assert_true(k not in areadesc_s, f"SupportedAreas must have unique AreaDesc({a.areaDesc}) values!")
+ areadesc_s.add(k)
+
+ if a.locationInfo is NullValue and a.landmarkTag is NullValue:
+ asserts.assert_true(
+ f"SupportedAreas entry with AreaID({a.areaID}) should not have null LocationInfo and null LandmarkTag")
+ if a.landmarkTag is not NullValue:
+ asserts.assert_true(a.landmarkTag <= self.MAX_LANDMARK_ID,
+ f"SupportedAreas entry with AreaID({a.areaID}) has invalid LandmarkTag({a.landmarkTag})")
+ asserts.assert_true(a.positionTag is NullValue or a.positionTag in range(0, self.MAX_RELPOS_ID),
+ f"SupportedAreas entry with AreaID({a.areaID}) has invalid PositionTag({a.positionTag})")
+ # save so other methods can use this if neeeded
+ self.areaid_list = areaid_list
+
+ async def read_and_validate_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}")
+
+ # TODO how to check if all entries are uint32?
+
+ asserts.assert_true(len(selected_areas) <= len(self.areaid_list),
+ 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!")
+
+ 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 = selected_areas
+
+ async def read_and_validate_current_area(self, step):
+ self.print_step(step, "Read CurrentArea attribute")
+ current_area = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.CurrentArea)
+ logging.info(f"CurrentArea {current_area}")
+
+ asserts.assert_true((len(self.selareaid_list) == 0 and current_area is NullValue)
+ or
+ current_area in self.selareaid_list,
+ f"CurrentArea {current_area} is invalid. SelectedAreas is {self.selareaid_list}.")
+ # save so other methods can use this if neeeded
+ self.current_area = current_area
+
+ async def read_and_validate_estimated_end_time(self, step):
+ import time
+ read_time = int(time.time())
+ self.print_step(step, "Read EstimatedEndTime attribute")
+ estimated_end_time = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.EstimatedEndTime)
+ logging.info(f"EstimatedEndTime {estimated_end_time}")
+
+ if self.current_area is NullValue:
+ asserts.assert_true(estimated_end_time is NullValue,
+ "EstimatedEndTime should be null if CurrentArea is null.")
+ else:
+ # allow for some clock skew
+ asserts.assert_true(estimated_end_time >= read_time - 3*60,
+ f"EstimatedEndTime({estimated_end_time}) should be greater than the time when it was read({read_time})")
+
+ async def read_and_validate_progress(self, step):
+ self.print_step(step, "Read Progress attribute")
+ progress = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.Progress)
+ logging.info(f"Progress {progress}")
+
+ asserts.assert_true(len(progress) <= len(self.areaid_list),
+ f"Progress(len {len(progress)}) should have at most {len(self.areaid_list)} entries")
+
+ progareaid_list = []
+ for p in progress:
+ if p.areaID in progareaid_list:
+ asserts.fail("Progress must have unique AreaID values!")
+ else:
+ progareaid_list.append(p.areaID)
+ asserts.assert_true(p.areaID in self.areaid_list,
+ f"Progress entry has invalid AreaID value ({p.areaID})")
+ asserts.assert_true(p.status in (Clusters.ServiceArea.OperationalStatusEnum.kPending,
+ Clusters.ServiceArea.OperationalStatusEnum.kOperating,
+ Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
+ Clusters.ServiceArea.OperationalStatusEnum.kCompleted),
+ f"Progress entry has invalid Status value ({p.status})")
+ if p.status not in (Clusters.ServiceArea.OperationalStatusEnum.kSkipped, Clusters.ServiceArea.OperationalStatusEnum.kCompleted):
+ asserts.assert_true(p.totalOperationalTime is NullValue,
+ f"Progress entry should have a null TotalOperationalTime value (Status is {p.status})")
+ # TODO how to check that InitialTimeEstimate is either null or uint32?
+
+ # Sends and out-of-band command to the rvc-app
+ def write_to_app_pipe(self, command):
+ with open(self.app_pipe, "w") as app_pipe:
+ app_pipe.write(command + "\n")
+ # Allow some time for the command to take effect.
+ # This removes the test flakyness which is very annoying for everyone in CI.
+ sleep(0.001)
+
+ def TC_SEAR_1_2(self) -> list[str]:
+ return ["SEAR.S"]
+
+ @async_test_body
+ async def test_TC_SEAR_1_2(self):
+ self.endpoint = self.matter_test_config.endpoint
+ asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
+ self.is_ci = self.check_pics("PICS_SDK_CI_ONLY")
+ if self.is_ci:
+ app_pid = self.matter_test_config.app_pid
+ if app_pid == 0:
+ asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set")
+ self.app_pipe = self.app_pipe + str(app_pid)
+
+ self.print_step(1, "Commissioning, already done")
+
+ # Ensure that the device is in the correct state
+ if self.is_ci:
+ self.write_to_app_pipe('{"Name": "Reset"}')
+
+ if self.check_pics("SEAR.S.F02"):
+ await self.read_and_validate_supported_maps(step=2)
+
+ await self.read_and_validate_supported_areas(step=3)
+
+ await self.read_and_validate_selected_areas(step=4)
+
+ if self.check_pics("SEAR.S.A0003"):
+ await self.read_and_validate_current_area(step=5)
+
+ if self.check_pics("SEAR.S.A0004"):
+ await self.read_and_validate_estimated_end_time(step=6)
+
+ if self.check_pics("SEAR.S.A0005"):
+ await self.read_and_validate_progress(step=7)
+
+ if self.check_pics("SEAR.S.F02") and self.check_pics("SEAR.S.M.REMOVE_MAP"):
+ test_step = "Manually ensure the SupportedMaps attribute is not empty and that the device is not operating"
+ self.print_step("8", 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.read_and_validate_supported_maps(step=9)
+ old_supported_maps = self.mapid_list
+
+ test_step = "Manually intervene to remove one or more entries in the SupportedMaps list"
+ self.print_step("10", 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.read_and_validate_supported_maps(step=11)
+ new_supported_maps = self.mapid_list
+ asserts.assert_true(len(old_supported_maps) > len(new_supported_maps), "Failed to remove map(s)")
+
+ # NOTE the following operations are all part of step 11 - read all these attributes and check the data consistency
+ # after removing map(s)
+ await self.read_and_validate_supported_areas(step=11)
+
+ await self.read_and_validate_selected_areas(step=11)
+
+ if self.check_pics("SEAR.S.A0003"):
+ await self.read_and_validate_current_area(step=11)
+
+ if self.check_pics("SEAR.S.A0004"):
+ await self.read_and_validate_estimated_end_time(step=11)
+
+ if self.check_pics("SEAR.S.A0005"):
+ await self.read_and_validate_progress(step=11)
+
+ if self.check_pics("SEAR.S.F02") and self.check_pics("SEAR.S.M.ADD_MAP"):
+ test_step = "Manually ensure the SupportedMaps attribute has less than 255 entries and that the device is not operating"
+ self.print_step("12", 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.read_and_validate_supported_maps(step=13)
+ old_supported_maps = self.mapid_list
+
+ test_step = "Manually intervene to add one or more entries to the SupportedMaps list"
+ 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")
+
+ await self.read_and_validate_supported_maps(step=15)
+ new_supported_maps = self.mapid_list
+ asserts.assert_true(len(old_supported_maps) < len(new_supported_maps), "Failed to add map(s)")
+
+ # NOTE the following operations are all part of step 15 - read all these attributes and check the data consistency
+ # after adding map(s)
+ await self.read_and_validate_supported_areas(step=15)
+
+ await self.read_and_validate_selected_areas(step=15)
+
+ if self.check_pics("SEAR.S.A0003"):
+ await self.read_and_validate_current_area(step=15)
+
+ if self.check_pics("SEAR.S.A0004"):
+ await self.read_and_validate_estimated_end_time(step=15)
+
+ if self.check_pics("SEAR.S.A0005"):
+ await self.read_and_validate_progress(step=15)
+
+ if self.check_pics("SEAR.S.M.REMOVE_AREA"):
+ test_step = "Manually ensure the SupportedAreas attribute is not empty and that the device is not operating"
+ self.print_step("16", 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.read_and_validate_supported_areas(step=17)
+ old_supported_areas = self.areaid_list
+
+ test_step = "Manually intervene to remove one or more entries from the SupportedAreas list"
+ 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")
+
+ await self.read_and_validate_supported_areas(step=19)
+ new_supported_areas = self.areaid_list
+ asserts.assert_true(len(old_supported_areas) > len(new_supported_areas), "Failed to remove area(s)")
+
+ # NOTE the following operations are all part of step 19 - read all these attributes and check the data consistency
+ # after removing areas(s)
+
+ await self.read_and_validate_selected_areas(step=19)
+
+ if self.check_pics("SEAR.S.A0003"):
+ await self.read_and_validate_current_area(step=19)
+
+ if self.check_pics("SEAR.S.A0004"):
+ await self.read_and_validate_estimated_end_time(step=19)
+
+ if self.check_pics("SEAR.S.A0005"):
+ await self.read_and_validate_progress(step=19)
+
+ if self.check_pics("SEAR.S.M.ADD_AREA"):
+ test_step = "Manually ensure the SupportedAreas attribute has less than 255 entries and that the device is not operating"
+ self.print_step("20", 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.read_and_validate_supported_areas(step=21)
+ old_supported_areas = self.areaid_list
+
+ test_step = "Manually intervene to add one or more entries to the SupportedAreas list"
+ 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")
+
+ await self.read_and_validate_supported_areas(step=23)
+ new_supported_areas = self.areaid_list
+ asserts.assert_true(len(old_supported_areas) < len(new_supported_areas), "Failed to add area(s)")
+
+ # NOTE the following operations are all part of step 23 - read all these attributes and check the data consistency
+ # after removing areas(s)
+
+ await self.read_and_validate_selected_areas(step=23)
+
+ if self.check_pics("SEAR.S.A0003"):
+ await self.read_and_validate_current_area(step=23)
+
+ if self.check_pics("SEAR.S.A0004"):
+ await self.read_and_validate_estimated_end_time(step=23)
+
+ if self.check_pics("SEAR.S.A0005"):
+ await self.read_and_validate_progress(step=23)
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_SEAR_1_3.py b/src/python_testing/TC_SEAR_1_3.py
new file mode 100644
index 0000000..0acee1b
--- /dev/null
+++ b/src/python_testing/TC_SEAR_1_3.py
@@ -0,0 +1,155 @@
+#
+# Copyright (c) 2024 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.
+#
+
+# TODO - this was copied/pasted from abother test, it needs to be reviewed and updated
+# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
+# for details about the block below.
+#
+# === BEGIN CI TEST ARGUMENTS ===
+# test-runner-runs: run1
+# test-runner-run/run1/app: ${CHIP_RVC_APP}
+# test-runner-run/run1/factoryreset: True
+# test-runner-run/run1/quiet: True
+# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
+# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values --endpoint 1 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# === END CI TEST ARGUMENTS ===
+
+import logging
+from time import sleep
+
+import chip.clusters as Clusters
+from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
+from mobly import asserts
+
+
+class TC_SEAR_1_3(MatterBaseTest):
+ def __init__(self, *args):
+ super().__init__(*args)
+ self.endpoint = None
+ self.is_ci = False
+ self.app_pipe = "/tmp/chip_rvc_fifo_"
+
+ async def read_sear_attribute_expect_success(self, endpoint, attribute):
+ cluster = Clusters.Objects.ServiceArea
+ return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
+
+ async def read_supported_areas(self, step):
+ self.print_step(step, "Read SupportedAreas attribute")
+ supported_areas = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SupportedAreas)
+ logging.info("SupportedAreas: %s" % (supported_areas))
+
+ 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}")
+
+ return 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})")
+ ret = await self.send_single_cmd(cmd=Clusters.Objects.ServiceArea.Commands.SelectAreas(newAreas=new_areas),
+ endpoint=self.endpoint)
+
+ asserts.assert_equal(ret.commandResponseState.errorStateID,
+ expected_response,
+ f"Command response ({ret.commandResponseState}) doesn't match the expected one")
+
+ # Sends and out-of-band command to the rvc-app
+ def write_to_app_pipe(self, command):
+ with open(self.app_pipe, "w") as app_pipe:
+ app_pipe.write(command + "\n")
+ # Allow some time for the command to take effect.
+ # This removes the test flakyness which is very annoying for everyone in CI.
+ sleep(0.001)
+
+ def TC_SEAR_1_3(self) -> list[str]:
+ return ["SEAR.S"]
+
+ @async_test_body
+ async def test_TC_SEAR_1_3(self):
+ self.endpoint = self.matter_test_config.endpoint
+ asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
+ self.is_ci = self.check_pics("PICS_SDK_CI_ONLY")
+ if self.is_ci:
+ app_pid = self.matter_test_config.app_pid
+ if app_pid == 0:
+ asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set")
+ self.app_pipe = self.app_pipe + str(app_pid)
+
+ self.print_step(1, "Commissioning, already done")
+
+ # Ensure that the device is in the correct state
+ if self.is_ci:
+ self.write_to_app_pipe('{"Name": "Reset"}')
+
+ supported_area_ids = await self.read_supported_areas(step=2)
+ asserts.assert_true(len(self.supported_areas) > 0, "SupportedAreas is empty")
+ valid_area_id = supported_area_ids[0]
+ invalid_area_id = 1 + max(supported_area_ids)
+
+ duplicated_areas = [valid_area_id, valid_area_id]
+
+ # FIXME need to check if this is the correct name of this status code
+ await self.send_cmd_select_areas_expect_response(step=3, new_areas=duplicated_areas, expected_response=Clusters.ServiceArea.SelectAreasStatus.kDuplicatedAreas)
+
+ await self.send_cmd_select_areas_expect_response(step=4, new_areas=[], expected_response=Clusters.ServiceArea.SelectAreasStatus.kSuccess)
+
+ selected_areas = await self.read_selected_areas(step=5)
+ asserts.assert_true(len(selected_areas) == 0, "SelectedAreas should be empty")
+
+ await self.send_cmd_select_areas_expect_response(step=6, new_areas=[invalid_area_id], expected_response=Clusters.ServiceArea.SelectAreasStatus.kUnsupportedArea)
+
+ if self.check_pics("SEAR.S.M.INVALID_STATE_FOR_SELECT_AREAS") and self.check_pics("SEAR.S.M.HAS_MANUAL_SELAREA_STATE_CONTROL"):
+ test_step = "Manually intervene to put the device in a state that prevents it from executing the SelectAreas 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_select_areas_expect_response(step=8, new_areas=[valid_area_id], expected_response=Clusters.ServiceArea.SelectAreasStatus.kInvalidInMode)
+
+ if self.check_pics("SEAR.S.M.VALID_STATE_FOR_SELECT_AREAS") and self.check_pics("SEAR.S.M.HAS_MANUAL_SELAREA_STATE_CONTROL"):
+ test_step = f"Manually intervene to put the device in a state that allows it to execute the SelectAreas({supported_area_ids}) command"
+ self.print_step("9", 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_select_areas_expect_response(step=10, new_areas=supported_area_ids, expected_response=Clusters.ServiceArea.SelectAreasStatus.kSuccess)
+
+ selected_areas = await self.read_selected_areas(step=11)
+ asserts.assert_true(len(selected_areas) == len(supported_area_ids),
+ f"SelectedAreas({selected_areas}) should match SupportedAreas({supported_area_ids})")
+
+ await self.send_cmd_select_areas_expect_response(step=12, new_areas=supported_area_ids, expected_response=Clusters.ServiceArea.SelectAreasStatus.kSuccess)
+
+ if self.check_pics("SEAR.S.M.VALID_STATE_FOR_SELECT_AREAS") and self.check_pics("SEAR.S.M.HAS_MANUAL_SELAREA_STATE_CONTROL") and self.check_pics("SEAR.S.M.SELECT_AREAS_WHILE_NON_IDLE"):
+ test_step = f"Manually intervene to put the device in a state that allows it to execute the SelectAreas({valid_area_id}) command, and put the device in a non-idle state"
+ self.print_step("13", 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.F00"):
+ await self.send_cmd_select_areas_expect_response(step=14, new_areas=[valid_area_id], expected_response=Clusters.ServiceArea.SelectAreasStatus.kSuccess)
+ else:
+ await self.send_cmd_select_areas_expect_response(step=14, new_areas=[valid_area_id], expected_response=Clusters.ServiceArea.SelectAreasStatus.kInvalidInMode)
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_SEAR_1_4.py b/src/python_testing/TC_SEAR_1_4.py
new file mode 100644
index 0000000..a632655
--- /dev/null
+++ b/src/python_testing/TC_SEAR_1_4.py
@@ -0,0 +1,84 @@
+#
+# Copyright (c) 2024 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.
+#
+
+# TODO - this was copied/pasted from abother test, it needs to be reviewed and updated
+# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
+# for details about the block below.
+#
+# === BEGIN CI TEST ARGUMENTS ===
+# test-runner-runs: run1
+# test-runner-run/run1/app: ${CHIP_RVC_APP}
+# test-runner-run/run1/factoryreset: True
+# test-runner-run/run1/quiet: True
+# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
+# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values --endpoint 1 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# === END CI TEST ARGUMENTS ===
+
+import logging
+
+import chip.clusters as Clusters
+from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
+from mobly import asserts
+
+
+class TC_SEAR_1_4(MatterBaseTest):
+ def __init__(self, *args):
+ super().__init__(*args)
+ self.endpoint = None
+ self.is_ci = False
+ self.app_pipe = "/tmp/chip_rvc_fifo_"
+
+ async def read_sear_attribute_expect_success(self, endpoint, attribute):
+ cluster = Clusters.Objects.ServiceArea
+ return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
+
+ def TC_SEAR_1_4(self) -> list[str]:
+ return ["SEAR.S"]
+
+ @async_test_body
+ async def test_TC_SEAR_1_4(self):
+ self.endpoint = self.matter_test_config.endpoint
+ asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
+ self.is_ci = self.check_pics("PICS_SDK_CI_ONLY")
+ if self.is_ci:
+ app_pid = self.matter_test_config.app_pid
+ if app_pid == 0:
+ asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set")
+ self.app_pipe = self.app_pipe + str(app_pid)
+
+ self.print_step(1, "Commissioning, already done")
+
+ # Ensure that the device is in the correct state
+ if self.is_ci:
+ self.write_to_app_pipe('{"Name": "Reset"}')
+
+ attribute_list = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.AttributeList)
+ logging.info("AttributeList: %s" % (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))
+ asserts.assert_true(Clusters.ServiceArea.Commands.SkipArea not in cmd_list,
+ "SkipArea command should not be implemented if both CurrentArea and Progress are not")
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_SEAR_1_5.py b/src/python_testing/TC_SEAR_1_5.py
new file mode 100644
index 0000000..adcf834
--- /dev/null
+++ b/src/python_testing/TC_SEAR_1_5.py
@@ -0,0 +1,283 @@
+#
+# Copyright (c) 2024 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.
+#
+
+# TODO - this was copied/pasted from abother test, it needs to be reviewed and updated
+# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
+# for details about the block below.
+#
+# === BEGIN CI TEST ARGUMENTS ===
+# test-runner-runs: run1
+# test-runner-run/run1/app: ${CHIP_RVC_APP}
+# test-runner-run/run1/factoryreset: True
+# test-runner-run/run1/quiet: True
+# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
+# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values --endpoint 1 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# === END CI TEST ARGUMENTS ===
+
+import logging
+from time import sleep
+
+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
+
+
+class TC_SEAR_1_5(MatterBaseTest):
+ def __init__(self, *args):
+ super().__init__(*args)
+ self.endpoint = None
+ self.is_ci = False
+ self.app_pipe = "/tmp/chip_rvc_fifo_"
+
+ async def read_sear_attribute_expect_success(self, endpoint, attribute):
+ cluster = Clusters.Objects.ServiceArea
+ return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
+
+ async def read_supported_areas(self, step):
+ self.print_step(step, "Read SupportedAreas attribute")
+ supported_areas = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SupportedAreas)
+ logging.info("SupportedAreas: %s" % (supported_areas))
+
+ 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}")
+
+ return selected_areas
+
+ async def read_progress(self, step):
+ self.print_step(step, "Read Progress attribute")
+ progress = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.Progress)
+ logging.info(f"Progress {progress}")
+
+ return progress
+
+ async def read_current_area(self, step):
+ self.print_step(step, "Read CurrentArea attribute")
+ current_area = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.CurrentArea)
+ logging.info(f"CurrentArea {current_area}")
+
+ return current_area
+
+ async def send_cmd_skip_area_expect_response(self, step, skipped_area, expected_response):
+ self.print_step(step, f"Send SkipArea command with SkippedArea({skipped_area})")
+ ret = await self.send_single_cmd(cmd=Clusters.Objects.ServiceArea.Commands.SkipArea(skippedArea=skipped_area),
+ endpoint=self.endpoint)
+
+ asserts.assert_equal(ret.commandResponseState.errorStateID,
+ expected_response,
+ f"Command response ({ret.commandResponseState}) doesn't match the expected one")
+
+ # Sends and out-of-band command to the rvc-app
+
+ def write_to_app_pipe(self, command):
+ with open(self.app_pipe, "w") as app_pipe:
+ app_pipe.write(command + "\n")
+ # Allow some time for the command to take effect.
+ # This removes the test flakyness which is very annoying for everyone in CI.
+ sleep(0.001)
+
+ def TC_SEAR_1_5(self) -> list[str]:
+ return ["SEAR.S", "SEAR.S.C02.Rsp"]
+
+ @async_test_body
+ async def test_TC_SEAR_1_5(self):
+ self.endpoint = self.matter_test_config.endpoint
+ asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
+ self.is_ci = self.check_pics("PICS_SDK_CI_ONLY")
+ if self.is_ci:
+ app_pid = self.matter_test_config.app_pid
+ if app_pid == 0:
+ asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set")
+ self.app_pipe = self.app_pipe + str(app_pid)
+
+ self.print_step(1, "Commissioning, already done")
+
+ # Ensure that the device is in the correct state
+ if self.is_ci:
+ self.write_to_app_pipe('{"Name": "Reset"}')
+
+ supported_area_ids = await self.read_supported_areas(step=2)
+ asserts.assert_true(len(supported_area_ids) > 0, "SupportedAreas is empty")
+ valid_area_id = supported_area_ids[0]
+ invalid_area_id = 1 + max(supported_area_ids)
+
+ if self.check_pics("SEAR.S.M.INVALID_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 prevents it from executing the SkipArea command \
+ (e.g. set CurrentArea to null or make it not operate, i.e. be in the idle state)"
+ self.print_step("3", 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=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, \
+ if SelectedAreas wasn't empty, and SelectedAreas is empty"
+ self.print_step("5", 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=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)
+
+ 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")
+
+ 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)
+ 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)")
+
+ prog_areas = [p.areaID for p in new_progress_list]
+
+ asserts.assert_true(old_current_area in prog_areas, f"Progress should include area {old_current_area}")
+
+ 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")
+
+ # Verify that if the device hasn't stopped operating, the `new_progress_list`'s entry matching `new_current_area` shows the Operating status
+ 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
+
+ # next, we need to check for something else (so the condition of the 'if' above becomes part of the 'then' statement below):
+ # if before skipping all areas, except the current one, were Skipped or Completed, the device MUST have stopped operating
+ 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 not self.check_pics("SEAR.S.A0005"):
+ 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")
+
+ 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:
+ if p.status in (Clusters.ServiceArea.OperationalStatusEnum.kPending,
+ Clusters.ServiceArea.OperationalStatusEnum.kOperating):
+ area_to_skip = p.areaID
+ break
+
+ 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)
+
+ if self.check_pics("SEAR.S.M.HAS_MANUAL_SKIP_STATE_CONTROL"):
+ 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")
+
+ 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
+
+ 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()
diff --git a/src/python_testing/TC_SEAR_1_6.py b/src/python_testing/TC_SEAR_1_6.py
new file mode 100644
index 0000000..02a0fcd
--- /dev/null
+++ b/src/python_testing/TC_SEAR_1_6.py
@@ -0,0 +1,148 @@
+#
+# Copyright (c) 2024 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.
+#
+
+# TODO - this was copied/pasted from abother test, it needs to be reviewed and updated
+# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
+# for details about the block below.
+#
+# === BEGIN CI TEST ARGUMENTS ===
+# test-runner-runs: run1
+# test-runner-run/run1/app: ${CHIP_RVC_APP}
+# test-runner-run/run1/factoryreset: True
+# test-runner-run/run1/quiet: True
+# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
+# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS examples/rvc-app/rvc-common/pics/rvc-app-pics-values --endpoint 1 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# === END CI TEST ARGUMENTS ===
+
+import logging
+from time import sleep
+
+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
+
+
+class TC_SEAR_1_6(MatterBaseTest):
+ def __init__(self, *args):
+ super().__init__(*args)
+ self.endpoint = None
+ self.is_ci = False
+ self.app_pipe = "/tmp/chip_rvc_fifo_"
+
+ async def read_sear_attribute_expect_success(self, endpoint, attribute):
+ cluster = Clusters.Objects.ServiceArea
+ return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
+
+ async def read_supported_areas(self, step):
+ self.print_step(step, "Read SupportedAreas attribute")
+ supported_areas = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.SupportedAreas)
+ logging.info("SupportedAreas: %s" % (supported_areas))
+
+ 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}")
+
+ return selected_areas
+
+ async def read_progress(self, step):
+ self.print_step(step, "Read Progress attribute")
+ progress = await self.read_sear_attribute_expect_success(
+ endpoint=self.endpoint, attribute=Clusters.ServiceArea.Attributes.Progress)
+ logging.info(f"Progress {progress}")
+
+ return progress
+
+ # Sends and out-of-band command to the rvc-app
+ def write_to_app_pipe(self, command):
+ with open(self.app_pipe, "w") as app_pipe:
+ app_pipe.write(command + "\n")
+ # Allow some time for the command to take effect.
+ # This removes the test flakyness which is very annoying for everyone in CI.
+ sleep(0.001)
+
+ def TC_SEAR_1_6(self) -> list[str]:
+ return ["SEAR.S", "SEAR.S.A0005", "SEAR.S.A0000", "SEAR.S.A0002", "SEAR.S.M.HAS_MANUAL_OPERATING_STATE_CONTROL"]
+
+ @async_test_body
+ async def test_TC_SEAR_1_6(self):
+ self.endpoint = self.matter_test_config.endpoint
+ asserts.assert_false(self.endpoint is None, "--endpoint <endpoint> must be included on the command line in.")
+ self.is_ci = self.check_pics("PICS_SDK_CI_ONLY")
+ if self.is_ci:
+ app_pid = self.matter_test_config.app_pid
+ if app_pid == 0:
+ asserts.fail("The --app-pid flag must be set when PICS_SDK_CI_ONLY is set")
+ self.app_pipe = self.app_pipe + str(app_pid)
+
+ self.print_step(1, "Commissioning, already done")
+
+ # Ensure that the device is in the correct state
+ if self.is_ci:
+ self.write_to_app_pipe('{"Name": "Reset"}')
+
+ test_step = "Manually intervene to put the device in the idle state and ensure SupportedAreas and SelectedAreas are not empty"
+ self.print_step("2", test_step)
+ if not self.is_ci:
+ self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
+
+ supported_area_ids = await self.read_supported_areas(step=3)
+ asserts.assert_true(len(supported_area_ids) > 0, "SupportedAreas is empty")
+
+ selected_areas = await self.read_selected_areas(step=4)
+ asserts.assert_true(len(selected_areas) > 0, "SelectedAreas is empty")
+
+ test_step = "Manually intervene to put the device in the operating state"
+ self.print_step("5", test_step)
+ if not self.is_ci:
+ self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
+
+ progress_list_operating = await self.read_progress(step=6)
+ asserts.assert_true(len(selected_areas) == len(progress_list_operating),
+ f"len of SelectedAreas({len(selected_areas)}) should be equal to len of Progress({len(progress_list_operating)})")
+
+ for p in progress_list_operating:
+ asserts.assert_true(p.areaID in selected_areas, f"Progress entry with unknown AreaID({p.areaID})")
+ asserts.assert_true(p.status in (Clusters.ServiceArea.OperationalStatusEnum.kPending,
+ Clusters.ServiceArea.OperationalStatusEnum.kOperating),
+ f"Progress entry with unexpected Status({p.status})")
+ asserts.assert_true(p.TotalOperationalTime is NullValue, "Progress entry with non-null TotalOperationalTime")
+
+ test_step = "While all entries in Progress show the Pending or Operating status (i.e. \
+ before any area is skipped or completed), manually intervene to put the device \
+ in the idle state, by ending the operation unexpectedly (e.g. force an error)"
+ 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")
+
+ progress_list_idle = await self.read_progress(step=8)
+ asserts.assert_true(len(selected_areas) == len(progress_list_idle),
+ f"len of SelectedAreas({len(selected_areas)}) should be equal to len of Progress({len(progress_list_idle)})")
+
+ for p in progress_list_idle:
+ asserts.assert_true(p.areaID in selected_areas, f"Progress entry with unknown AreaID({p.areaID})")
+ asserts.assert_true(p.status == Clusters.ServiceArea.OperationalStatusEnum.kSkipped,
+ f"Progress entry with unexpected Status({p.status})")
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
