Add OvenOpState 2.6

.github/workflows/tests.yaml

@@ -587,6 +587,7 @@ jobs:
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_OVENOPSTATE_2_3.py'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_OVENOPSTATE_2_4.py'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_OVENOPSTATE_2_5.py'
+                  scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_OVENOPSTATE_2_6.py'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_MWOCTRL_2_1.py'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_MWOCTRL_2_2.py'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_MWOCTRL_2_4.py'

src/python_testing/TC_OVENOPSTATE_2_6.py

@@ -0,0 +1,62 @@
+#
+#    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.
+#
+
+# 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: ${ALL_CLUSTERS_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: --endpoint 1 --int-arg PIXIT.WAITTIME.REBOOT: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:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# === END CI TEST ARGUMENTS ===
+
+
+import chip.clusters as Clusters
+from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from TC_OpstateCommon import TC_OPSTATE_BASE, TestInfo
+
+
+class TC_OVENOPSTATE_2_6(MatterBaseTest, TC_OPSTATE_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_6(self) -> list[TestStep]:
+        return self.steps_TC_OPSTATE_BASE_2_6()
+
+    def pics_TC_OVENOPSTATE_2_6(self) -> list[str]:
+        return ["OVENOPSTATE.S", "OVENOPSTATE.S.A0002"]
+
+    @async_test_body
+    async def test_TC_OVENOPSTATE_2_6(self):
+        # endpoint = self.matter_test_config.endpoint
+
+        # await self.TEST_TC_OPSTATE_BASE_2_6(endpoint=endpoint)
+        await self.TEST_TC_OPSTATE_BASE_2_6(endpoint=1)
+
+
+if __name__ == "__main__":
+    default_matter_test_main()

src/python_testing/TC_OpstateCommon.py

@@ -1267,25 +1267,32 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1):
             await self.read_and_expect_value(endpoint=endpoint,
                                              attribute=attributes.OperationalState,
                                              expected_value=cluster.Enums.OperationalStateEnum.kRunning)
-            count = sub_handler.attribute_report_counts[attributes.CountdownTime]
-            asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
+            countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
+            if countdownTime is not NullValue:
+                count = sub_handler.attribute_report_counts[attributes.CountdownTime]
+                asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
         else:
             self.skip_step(3)
 
         sub_handler.reset()
-        self.step(4)
-        logging.info('Test will now collect data for 30 seconds')
-        time.sleep(30)
+        countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
+        if countdownTime is not NullValue:
+            self.step(4)
+            logging.info('Test will now collect data for 30 seconds')
+            time.sleep(30)
 
-        count = sub_handler.attribute_report_counts[attributes.CountdownTime]
-        sub_handler.reset()
-        asserts.assert_less_equal(count, 5, "Received more than 5 reports for CountdownTime")
-        asserts.assert_greater_equal(count, 0, "Did not receive any reports for CountdownTime")
+            count = sub_handler.attribute_report_counts[attributes.CountdownTime]
+            sub_handler.reset()
+            asserts.assert_less_equal(count, 5, "Received more than 5 reports for CountdownTime")
+            asserts.assert_greater_equal(count, 0, "Did not receive any reports for CountdownTime")
+        else:
+            self.skip_step(4)
 
         attr_value = await self.read_expect_success(
             endpoint=endpoint,
             attribute=attributes.OperationalState)
-        if attr_value == cluster.Enums.OperationalStateEnum.kRunning:
+        countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
+        if attr_value == cluster.Enums.OperationalStateEnum.kRunning and countdownTime is not NullValue:
             self.step(5)
             wait_count = 0
             while (attr_value != cluster.Enums.OperationalStateEnum.kStopped) and (wait_count < 20):
@@ -1301,17 +1308,19 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1):
             self.skip_step(5)
 
         sub_handler.reset()
-        self.step(6)
         if self.pics_guard(self.check_pics(f"{self.test_info.pics_code}.S.M.ST_RUNNING")):
+            self.step(6)
             self.send_manual_or_pipe_command(name="OperationalStateChange",
                                              device=self.device,
                                              operation="Start")
             time.sleep(1)
             await self.read_and_expect_value(endpoint=endpoint,
                                              attribute=attributes.OperationalState,
                                              expected_value=cluster.Enums.OperationalStateEnum.kRunning)
-            count = sub_handler.attribute_report_counts[attributes.CountdownTime]
-            asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
+            countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
+            if countdownTime is not NullValue:
+                count = sub_handler.attribute_report_counts[attributes.CountdownTime]
+                asserts.assert_greater(count, 0, "Did not receive any reports for CountdownTime")
         else:
             self.skip_step(6)
 
@@ -1321,8 +1330,9 @@ async def TEST_TC_OPSTATE_BASE_2_6(self, endpoint=1):
                                          expected_value=cluster.Enums.OperationalStateEnum.kRunning)
 
         sub_handler.reset()
-        self.step(8)
-        if self.pics_guard(self.check_pics(f"{self.test_info.pics_code}.S.M.ST_PAUSED")):
+        countdownTime = await self.read_expect_success(endpoint=endpoint, attribute=attributes.CountdownTime)
+        if self.pics_guard(self.check_pics(f"{self.test_info.pics_code}.S.M.ST_PAUSED")) and countdownTime is not NullValue:
+            self.step(8)
             self.send_manual_or_pipe_command(name="OperationalStateChange",
                                              device=self.device,
                                              operation="Pause")