Change name of decorator

Author: cecille

src/python_testing/TC_CCTRL_2_1.py

@@ -16,7 +16,7 @@
 #
 
 import chip.clusters as Clusters
-from matter_testing_support import MatterBaseTest, TestStep, default_matter_test_main, has_cluster, per_endpoint_test
+from matter_testing_support import MatterBaseTest, TestStep, default_matter_test_main, has_cluster, run_for_each_matching_endpoint
 from mobly import asserts
 
 
@@ -27,7 +27,7 @@ def steps_TC_CCTRL_2_1(self) -> list[TestStep]:
                  TestStep(2, "Validate SupportedDeviceCategories is set accordingly based on MCORE.FS")]
         return steps
 
-    @per_endpoint_test(has_cluster(Clusters.CommissionerControl))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.CommissionerControl))
     async def test_TC_CCTRL_2_1(self):
         self.step(1)
         is_fabric_sync_pics_enabled = self.check_pics("MCORE.FS")

src/python_testing/TC_CCTRL_2_2.py

@@ -35,7 +35,7 @@
 from chip import ChipDeviceCtrl
 from chip.interaction_model import InteractionModelError, Status
 from matter_testing_support import (MatterBaseTest, TestStep, async_test_body, default_matter_test_main, has_cluster,
-                                    per_endpoint_test)
+                                    run_for_each_matching_endpoint)
 from mobly import asserts
 
 
@@ -112,7 +112,7 @@ def steps_TC_CCTRL_2_2(self) -> list[TestStep]:
 
         return steps
 
-    @per_endpoint_test(has_cluster(Clusters.CommissionerControl))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.CommissionerControl))
     async def test_TC_CCTRL_2_2(self):
         self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
 

src/python_testing/TC_CC_2_2.py

@@ -33,7 +33,7 @@
 import chip.clusters as Clusters
 from chip.clusters import ClusterObjects as ClusterObjects
 from matter_testing_support import (ClusterAttributeChangeAccumulator, MatterBaseTest, TestStep, default_matter_test_main,
-                                    has_cluster, per_endpoint_test)
+                                    has_cluster, run_for_each_matching_endpoint)
 from mobly import asserts
 from test_plan_support import commission_if_required, if_feature_supported, read_attribute, verify_success
 
@@ -107,7 +107,7 @@ def entry_count_verification() -> str:
                          "The third entry in _reportedRemainingTimeValuesList_ is equal to 0")
                 ]
 
-    @per_endpoint_test(has_cluster(Clusters.ColorControl))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.ColorControl))
     async def test_TC_CC_2_2(self):
         gather_time = 20
 

src/python_testing/TC_LVL_2_3.py

@@ -33,7 +33,7 @@
 import chip.clusters as Clusters
 import test_plan_support
 from matter_testing_support import (ClusterAttributeChangeAccumulator, MatterBaseTest, TestStep, default_matter_test_main,
-                                    has_cluster, per_endpoint_test)
+                                    has_cluster, run_for_each_matching_endpoint)
 from mobly import asserts
 
 
@@ -81,7 +81,7 @@ def steps_TC_LVL_2_3(self) -> list[TestStep]:
                          "The third entry in reportedRemainingTimeValuesList is equal to 0")
                 ]
 
-    @per_endpoint_test(has_cluster(Clusters.LevelControl))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.LevelControl))
     async def test_TC_LVL_2_3(self):
         # Commissioning - already done
         self.step(1)

src/python_testing/TC_SWTCH.py

@@ -39,7 +39,7 @@
 from chip.clusters.Attribute import EventReadResult
 from chip.tlv import uint
 from matter_testing_support import (ClusterAttributeChangeAccumulator, EventChangeCallback, MatterBaseTest, TestStep,
-                                    await_sequence_of_reports, default_matter_test_main, has_feature, per_endpoint_test)
+                                    await_sequence_of_reports, default_matter_test_main, has_feature, run_for_each_matching_endpoint)
 from mobly import asserts
 
 logger = logging.getLogger(__name__)
