Fix TC-OCC-3.1 and TC-OCC-3.2 runtime errors (#34747)

* Fix TC-OCC-3.1 and TC-OCC-3.2 runtime errors

- Refactor common code to be clearer in TC-OCC-3.1/3.2
- Fix runtime errors
- Move `await_sequence_of_reports` to matter_testing_support.py
- Make a helper to write attributes: `write_single_attribute`

Fixes #34730

Testing done:

- TC-OCC-3.1 and 3.2 no longer break locally (CI integration
  pending named pipe update in follow-up)
- TC_SWTCH still works

* Fix lint

* restyle

* poke styler

* restyle

Author: tcarmelveilleux

src/python_testing/TC_OCC_3_1.py

@@ -16,11 +16,11 @@
 #
 # === BEGIN CI TEST ARGUMENTS ===
 # test-runner-runs: run1
-# test-runner-run/run1/app: ${TYPE_OF_APP}
+# 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: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto --endpoint 1
 # === END CI TEST ARGUMENTS ===
 #  There are CI issues to be followed up for the test cases below that implements manually controlling sensor device for
 #  the occupancy state ON/OFF change.
@@ -29,19 +29,29 @@
 
 import logging
 import time
+from typing import Any, Optional
 
 import chip.clusters as Clusters
-from chip import ChipDeviceCtrl
 from chip.interaction_model import Status
 from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
 from mobly import asserts
 
 
 class TC_OCC_3_1(MatterBaseTest):
-    async def read_occ_attribute_expect_success(self, endpoint, attribute):
+    async def read_occ_attribute_expect_success(self, attribute):
         cluster = Clusters.Objects.OccupancySensing
+        endpoint = self.matter_test_config.endpoint
         return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
 
+    async def write_hold_time(self, hold_time: Optional[Any]) -> Status:
+        dev_ctrl = self.default_controller
+        node_id = self.dut_node_id
+        endpoint = self.matter_test_config.endpoint
+
+        cluster = Clusters.OccupancySensing
+        write_result = await dev_ctrl.WriteAttribute(node_id, [(endpoint, cluster.Attributes.HoldTime(hold_time))])
+        return write_result[0].Status
+
     def desc_TC_OCC_3_1(self) -> str:
         return "[TC-OCC-3.1] Primary functionality with server as DUT"
 
@@ -63,58 +73,56 @@ def pics_TC_OCC_3_1(self) -> list[str]:
 
     @async_test_body
     async def test_TC_OCC_3_1(self):
-
-        endpoint = self.user_params.get("endpoint", 1)
-        node_id = self.matter_test_config.dut_node_ids[0]
         hold_time = 10  # 10 seconds for occupancy state hold time
 
         self.step(1)  # commissioning and getting cluster attribute list
-        attributes = Clusters.OccupancySensing.Attributes
-        attribute_list = await self.read_occ_attribute_expect_success(endpoint=endpoint, attribute=attributes.AttributeList)
+        cluster = Clusters.OccupancySensing
+        attributes = cluster.Attributes
+        attribute_list = await self.read_occ_attribute_expect_success(attribute=attributes.AttributeList)
 
-        self.step(2)
-        if attributes.HoldTime.attribute_id in attribute_list:
-            # write 10 as a HoldTime attibute
-            write_res = await ChipDeviceCtrl.WriteAttribute(node_id, [(endpoint, attributes.HoldTime(hold_time))])
-            asserts.assert_equal(write_res[0].status, Status.Success, "Write HoldTime failed")
+        has_hold_time = attributes.HoldTime.attribute_id in attribute_list
 
+        self.step(2)
+        if has_hold_time:
+            # write 10 as a HoldTime attribute
+            await self.write_single_attribute(cluster.Attributes.HoldTime(hold_time))
         else:
             logging.info("No HoldTime attribute supports. Will test only occupancy attribute triggering functionality")
 
         self.step(3)
         # check if Occupancy attribute is 0
-        occupancy_dut = await self.read_occ_attribute_expect_success(endpoint=endpoint, attribute=attributes.Occupancy)
+        occupancy_dut = await self.read_occ_attribute_expect_success(attribute=attributes.Occupancy)
 
         # if occupancy is on, then wait until the sensor occupancy state is 0.
         if occupancy_dut == 1:
             # Don't trigger occupancy sensor to render occupancy attribute to 0
-            if attributes.HoldTime.attribute_id in attribute_list:
-                time.sleep(hold_time + 2)  # add some extra 2 seconds to ensure hold time has passed.
+            if has_hold_time:
+                time.sleep(hold_time + 2.0)  # add some extra 2 seconds to ensure hold time has passed.
             else:  # a user wait until a sensor specific time to change occupancy attribute to 0.  This is the case where the sensor doesn't support HoldTime.
                 self.wait_for_user_input(
                     prompt_msg="Type any letter and press ENTER after the sensor occupancy is detection ready state (occupancy attribute = 0)")
 
         # check sensor occupancy state is 0 for the next test step
-        occupancy_dut = await self.read_occ_attribute_expect_success(endpoint=endpoint, attribute=attributes.Occupancy)
+        occupancy_dut = await self.read_occ_attribute_expect_success(attribute=attributes.Occupancy)
         asserts.assert_equal(occupancy_dut, 0, "Occupancy attribute is still 1.")
 
         self.step(4)
         # Trigger occupancy sensor to change Occupancy attribute value to 1 => TESTER ACTION on DUT
         self.wait_for_user_input(prompt_msg="Type any letter and press ENTER after a sensor occupancy is triggered.")
 
         # And then check if Occupancy attribute has changed.
-        occupancy_dut = await self.read_occ_attribute_expect_success(endpoint=endpoint, attribute=attributes.Occupancy)
+        occupancy_dut = await self.read_occ_attribute_expect_success(attribute=attributes.Occupancy)
         asserts.assert_equal(occupancy_dut, 1, "Occupancy state is not changed to 1")
 
         self.step(5)
         # check if Occupancy attribute is back to 0 after HoldTime attribute period
         # Tester should not be triggering the sensor for this test step.
-        if attributes.HoldTime.attribute_id in attribute_list:
+        if has_hold_time:
 
             # Start a timer based on HoldTime
-            time.sleep(hold_time+2)  # add some extra 2 seconds to ensure hold time has passed.
+            time.sleep(hold_time + 2.0)  # add some extra 2 seconds to ensure hold time has passed.
 
-            occupancy_dut = await self.read_occ_attribute_expect_success(endpoint=endpoint, attribute=attributes.Occupancy)
+            occupancy_dut = await self.read_occ_attribute_expect_success(attribute=attributes.Occupancy)
             asserts.assert_equal(occupancy_dut, 0, "Occupancy state is not 0 after HoldTime period")
 
         else: