From d807f5e08f238e33d4e077ca8e1d02053aa14148 Mon Sep 17 00:00:00 2001 From: LUCAS BECKER Date: Sun, 11 Feb 2024 23:15:17 -0300 Subject: [PATCH] Added test 2.4 --- .github/workflows/tests.yaml | 1 + src/python_testing/TC_OVENOPSTATE_2_4.py | 50 +++++ src/python_testing/TC_OVENOPSTATE_2_5.py | 238 ----------------------- src/python_testing/TC_OpstateCommon.py | 123 +++++++++--- 4 files changed, 145 insertions(+), 267 deletions(-) create mode 100644 src/python_testing/TC_OVENOPSTATE_2_4.py delete mode 100644 src/python_testing/TC_OVENOPSTATE_2_5.py diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index d6e103ce4e69ef..96677011889014 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -516,6 +516,7 @@ jobs: scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_OVENOPSTATE_2_1.py" --script-args "--endpoint 1 --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"' scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_OVENOPSTATE_2_2.py" --script-args "--endpoint 1 --int-arg PIXIT.WAITTIME:5 --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"' scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_OVENOPSTATE_2_3.py" --script-args "--endpoint 1 --int-arg PIXIT.WAITTIME:5 --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"' + scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app --factoryreset --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_OVENOPSTATE_2_4.py" --script-args "--endpoint 1 --int-arg PIXIT.OVENOPSTATE.ErrorEventGen:1 --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"' - name: Uploading core files uses: actions/upload-artifact@v4 if: ${{ failure() && !env.ACT }} diff --git a/src/python_testing/TC_OVENOPSTATE_2_4.py b/src/python_testing/TC_OVENOPSTATE_2_4.py new file mode 100644 index 00000000000000..11edc463e3162e --- /dev/null +++ b/src/python_testing/TC_OVENOPSTATE_2_4.py @@ -0,0 +1,50 @@ +# +# Copyright (c) 2023 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. +# + + +import chip.clusters as Clusters +from matter_testing_support import TestStep, MatterBaseTest, async_test_body, default_matter_test_main +from TC_OpstateCommon import TC_OVENOPSTATE_BASE, TestInfo + + +class TC_OVENOPSTATE_2_4(MatterBaseTest, TC_OVENOPSTATE_BASE): + def __init__(self, *args): + super().__init__(*args) + + test_info = TestInfo( + pics_code="OVENOPSTATE", + cluster=Clusters.OvenCavityOperationalState + ) + + super().setup_base(test_info=test_info) + + def steps_TC_OVENOPSTATE_2_4(self) -> list[TestStep]: + return self.STEPS_TC_OPSTATE_BASE_2_4() + + def pics_TC_OVENOPSTATE_2_4(self) -> list[str]: + return ["OVENOPSTATE.S"] + + @async_test_body + async def test_TC_OVENOPSTATE_2_4(self): + endpoint = self.matter_test_config.endpoint + + await self.TEST_TC_OPSTATE_BASE_2_4(endpoint=endpoint) + + +if __name__ == "__main__": + default_matter_test_main() + diff --git a/src/python_testing/TC_OVENOPSTATE_2_5.py b/src/python_testing/TC_OVENOPSTATE_2_5.py deleted file mode 100644 index a54d1da320a3ec..00000000000000 --- a/src/python_testing/TC_OVENOPSTATE_2_5.py +++ /dev/null @@ -1,238 +0,0 @@ -# -# Copyright (c) 2023 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. -# - -import logging -import time - -import chip.clusters as Clusters -from chip.clusters.Types import NullValue -from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, type_matches -from mobly import asserts - -# This test requires several additional command line arguments -# run with -# --int-arg PIXIT_ENDPOINT: - -class TC_OVENOPSTATE_2_5(MatterBaseTest): - - async def read_mod_attribute_expect_success(self, endpoint, attribute): - cluster = Clusters.Objects.OvenCavityOperationalState - return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute) - - async def send_start_cmd(self) -> Clusters.Objects.OvenCavityOperationalState.Commands.Start: - ret = await self.send_single_cmd(cmd=Clusters.Objects.OvenCavityOperationalState.Commands.Start(), endpoint=self.endpoint) - asserts.assert_true(type_matches(ret, Clusters.Objects.OvenCavityOperationalState.Commands.OperationalCommandResponse), - "Unexpected return type for Start") - return ret - - async def send_pause_cmd(self) -> Clusters.Objects.OvenCavityOperationalState.Commands.Pause: - ret = await self.send_single_cmd(cmd=Clusters.Objects.OvenCavityOperationalState.Commands.Pause(), endpoint=self.endpoint) - asserts.assert_true(type_matches(ret, Clusters.Objects.OvenCavityOperationalState.Commands.OperationalCommandResponse), - "Unexpected return type for Pause") - return ret - - async def send_resume_cmd(self) -> Clusters.Objects.OvenCavityOperationalState.Commands.Resume: - ret = await self.send_single_cmd(cmd=Clusters.Objects.OvenCavityOperationalState.Commands.Resume(), endpoint=self.endpoint) - asserts.assert_true(type_matches(ret, Clusters.Objects.OvenCavityOperationalState.Commands.OperationalCommandResponse), - "Unexpected return type for Resume") - return ret - - async def send_stop_cmd(self) -> Clusters.Objects.OvenCavityOperationalState.Commands.Stop: - ret = await self.send_single_cmd(cmd=Clusters.Objects.OvenCavityOperationalState.Commands.Stop(), endpoint=self.endpoint) - asserts.assert_true(type_matches(ret, Clusters.Objects.OvenCavityOperationalState.Commands.OperationalCommandResponse), - "Unexpected return type for Stop") - return ret - - async def read_and_validate_opstate(self, step, expected_state): - self.print_step(step, "Read OperationalState attribute") - operational_state = await self.read_mod_attribute_expect_success( - endpoint=self.endpoint, attribute=Clusters.OvenCavityOperationalState.Attributes.OperationalState) - logging.info("OperationalState: %s" % (operational_state)) - asserts.assert_equal(operational_state, expected_state, - "OperationalState(%s) should equal %s" % (operational_state, expected_state)) - - async def read_and_validate_operror(self, step, expected_error): - self.print_step(step, "Read OperationalError attribute") - operational_error = await self.read_mod_attribute_expect_success( - endpoint=self.endpoint, attribute=Clusters.OvenCavityOperationalState.Attributes.OperationalError) - logging.info("OperationalError: %s" % (operational_error)) - asserts.assert_equal(operational_error.errorStateID, expected_error, - "errorStateID(%s) should equal %s" % (operational_error.errorStateID, expected_error)) - - @async_test_body - async def test_TC_OVENOPSTATE_2_5(self): - - asserts.assert_true('PIXIT_ENDPOINT' in self.matter_test_config.global_test_params, - "PIXIT_ENDPOINT must be included on the command line in " - "the --int-arg flag as PIXIT_ENDPOINT:") - - self.endpoint = self.matter_test_config.global_test_params['PIXIT_ENDPOINT'] - - attributes = Clusters.OvenCavityOperationalState.Attributes - - - self.print_step(1, "Commissioning, already done") - - # step 2 set subscription to the OperationCompletion event - - if self.check_pics("OVENOPSTATE.A0004"): - self.print_step(3, "Read operationalState Attribute") - operational_state=await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.OperationalState) - logging.info("OperationalState: %s" % (operational_state)) - asserts.assert_equal(operational_state, Clusters.OvenCavityOperationalState.Enums.OperationalStateEnum.kNoError, - "OperationalState ID should be NoError(0x00)") - - self.print_step(4, "Manually put the DUT into a state wherein it can receive a Start Command") - input("press enter when done, \n") - - if self.check_pics("OVENOPSTATE.S.A0002"): - self.print_step(5, "Read CountdownTime attribute") - initial_countdown_time = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.CountdownTime) - logging.info("CountdownTime: %s" % (initial_countdown_time)) - if initial_countdown_time is not NullValue: - in_range = (1 > initial_countdown_time <= 3600) - in_range1=(initial_countdown_time > "PIXIT.COUNTDOWN.THRESHOLD") - - asserts.assert_true(initial_countdown_time is NullValue or in_range or in_range1, - "invalid CountdownTime(%s). Must be in between 1 and 1 hour, or null " % initial_countdown_time) - - if self.check_pics("OVENOPSTATE.S.C02.Rsp" and "OVENOPSTATE.S.C04.Tx"): - self.print_step(6, "Send Start command") - ret = await self.send_start_cmd() - asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OvenCavityOperationalState.Enums.ErrorStateEnum.kNoError, - "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) - - self.print_step(7, "TH Waits for PIXIT.WAITTIME") - time.sleep("PIXIT.WAITTIME") - - if self.check_pics("OVENOPSTATE.S.A0004"): - self.print_step(8, "Read operationalState Attribute") - operational_state=await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.OperationalState) - logging.info("OperationalState: %s" % (operational_state)) - asserts.assert_equal(operational_state, Clusters.OvenCavityOperationalState.Enums.OperationalStateEnum.kRunning, - "OperationalState ID should be Running(0x01)") - - self.print_step(9, "TH Waits for initial_countdown_time ") - time.sleep(initial_countdown_time) - - if self.check_pics("OVENOPSTATE.S.C01.Rsp" and "OVENOPSTATE.S.C04.Tx"): - self.print_step(10, "Send Stop command") - ret = await self.send_stop_cmd() - asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OvenCavityOperationalState.Enums.ErrorStateEnum.kNoError, - "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) - - self.print_step(11, "Operationalcompletion") - - - if self.check_pics("OVENOPSTATE.S.A0004"): - self.print_step(12, "Read operationalState Attribute") - operational_state=await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.OperationalState) - logging.info("OperationalState: %s" % (operational_state)) - asserts.assert_equal(operational_state, Clusters.OvenCavityOperationalState.Enums.OperationalStateEnum.kStopped, - "OperationalState ID should be Stopped (0x00)") - - self.print_step(13, "Manually RESTART THE DUT") - input("press enter when done, \n") - if self.check_pics("OVENOPSTATE.S.A0002"): - self.print_step(13, "Read CountdownTime attribute") - initial_countdown_time = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.CountdownTime) - logging.info("CountdownTime: %s" % (initial_countdown_time)) - if initial_countdown_time is not NullValue: - in_range = (1 > initial_countdown_time <= 3600) - in_range1=(initial_countdown_time > "PIXIT.COUNTDOWN.THRESHOLD") - - if self.check_pics("OVENOPSTATE.S.C02.Rsp" and "OVENOPSTATE.S.C04.Tx"): - self.print_step(14, "Send Start command") - ret = await self.send_start_cmd() - asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OvenCavityOperationalState.Enums.ErrorStateEnum.kNoError, - "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) - - self.print_step(15, "TH Waits for PIXIT.WAITTIME") - time.sleep("PIXIT.WAITTIME") - - if self.check_pics("OVENOPSTATE.S.A0004"): - self.print_step(16, "Read operationalState Attribute") - operational_state=await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.OperationalState) - logging.info("OperationalState: %s" % (operational_state)) - asserts.assert_equal(operational_state, Clusters.OvenCavityOperationalState.Enums.OperationalStateEnum.kRunning, - "OperationalState ID should be Running (0x01)") - - if self.check_pics("OVENOPSTATE.S.C00.Rsp" and "OVENOPSTATE.S.C04.Tx"): - self.print_step(17, "Send Pause command") - ret = await self.send_pause_cmd() - asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OvenCavityOperationalState.Enums.ErrorStateEnum.kNoError, - "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) - - if self.check_pics("OVENOPSTATE.S.A0004"): - self.print_step(18, "Read operationalState Attribute") - operational_state=await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.OperationalState) - logging.info("OperationalState: %s" % (operational_state)) - asserts.assert_equal(operational_state, Clusters.OvenCavityOperationalState.Enums.OperationalStateEnum.kPaused, - "OperationalState ID should be paused (0x02)") - - self.print_step(19, "TH Waits for half of initial_countdown_time ") - time.sleep(initial_countdown_time//2) - - if self.check_pics("OVENOPSTATE.S.C03.Rsp" and "OVENOPSTATE.S.C04.Tx"): - self.print_step(20, "Send Resume command") - ret = await self.send_resume_cmd() - asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OvenCavityOperationalState.Enums.ErrorStateEnum.kNoError, - "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) - - if self.check_pics("OVENOPSTATE.S.A0004"): - self.print_step(21, "Read operationalState Attribute") - operational_state=await self.read_mod_attribute_expect_success(endpoint=self.endpoint, - attribute=attributes.OperationalState) - logging.info("OperationalState: %s" % (operational_state)) - asserts.assert_equal(operational_state, Clusters.OvenCavityOperationalState.Enums.OperationalStateEnum.kRunning, - "OperationalState ID should be Running(0x01)") - - self.print_step(22, "TH Waits for initial_countdown_time") - time.sleep(initial_countdown_time) - - if self.check_pics("OVENOPSTATE.S.C01.Rsp" and "OVENOPSTATE.S.C04.Tx"): - self.print_step(23, "Send Stop command") - ret = await self.send_stop_cmd() - asserts.assert_equal(ret.commandResponseState.errorStateID, Clusters.OvenCavityOperationalState.Enums.ErrorStateEnum.kNoError, - "errorStateID(%s) should be NoError(0x00)" % ret.commandResponseState.errorStateID) - - self.print_step(24, "Operationalcompletion") - -if __name__ == "__main__": - default_matter_test_main() - - - - - - - - - - - - - - - diff --git a/src/python_testing/TC_OpstateCommon.py b/src/python_testing/TC_OpstateCommon.py index d52c484df9cf58..5a3d78e5699744 100644 --- a/src/python_testing/TC_OpstateCommon.py +++ b/src/python_testing/TC_OpstateCommon.py @@ -23,7 +23,7 @@ from chip.interaction_model import InteractionModelError from mobly import asserts from dataclasses import dataclass -from matter_testing_support import TestStep +from matter_testing_support import TestStep, EventChangeCallback import psutil @@ -222,20 +222,20 @@ async def TEST_TC_OPSTATE_BASE_1_1(self, endpoint=1, cluster_revision=1, feature self.step(6) expected_value = [] - if self.check_pics(("%s.S.C00.Rsp" % self.test_info.pics_code)) or \ - self.check_pics(("%s.S.C03.Rsp" % self.test_info.pics_code)): - expected_value.append(commands.Pause.command_id) + if (self.check_pics(("%s.S.C00.Rsp" % self.test_info.pics_code)) or + self.check_pics(("%s.S.C03.Rsp" % self.test_info.pics_code))): + expected_value.append(commands.Pause.command_id) - if self.check_pics(("%s.S.C01.Rsp" % self.test_info.pics_code)) or \ - self.check_pics(("%s.S.C02.Rsp" % self.test_info.pics_code)): - expected_value.append(commands.Stop.command_id) + if (self.check_pics(("%s.S.C01.Rsp" % self.test_info.pics_code)) or \ + self.check_pics(("%s.S.C02.Rsp" % self.test_info.pics_code))): + expected_value.append(commands.Stop.command_id) if self.check_pics(("%s.S.C02.Rsp" % self.test_info.pics_code)): expected_value.append(commands.Start.command_id) - if self.check_pics(("%s.S.C03.Rsp" % self.test_info.pics_code)) or \ - self.check_pics(("%s.S.C00.Rsp" % self.test_info.pics_code)): - expected_value.append(commands.Resume.command_id) + if (self.check_pics(("%s.S.C03.Rsp" % self.test_info.pics_code)) or + self.check_pics(("%s.S.C00.Rsp" % self.test_info.pics_code))): + expected_value.append(commands.Resume.command_id) await self.read_and_expect_array_contains(endpoint=endpoint, attribute=attributes.AcceptedCommandList, @@ -245,10 +245,10 @@ async def TEST_TC_OPSTATE_BASE_1_1(self, endpoint=1, cluster_revision=1, feature self.step(7) expected_value = [] - if self.check_pics(("%s.S.C00.Rsp" % self.test_info.pics_code)) or \ - self.check_pics(("%s.S.C01.Rsp" % self.test_info.pics_code)) or \ - self.check_pics(("%s.S.C02.Rsp" % self.test_info.pics_code)) or \ - self.check_pics(("%s.S.C03.Rsp" % self.test_info.pics_code)): + if (self.check_pics(("%s.S.C00.Rsp" % self.test_info.pics_code)) or + self.check_pics(("%s.S.C01.Rsp" % self.test_info.pics_code)) or + self.check_pics(("%s.S.C02.Rsp" % self.test_info.pics_code)) or + self.check_pics(("%s.S.C03.Rsp" % self.test_info.pics_code))): expected_value.append(commands.OperationalCommandResponse.command_id) await self.read_and_expect_array_contains(endpoint=endpoint, @@ -542,9 +542,9 @@ async def TEST_TC_OPSTATE_BASE_2_2(self, endpoint=1): self.step(4) if self.pics_guard(self.check_pics("%s.S.C02.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): - ret = await self.send_cmd_expect_response(endpoint=endpoint, - cmd=commands.Start(), - expected_response=cluster.Enums.ErrorStateEnum.kNoError) + await self.send_cmd_expect_response(endpoint=endpoint, + cmd=commands.Start(), + expected_response=cluster.Enums.ErrorStateEnum.kNoError) # STEP 5: TH reads from the DUT the OperationalState attribute self.step(5) @@ -619,7 +619,7 @@ async def TEST_TC_OPSTATE_BASE_2_2(self, endpoint=1): # STEP 13: TH sends Stop command to the DUT self.step(13) if self.pics_guard(self.check_pics("%s.S.C01.Rsp" % self.test_info.pics_code) and - self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): + self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Stop(), expected_response=cluster.Enums.ErrorStateEnum.kNoError) @@ -682,7 +682,6 @@ def STEPS_TC_OPSTATE_BASE_2_3(self) -> list[TestStep]: async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): cluster = self.test_info.cluster attributes = cluster.Attributes - events = cluster.Events commands = cluster.Commands self.init_test() @@ -702,7 +701,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 2: Manually put the DUT into a state wherein it can receive a Pause Command self.step(2) self.send_manual_or_pipe_command('{"Name":"OperationalStateChange", "Device":"%s", "Operation":"OnFault", "Param": %s}' % - (self.device, cluster.Enums.ErrorStateEnum.kNoError)) + (self.device, cluster.Enums.ErrorStateEnum.kNoError)) time.sleep(1) self.send_manual_or_pipe_command('{"Name":"OperationalStateChange", "Device":"%s", "Operation":"Start"}' % self.device) @@ -723,7 +722,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 4: TH sends Pause command to the DUT self.step(4) - if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Pause(), @@ -757,11 +756,11 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): if (countdown_time is not NullValue) and (initial_countdown_time is not NullValue): asserts.assert_equal(countdown_time, initial_countdown_time, - "The countdown time shall be equal since pause command") + "The countdown time shall be equal since pause command") # STEP 9: TH sends Pause command to the DUT self.step(9) - if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Pause(), @@ -769,7 +768,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 10: TH sends Resume command to the DUT self.step(10) - if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Resume(), @@ -784,7 +783,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 12: TH sends Resume command to the DUT self.step(12) - if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Resume(), @@ -797,7 +796,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 14: TH sends Pause command to the DUT self.step(14) - if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Pause(), @@ -805,7 +804,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 15: TH sends Resume command to the DUT self.step(15) - if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Resume(), @@ -818,7 +817,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 17: TH sends Pause command to the DUT self.step(17) - if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C00.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Pause(), @@ -826,9 +825,75 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1): # STEP 18: TH sends Resume command to the DUT self.step(18) - if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and \ + if self.pics_guard(self.check_pics("%s.S.C03.Rsp" % self.test_info.pics_code) and self.check_pics("%s.S.C04.Tx" % self.test_info.pics_code)): await self.send_cmd_expect_response(endpoint=endpoint, cmd=commands.Resume(), expected_response=cluster.Enums.ErrorStateEnum.kCommandInvalidInState) + ############################ + # TEST CASE 2.4 + ############################ + def STEPS_TC_OPSTATE_BASE_2_4(self) -> list[TestStep]: + steps = [TestStep(1, "Commissioning, already done", is_commissioning=True), + TestStep(2, "Set up a subscription to the OperationalError event"), + TestStep(3, "At the DUT take the vendor defined action to generate an OperationalError event"), + TestStep(4, "TH reads from the DUT the OperationalState attribute") + ] + return steps + + async def TEST_TC_OPSTATE_BASE_2_4(self, endpoint=1): + cluster = self.test_info.cluster + attributes = cluster.Attributes + events = cluster.Events + commands = cluster.Commands + + self.init_test() + + pixit_var_name = 'PIXIT.%s.ErrorEventGen' % self.test_info.pics_code + print(pixit_var_name in self.matter_test_config.global_test_params) + asserts.assert_true(pixit_var_name in self.matter_test_config.global_test_params, + "%s must be included on the command line in " + "the --int-arg flag as %s:<0 or 1>" % (pixit_var_name, pixit_var_name)) + + error_event_gen = self.matter_test_config.global_test_params[pixit_var_name] + + # STEP 1: Commission DUT to TH (can be skipped if done in a preceding test) + self.step(1) + + if self.pics_guard(error_event_gen): + # STEP 2: Set up a subscription to the OperationalError event + self.step(2) + # Subscribe to Events and when they are sent push them to a queue for checking later + events_callback = EventChangeCallback(cluster) + await events_callback.start(self.default_controller, + self.dut_node_id, + endpoint) + + # STEP 3: At the DUT take the vendor defined action to generate an OperationalError event + self.step(3) + self.send_manual_or_pipe_command('{"Name":"OperationalStateChange", "Device":"%s", "Operation":"OnFault", "Param": 1}' % self.device) + event_data = events_callback.wait_for_event_report(cluster.Events.OperationalError).errorState + + # Defined Errors + defined_errors = [error.value for error in cluster.Enums.ErrorStateEnum + if (error is not cluster.Enums.ErrorStateEnum.kUnknownEnumValue or + error is not cluster.Enums.ErrorStateEnum.kNoError)] + + in_range = (0x80 <= event_data.errorStateID <= 0xBF) + asserts.assert_true(event_data.errorStateID in defined_errors + or in_range, "Event has an invalid ID value!") + if in_range: + asserts.assert_true(event_data.errorStateLabel is not None, "ErrorStateLabel should be populated") + + # STEP 4: TH reads from the DUT the OperationalState attribute + self.step(4) + if self.pics_guard(self.check_pics("%s.S.A0004" % self.test_info.pics_code)): + await self.read_and_expect_value(endpoint=endpoint, + attribute=attributes.OperationalState, + expected_value=cluster.Enums.OperationalStateEnum.kError) + else: + self.skip_step(2) + self.skip_step(3) + self.skip_step(4) +