@@ -260,7 +260,7 @@ def steps_TC_SWTCH_2_2(self):
                 TestStep(10, "TH reads the CurrentPosition attribute from the DUT", "Verify that the value is 0"),
                 ]
 
-    @per_endpoint_test(has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kLatchingSwitch))
+    @run_for_each_matching_endpoint(has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kLatchingSwitch))
     async def test_TC_SWTCH_2_2(self):
         post_prompt_settle_delay_seconds = 10.0
 
@@ -346,7 +346,7 @@ def steps_TC_SWTCH_2_3(self):
                 TestStep(9, "TH reads the CurrentPosition attribute from the DUT", "Verify that the value is 0"),
                 ]
 
-    @per_endpoint_test(has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kMomentarySwitch))
+    @run_for_each_matching_endpoint(has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kMomentarySwitch))
     async def test_TC_SWTCH_2_3(self):
         # Commissioning - already done
         self.step(1)
@@ -425,7 +425,7 @@ def steps_TC_SWTCH_2_4(self):
                 """)
                 ]
 
-    @per_endpoint_test(has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kMomentarySwitch))
+    @run_for_each_matching_endpoint(has_feature(Clusters.Switch, Clusters.Switch.Bitmaps.Feature.kMomentarySwitch))
     async def test_TC_SWTCH_2_4(self):
         switch_pressed_position = self._default_pressed_position
         post_prompt_settle_delay_seconds = 10.0
@@ -598,7 +598,7 @@ def should_run_SWTCH_2_5(wildcard, endpoint):
         asf = has_feature(Clusters.Switch, 0x20)
         return msm(wildcard, endpoint) and not asf(wildcard, endpoint)
 
-    @per_endpoint_test(should_run_SWTCH_2_5)
+    @run_for_each_matching_endpoint(should_run_SWTCH_2_5)
     async def test_TC_SWTCH_2_5(self):
         # Commissioning - already done
         self.step(1)
@@ -773,7 +773,7 @@ def should_run_SWTCH_2_6(wildcard, endpoint):
         asf = has_feature(Clusters.Switch, 0x20)
         return msm(wildcard, endpoint) and asf(wildcard, endpoint)
 
-    @per_endpoint_test(should_run_SWTCH_2_6)
+    @run_for_each_matching_endpoint(should_run_SWTCH_2_6)
     async def test_TC_SWTCH_2_6(self):
         # Commissioning - already done
         self.step(1)

src/python_testing/TC_TIMESYNC_2_1.py

@@ -32,7 +32,7 @@
 
 import chip.clusters as Clusters
 from chip.clusters.Types import NullValue
-from matter_testing_support import (MatterBaseTest, default_matter_test_main, has_attribute, has_cluster, per_endpoint_test,
+from matter_testing_support import (MatterBaseTest, default_matter_test_main, has_attribute, has_cluster, run_for_each_matching_endpoint,
                                     utc_time_in_matter_epoch)
 from mobly import asserts
 
@@ -42,7 +42,7 @@ async def read_ts_attribute_expect_success(self, attribute):
         cluster = Clusters.Objects.TimeSynchronization
         return await self.read_single_attribute_check_success(endpoint=None, cluster=cluster, attribute=attribute)
 
-    @per_endpoint_test(has_cluster(Clusters.TimeSynchronization) and has_attribute(Clusters.TimeSynchronization.Attributes.TimeSource))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.TimeSynchronization) and has_attribute(Clusters.TimeSynchronization.Attributes.TimeSource))
     async def test_TC_TIMESYNC_2_1(self):
         attributes = Clusters.TimeSynchronization.Attributes
         features = await self.read_ts_attribute_expect_success(attribute=attributes.FeatureMap)

src/python_testing/matter_testing_support.py

@@ -1745,15 +1745,15 @@ def async_runner(self: MatterBaseTest, *args, **kwargs):
     return async_runner
 
 
-def per_node_test(body):
+def run_once_for_node(body):
     """ Decorator to be used for PICS-free tests that apply to the entire node.
 
     Use this decorator when your script needs to be run once to validate the whole node.
     To use this decorator, the test must NOT have an associated pics_ method.
     """
 
     def whole_node_runner(self: MatterBaseTest, *args, **kwargs):
-        asserts.assert_false(self.get_test_pics(self.current_test_info.name), "pics_ method supplied for per_node_test.")
+        asserts.assert_false(self.get_test_pics(self.current_test_info.name), "pics_ method supplied for run_once_for_node.")
         return _async_runner(body, self, *args, **kwargs)
 
     return whole_node_runner
@@ -1770,9 +1770,9 @@ def _has_cluster(wildcard, endpoint, cluster: ClusterObjects.Cluster) -> bool:
 
 
 def has_cluster(cluster: ClusterObjects.ClusterObjectDescriptor) -> EndpointCheckFunction:
-    """ EndpointCheckFunction that can be passed as a parameter to the per_endpoint_test decorator.
+    """ EndpointCheckFunction that can be passed as a parameter to the run_for_each_matching_endpoint decorator.
 
-        Use this function with the per_endpoint_test decorator to run this test on all endpoints with
+        Use this function with the run_for_each_matching_endpoint decorator to run this test on all endpoints with
         the specified cluster. For example, given a device with the following conformance
 
         EP0: cluster A, B, C
@@ -1781,7 +1781,7 @@ def has_cluster(cluster: ClusterObjects.ClusterObjectDescriptor) -> EndpointChec
         EP3, cluster E
 
         And the following test specification:
-        @per_endpoint_test(has_cluster(Clusters.D))
+        @run_for_each_matching_endpoint(has_cluster(Clusters.D))
         test_mytest(self):
             ...
 
@@ -1803,9 +1803,9 @@ def _has_attribute(wildcard, endpoint, attribute: ClusterObjects.ClusterAttribut
 
 
 def has_attribute(attribute: ClusterObjects.ClusterAttributeDescriptor) -> EndpointCheckFunction:
-    """ EndpointCheckFunction that can be passed as a parameter to the per_endpoint_test decorator.
+    """ EndpointCheckFunction that can be passed as a parameter to the run_for_each_matching_endpoint decorator.
 
-        Use this function with the per_endpoint_test decorator to run this test on all endpoints with
+        Use this function with the run_for_each_matching_endpoint decorator to run this test on all endpoints with
         the specified attribute. For example, given a device with the following conformance
 
         EP0: cluster A, B, C
@@ -1814,7 +1814,7 @@ def has_attribute(attribute: ClusterObjects.ClusterAttributeDescriptor) -> Endpo
         EP3, cluster D without attribute d
 
         And the following test specification:
-        @per_endpoint_test(has_attribute(Clusters.D.Attributes.d))
+        @run_for_each_matching_endpoint(has_attribute(Clusters.D.Attributes.d))
         test_mytest(self):
             ...
 
@@ -1835,9 +1835,9 @@ def _has_feature(wildcard, endpoint, cluster: ClusterObjects.ClusterObjectDescri
 
 
 def has_feature(cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFlag) -> EndpointCheckFunction:
-    """ EndpointCheckFunction that can be passed as a parameter to the per_endpoint_test decorator.
+    """ EndpointCheckFunction that can be passed as a parameter to the run_for_each_matching_endpoint decorator.
 
-        Use this function with the per_endpoint_test decorator to run this test on all endpoints with
+        Use this function with the run_for_each_matching_endpoint decorator to run this test on all endpoints with
         the specified feature. For example, given a device with the following conformance
 
         EP0: cluster A, B, C
@@ -1846,7 +1846,7 @@ def has_feature(cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFla
         EP3, cluster D without feature F0
 
         And the following test specification:
-        @per_endpoint_test(has_feature(Clusters.D.Bitmaps.Feature.F0))
+        @run_for_each_matching_endpoint(has_feature(Clusters.D.Bitmaps.Feature.F0))
         test_mytest(self):
             ...
 
@@ -1859,15 +1859,15 @@ def has_feature(cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFla
 
 
 async def get_accepted_endpoints_for_test(self: MatterBaseTest, accept_function: EndpointCheckFunction) -> list[uint]:
-    """ Helper function for the per_endpoint_test decorator.
+    """ Helper function for the run_for_each_matching_endpoint decorator.
 
         Returns a list of endpoints on which the test should be run given the accept_function for the test.
     """
     wildcard = await self.default_controller.Read(self.dut_node_id, [()])
     return [e for e in wildcard.attributes.keys() if accept_function(wildcard, e)]
 
 
