From 2304948522892298f4b54acf8332eb9fc328fca1 Mon Sep 17 00:00:00 2001 From: Cecille Freeman Date: Sat, 27 Jul 2024 14:49:30 -0400 Subject: [PATCH] Add has_feature partial for decorator --- src/python_testing/matter_testing_support.py | 36 +++++++++++++++++-- .../test_testing/TestDecorators.py | 17 +++++++-- 2 files changed, 48 insertions(+), 5 deletions(-) diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index 3d235e49873a8a..812713f243c953 100644 --- a/src/python_testing/matter_testing_support.py +++ b/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. diff --git a/src/python_testing/test_testing/TestDecorators.py b/src/python_testing/test_testing/TestDecorators.py index 60a75bfca466ef..2ce418d6c5e43a 100644 --- a/src/python_testing/test_testing/TestDecorators.py +++ b/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')