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 00000000000000..f83ab82d38ab69 --- /dev/null +++ b/src/python_testing/TC_SEAR_1_2.py @@ -0,0 +1,384 @@ +# +# 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_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 = [] + for m in supported_maps: + if m.mapID in mapid_list: + asserts.fail("SupportedMaps must have unique MapID values!") + else: + mapid_list.append(m.mapID) + name_list = [] + for m in supported_maps: + if m.name in name_list: + asserts.fail("SupportedMaps must have unique Name values!") + else: + name_list.append(m.name) + #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 = [] + for a in supported_areas: + if a.areaID in areaid_list: + asserts.fail("SupportedAreas must have unique AreaID values!") + else: + 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})") + areadesc_same_map_list = [] + areadesc_same_map_list.append(a.areaDesc) + for b in supported_areas: + if a.mapID == b.mapID: + if b.areaDesc in areadesc_same_map_list: + asserts.fail(f"SupportedAreas must have unique MapID({a.mapID}) + AreaDesc({a.areaDesc}) values!") + else: + areadesc_same_map_list.append(b.areaDesc) + else: + #empty SupportedMaps + asserts.assert_is(a.mapID, NullValue, + f"SupportedAreas entry with AreaID({a.areaID}) should have null MapID") + areadesc_list = [] + areadesc_list.append(a.areaDesc) + for b in supported_areas: + if b.areaDesc in areadesc_list: + asserts.fail(f"SupportedAreas must have unique AreaDesc({a.areaDesc}) values!") + else: + areadesc_list.append(b.areaDesc) + 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 in range(0, 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") + + selareaid_list = [] + for a in selected_areas: + if a.areaID in selareaid_list: + asserts.fail("SelectedAreas must have unique AreaID values!") + else: + selareaid_list.append(a.areaID) + asserts.assert_true(a.areaID in self.areaid_list, + f"SelectedAreas entry {a.areaID} has invalid value") + #save so other methods can use this if neeeded + self.selareaid_list = selareaid_list + + 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, + f"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(self.selected_areas) <= 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 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(2) + + await self.read_and_validate_supported_areas(3) + + await self.read_and_validate_selected_areas(4) + + if self.check_pics("SEAR.S.A0003"): + await self.read_and_validate_current_area(5) + + if self.check_pics("SEAR.S.A0004"): + await self.read_and_validate_estimated_end_time(6) + + if self.check_pics("SEAR.S.A0005"): + await self.read_and_validate_progress(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(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(11) + new_supported_maps = self.mapid_list + asserts.assert_true(len(old_supported_maps > len(new_supported_maps)), f"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(11) + + await self.read_and_validate_selected_areas(11) + + if self.check_pics("SEAR.S.A0003"): + await self.read_and_validate_current_area(11) + + if self.check_pics("SEAR.S.A0004"): + await self.read_and_validate_estimated_end_time(11) + + if self.check_pics("SEAR.S.A0005"): + await self.read_and_validate_progress(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(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(15) + new_supported_maps = self.mapid_list + asserts.assert_true(len(old_supported_maps < len(new_supported_maps)), f"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(15) + + await self.read_and_validate_selected_areas(15) + + if self.check_pics("SEAR.S.A0003"): + await self.read_and_validate_current_area(15) + + if self.check_pics("SEAR.S.A0004"): + await self.read_and_validate_estimated_end_time(15) + + if self.check_pics("SEAR.S.A0005"): + await self.read_and_validate_progress(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(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(19) + new_supported_areas = self.areaid_list + asserts.assert_true(len(old_supported_areas > len(new_supported_areas)), f"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(19) + + if self.check_pics("SEAR.S.A0003"): + await self.read_and_validate_current_area(19) + + if self.check_pics("SEAR.S.A0004"): + await self.read_and_validate_estimated_end_time(19) + + if self.check_pics("SEAR.S.A0005"): + await self.read_and_validate_progress(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(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(23) + new_supported_areas = self.areaid_list + asserts.assert_true(len(old_supported_areas < len(new_supported_areas)), f"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(23) + + if self.check_pics("SEAR.S.A0003"): + await self.read_and_validate_current_area(23) + + if self.check_pics("SEAR.S.A0004"): + await self.read_and_validate_estimated_end_time(23) + + if self.check_pics("SEAR.S.A0005"): + await self.read_and_validate_progress(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 00000000000000..bdc133f1fa7103 --- /dev/null +++ b/src/python_testing/TC_SEAR_1_3.py @@ -0,0 +1,163 @@ +# +# 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_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)) + + areaid_list = [] + for a in supported_areas: + areaid_list.append(a.areaID) + return areaid_list + + 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 + + + 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 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(2) + asserts.assert_true(len(self.supported_areas > 0), f"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(3, duplicated_areas, Clusters.ServiceArea.SelectAreasStatus.kDuplicatedAreas) + + await self.send_cmd_select_areas_expect_response(4, [], Clusters.ServiceArea.SelectAreasStatus.kSuccess) + + selected_areas = await self.read_selected_areas(5) + asserts.assert_true(len(selected_areas == 0), f"SelectedAreas should be empty") + + await self.send_cmd_select_areas_expect_response(6, [invalid_area_id], 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(8, [valid_area_id], 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(10, supported_area_ids, Clusters.ServiceArea.SelectAreasStatus.kSuccess) + + selected_areas = await self.read_selected_areas(11) + asserts.assert_true(len(selected_areas == supported_area_ids), f"SelectedAreas({selected_areas}) should match SupportedAreas({supported_area_ids})") + + await self.send_cmd_select_areas_expect_response(12, supported_area_ids, 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(14, [valid_area_id], Clusters.ServiceArea.SelectAreasStatus.kSuccess) + else: + await self.send_cmd_select_areas_expect_response(14, [valid_area_id], 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 00000000000000..6f30423969618e --- /dev/null +++ b/src/python_testing/TC_SEAR_1_4.py @@ -0,0 +1,83 @@ +# +# 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_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 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 not (Clusters.ServiceArea.Attributes.CurrentArea in attribute_list and Clusters.ServiceArea.Attributes.Progress 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, f"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_6.py b/src/python_testing/TC_SEAR_1_6.py new file mode 100644 index 00000000000000..b3c2c117474949 --- /dev/null +++ b/src/python_testing/TC_SEAR_1_6.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 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)) + + areaid_list = [] + for a in supported_areas: + areaid_list.append(a.areaID) + return areaid_list + + 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 + + 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.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 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"}') + + #FIXME is this necesssary? I'm not sure what TC_SEAR_1_6() is used for + if not (self.check_pics("SEAR.S.A0005") and self.check_pics("SEAR.S.M.HAS_MANUAL_OPERATING_STATE_CONTROL")): + return + + 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(3) + asserts.assert_true(len(supported_area_ids) > 0, f"SupportedAreas is empty") + + selected_areas = await self.read_selected_areas(4) + asserts.assert_true(len(selected_areas) > 0, f"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(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, f"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(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()