-def per_endpoint_test(accept_function: EndpointCheckFunction):
+def run_for_each_matching_endpoint(accept_function: EndpointCheckFunction):
     """ Test decorator for a test that needs to be run once per endpoint that meets the accept_function criteria.
 
         Place this decorator above the test_ method to have the test framework run this test once per endpoint.
@@ -1882,7 +1882,7 @@ def per_endpoint_test(accept_function: EndpointCheckFunction):
         EP3, cluster E
 
         And the following test specification:
-        @per_endpoint_test(has_cluster(Clusters.D))
+        @run_for_each_matching_endpoint(has_cluster(Clusters.D))
         test_mytest(self):
             ...
 
@@ -1898,9 +1898,10 @@ def per_endpoint_test(accept_function: EndpointCheckFunction):
         Tests that use this decorator cannot use a pics_ method for test selection and should not reference any
         PICS values internally.
     """
-    def per_endpoint_test_internal(body):
+    def run_for_each_matching_endpoint_internal(body):
         def per_endpoint_runner(self: MatterBaseTest, *args, **kwargs):
-            asserts.assert_false(self.get_test_pics(self.current_test_info.name), "pics_ method supplied for per_endpoint_test.")
+            asserts.assert_false(self.get_test_pics(self.current_test_info.name),
+                                 "pics_ method supplied for run_for_each_matching_endpoint.")
             runner_with_timeout = asyncio.wait_for(get_accepted_endpoints_for_test(self, accept_function), timeout=30)
             endpoints = asyncio.run(runner_with_timeout)
             if not endpoints:
