Add has_feature partial for decorator

src/python_testing/matter_testing_support.py

@@ -33,7 +33,7 @@
 from dataclasses import asdict as dataclass_asdict
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
-from enum import Enum
+from enum import Enum, IntFlag
 from functools import partial
 from typing import Any, List, Optional, Tuple
 
@@ -1093,7 +1093,7 @@ def mark_current_step_skipped(self):
             steps = self.get_test_steps(self.current_test_info.name)
             if self.current_step_index == 0:
                 asserts.fail("Script error: mark_current_step_skipped cannot be called before step()")
-            num = steps[self.current_step_index-1].test_plan_number
+            num = steps[self.current_step_index - 1].test_plan_number
         except KeyError:
             num = self.current_step_index
 
@@ -1730,6 +1730,38 @@ def has_attribute(attribute: ClusterObjects.ClusterAttributeDescriptor) -> Endpo
     return partial(_has_attribute, attribute=attribute)
 
 
+def _has_feature(wildcard, endpoint, cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFlag) -> bool:
+    try:
+        feature_map = wildcard.attributes[endpoint][cluster][cluster.Attributes.FeatureMap]
+        return (feature & feature_map) != 0
+    except KeyError:
+        return False
+
+
+def has_feature(cluster: ClusterObjects.ClusterObjectDescriptor, feature: IntFlag) -> EndpointCheckFunction:
+    """ EndpointCheckFunction that can be passed as a parameter to the per_endpoint_test decorator.
+
+        Use this function with the per_endpoint_test 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
+        EP1: cluster D with feature F0
+        EP2, cluster D with feature F0
+        EP3, cluster D without feature F0
+
+        And the following test specification:
+        @per_endpoint_test(has_feature(Clusters.D.Bitmaps.Feature.F0))
+        test_mytest(self):
+            ...
+
+        The test would be run on endpoint 1 and on endpoint 2.
+
+        If the cluster is not found on any endpoint the decorator will call the on_skip function to
+        notify the test harness that the test is not applicable to this node and the test will not be run.
+    """
+    return partial(_has_feature, cluster=cluster, feature=feature)
+
+
 async def get_accepted_endpoints_for_test(self: MatterBaseTest, accept_function: EndpointCheckFunction) -> list[uint]:
     """ Helper function for the per_endpoint_test decorator.
 

src/python_testing/test_testing/TestDecorators.py

@@ -30,15 +30,14 @@
 
 import chip.clusters as Clusters
 from chip.clusters import Attribute
-from chip.clusters import ClusterObjects as ClusterObjects
 
 try:
     from matter_testing_support import (MatterBaseTest, async_test_body, get_accepted_endpoints_for_test, has_attribute,
-                                        has_cluster, per_endpoint_test, per_node_test)
+                                        has_cluster, has_feature, per_endpoint_test, per_node_test)
 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, per_endpoint_test, per_node_test)
+                                        has_cluster, has_feature, per_endpoint_test, per_node_test)
 
 from typing import Optional
 
@@ -204,6 +203,16 @@ async def test_endpoint_attribute_supported_cluster_no(self):
     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))
+    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))
+    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))
     async def test_endpoint_boolean_yes(self):
@@ -292,6 +301,8 @@ def check_skipped(test_name: str):
     check_once_per_endpoint('test_endpoint_attribute_yes')
     check_skipped('test_endpoint_attribute_supported_cluster_no')
     check_skipped('test_endpoint_attribute_unsupported_cluster_no')
+    check_once_per_endpoint('test_endpoint_feature_yes')
+    check_skipped('test_endpoint_feature_unsupported_cluster_no')
     check_once_per_endpoint('test_endpoint_boolean_yes')
     check_skipped('test_endpoint_boolean_no')