TC-PS-2.3:Add

Test plan:
CHIP-Specifications/chip-test-plans#4308

.github/workflows/tests.yaml

@@ -567,6 +567,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_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'
                   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_MWOM_1_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_PS_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_RVCRUNM_1_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_RVCRUNM_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_RVCRUNM_2_2.py'

src/python_testing/TC_PS_2_3.py

@@ -0,0 +1,72 @@
+#
+#    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: --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
+# === END CI TEST ARGUMENTS ===
+
+import logging
+import time
+
+import chip.clusters as Clusters
+from chip.clusters.Types import NullValue
+from matter_testing_support import ClusterAttributeChangeAccumulator, MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from mobly import asserts
+
+
+class TC_PS_2_3(MatterBaseTest):
+
+    def pics_TC_PS_2_3(self) -> list[str]:
+        return ["PWRTL.S"]
+
+    def steps_TC_PS_2_3(self):
+        return [TestStep(1, "Commission DUT to TH", "", is_commissioning=True),
+                TestStep(2, "Subscribe to all attributes of the PowerSource Cluster"),
+                TestStep(3, "Accumulate all attribute reports on the endpoint under test for 30 seconds",
+                            "For each of the attributes in the set of BatTimeToFullCharge, BatPercentRemaining and BatTimeRemaining, verify that there are not more than 4 reports per attribute where the value is non-null over the period of accumulation.")]
+
+    @async_test_body
+    async def test_TC_PS_2_3(self):
+        # Commissioning, already done.
+        self.step(1)
+
+        self.step(2)
+        ps = Clusters.PowerSource
+        sub_handler = ClusterAttributeChangeAccumulator(ps)
+        await sub_handler.start(self.default_controller, self.dut_node_id, self.matter_test_config.endpoint)
+
+        self.step(3)
+        logging.info("This test will now wait for 30 seconds.")
+        time.sleep(30)
+
+        counts = sub_handler.attribute_report_counts
+        asserts.assert_less_equal(counts[ps.Attributes.BatTimeToFullCharge], 4, "Too many reports for BatTimeToFullCharge")
+        asserts.assert_less_equal(counts[ps.Attributes.BatPercentRemaining], 4, "Too many reports for BatPercentRemaining")
+        asserts.assert_less_equal(counts[ps.Attributes.BatTimeRemaining], 4, "Too many reports for BatTimeRemaining")
+
+
+
+if __name__ == "__main__":
+    default_matter_test_main()

src/python_testing/matter_testing_support.py

@@ -34,7 +34,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import List, Optional, Tuple
+from typing import Any, List, Optional, Tuple
 
 from chip.tlv import float32, uint
 
@@ -299,6 +299,56 @@ def wait_for_report(self):
             asserts.fail("[AttributeChangeCallback] Attribute {expected_attribute} not found in returned report")
 
 
+
+@dataclass
+class AttributeValue:
+    endpoint_id: int
+    attribute: ClusterObjects.ClusterAttributeDescriptor
+    value: Any
+    timestamp_utc: datetime
+
+
+class ClusterAttributeChangeAccumulator:
+    def __init__(self, expected_cluster: ClusterObjects.Cluster):
+        self._q = queue.Queue()
+        self._expected_cluster = expected_cluster
+        self._subscription = None
+        self._attribute_report_counts = {}
+        attrs = [cls for name, cls in inspect.getmembers(expected_cluster.Attributes) if inspect.isclass(cls) and issubclass(cls, ClusterObjects.ClusterAttributeDescriptor)]
+        for a in attrs:
+            self._attribute_report_counts[a] = 0
+
+    async def start(self, dev_ctrl, node_id: int, endpoint: int, fabric_filtered: bool = False, min_interval_sec: int = 0, max_interval_sec: int = 5) -> Any:
+        """This starts a subscription for attributes on the specified node_id and endpoint. The cluster is specified when the class instance is created."""
+        self._subscription = await dev_ctrl.ReadAttribute(
+            nodeid=node_id,
+            attributes=[(endpoint, self._expected_cluster)],
+            reportInterval=(min_interval_sec, max_interval_sec),
+            fabricFiltered=fabric_filtered,
+            keepSubscriptions=True
+        )
+        self._subscription.SetAttributeUpdateCallback(self.__call__)
+        return self._subscription
+
+    def __call__(self, path: TypedAttributePath, transaction: SubscriptionTransaction):
+        """This is the subscription callback when an attribute report is received.
+           It checks the report is from the expected_cluster and then posts it into the queue for later processing."""
+        if path.ClusterType == self._expected_cluster:
+            data = transaction.GetAttribute(path)
+            value = AttributeValue(endpoint_id=path.Path.EndpointId, attribute=path.AttributeType, value=data, timestamp_utc=datetime.now(timezone.utc))
+            logging.info(f"Got subscription report for {path.AttributeType}: {data}")
+            self._q.put(value)
+            self._attribute_report_counts[path.AttributeType] += 1
+
+    @property
+    def attribute_queue(self) -> queue.Queue:
+        return self._q
+
+    @property
+    def attribute_report_counts(self) -> dict[ClusterObjects.ClusterAttributeDescriptor, int]:
+        return self._attribute_report_counts
+
+
 class InternalTestRunnerHooks(TestRunnerHooks):
 
     def start(self, count: int):