@@ -1927,7 +1928,7 @@ def per_endpoint_runner(self: MatterBaseTest, *args, **kwargs):
                     self.runner_hook.test_stop(exception=None, duration=test_duration)
             self.matter_test_config.endpoint = original_ep
         return per_endpoint_runner
-    return per_endpoint_test_internal
+    return run_for_each_matching_endpoint_internal
 
 
 class CommissionDeviceTest(MatterBaseTest):

src/python_testing/test_testing/TestDecorators.py

@@ -33,11 +33,11 @@
 
 try:
     from matter_testing_support import (MatterBaseTest, async_test_body, get_accepted_endpoints_for_test, has_attribute,
-                                        has_cluster, has_feature, per_endpoint_test, per_node_test)
+                                        has_cluster, has_feature, run_for_each_matching_endpoint, run_once_for_node)
 except ImportError:
     sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
     from matter_testing_support import (MatterBaseTest, async_test_body, get_accepted_endpoints_for_test, has_attribute,
-                                        has_cluster, has_feature, per_endpoint_test, per_node_test)
+                                        has_cluster, has_feature, run_for_each_matching_endpoint, run_once_for_node)
 
 from typing import Optional
 
@@ -154,7 +154,7 @@ async def test_endpoints(self):
         asserts.assert_equal(endpoints, [], msg)
 
     # This test should cause an assertion because it has pics_ method
-    @per_node_test
+    @run_once_for_node
     async def test_whole_node_with_pics(self):
         pass
 
@@ -164,7 +164,7 @@ def pics_whole_node_with_pics(self):
         return ['EXAMPLE.S']
 
     # This test should cause an assertion because it has a pics_ method
-    @per_endpoint_test(has_cluster(Clusters.OnOff))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.OnOff))
     async def test_per_endpoint_with_pics(self):
         pass
 
@@ -174,66 +174,66 @@ def pics_per_endpoint_with_pics(self):
         return ['EXAMPLE.S']
 
     # This test should be run once
-    @per_node_test
+    @run_once_for_node
     async def test_whole_node_ok(self):
         pass
 
     # This test should be run once per endpoint
-    @per_endpoint_test(has_cluster(Clusters.OnOff))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.OnOff))
     async def test_endpoint_cluster_yes(self):
         pass
 
     # This test should be skipped since this cluster isn't on any endpoint
-    @per_endpoint_test(has_cluster(Clusters.TimeSynchronization))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.TimeSynchronization))
     async def test_endpoint_cluster_no(self):
         pass
 
     # This test should be run once per endpoint
-    @per_endpoint_test(has_attribute(Clusters.OnOff.Attributes.OnOff))
+    @run_for_each_matching_endpoint(has_attribute(Clusters.OnOff.Attributes.OnOff))
     async def test_endpoint_attribute_yes(self):
         pass
 
     # This test should be skipped since this attribute isn't on the supported cluster
-    @per_endpoint_test(has_attribute(Clusters.OnOff.Attributes.OffWaitTime))
+    @run_for_each_matching_endpoint(has_attribute(Clusters.OnOff.Attributes.OffWaitTime))
     async def test_endpoint_attribute_supported_cluster_no(self):
         pass
 
     # This test should be skipped since this attribute is part of an unsupported cluster
-    @per_endpoint_test(has_attribute(Clusters.TimeSynchronization.Attributes.Granularity))
+    @run_for_each_matching_endpoint(has_attribute(Clusters.TimeSynchronization.Attributes.Granularity))
     async def test_endpoint_attribute_unsupported_cluster_no(self):
         pass
 
     # This test should be run once per endpoint
-    @per_endpoint_test(has_feature(Clusters.OnOff, Clusters.OnOff.Bitmaps.Feature.kLighting))
+    @run_for_each_matching_endpoint(has_feature(Clusters.OnOff, Clusters.OnOff.Bitmaps.Feature.kLighting))
     async def test_endpoint_feature_yes(self):
         pass
 
     # This test should be skipped since this attribute is part of an unsupported cluster
-    @per_endpoint_test(has_feature(Clusters.TimeSynchronization, Clusters.TimeSynchronization.Bitmaps.Feature.kNTPClient))
+    @run_for_each_matching_endpoint(has_feature(Clusters.TimeSynchronization, Clusters.TimeSynchronization.Bitmaps.Feature.kNTPClient))
     async def test_endpoint_feature_unsupported_cluster_no(self):
         pass
 
     # This test should be run since both are present
-    @per_endpoint_test(has_attribute(Clusters.OnOff.Attributes.OnOff) and has_cluster(Clusters.OnOff))
+    @run_for_each_matching_endpoint(has_attribute(Clusters.OnOff.Attributes.OnOff) and has_cluster(Clusters.OnOff))
     async def test_endpoint_boolean_yes(self):
         pass
 
     # This test should be skipped since we have an OnOff cluster, but no Time sync
-    @per_endpoint_test(has_cluster(Clusters.OnOff) and has_cluster(Clusters.TimeSynchronization))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.OnOff) and has_cluster(Clusters.TimeSynchronization))
     async def test_endpoint_boolean_no(self):
         pass
 
-    @per_endpoint_test(has_cluster(Clusters.OnOff))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.OnOff))
     async def test_fail_on_ep0(self):
         if self.matter_test_config.endpoint == 0:
             asserts.fail("Expected failure")
 
-    @per_endpoint_test(has_cluster(Clusters.OnOff))
+    @run_for_each_matching_endpoint(has_cluster(Clusters.OnOff))
     async def test_fail_on_ep1(self):
         if self.matter_test_config.endpoint == 1:
             asserts.fail("Expected failure")
 
-    @per_node_test
+    @run_once_for_node
     async def test_fail_on_whole_node(self):
         asserts.fail("Expected failure")