[MatterYamlTests] Allow enum names in YAML instead of raw values (#32107

)

* [YAML] Allow the YAML tests to use the enum names instead of the raw value

* Update the YAML tests

* [MatterYamlTests] Get test_yaml_parser.py to be runned in CI

* [MatterYamlTests] Add tests to test_yaml_parser.py

* Update errors.py

Co-authored-by: Boris Zbarsky <[email protected]>

---------

Co-authored-by: Boris Zbarsky <[email protected]>

scripts/py_matter_yamltests/BUILD.gn

@@ -55,6 +55,7 @@ pw_python_package("matter_yamltests") {
     "test_pics_checker.py",
     "test_parser_builder.py",
     "test_pseudo_clusters.py",
+    "test_yaml_parser.py",
     "test_yaml_loader.py",
   ]
 

scripts/py_matter_yamltests/matter_yamltests/errors.py

@@ -222,3 +222,45 @@ def __init__(self, content):
         self.tag_key_with_error(content, 'attribute')
         response = content.get('response')
         self.tag_key_with_error(response, 'saveAs')
+
+
+class TestStepEnumError(TestStepError):
+    """
+    Raise when an enum value or an enum name is not found in the definitions.
+
+    Parameters:
+        - enum_name_or_value (str|int): The name (str) or value (int) of the enumeration in the step.
+          If a string is provided, it is considered the name of the enumeration; if an integer is provided, it is considered the value of the enumeration.
+        - enum_candidates (dict): A dictionary mapping enumeration names (as strings) to their corresponding values
+          (as integers). This dictionary represents all known values of the enumeration.
+    """
+
+    def __init__(self, enum_name_or_value, enum_candidates: dict):
+        if type(enum_name_or_value) is str:
+            message = f'Unknown enum name: "{enum_name_or_value}". The possible values are: "{enum_candidates}"'
+
+            for enum_name in enum_candidates:
+                if enum_name.lower() == enum_name_or_value.lower():
+                    message = f'Unknown enum name: "{enum_name_or_value}". Did you mean "{enum_name}" ?'
+                    break
+
+        else:
+            message = f'Unknown enum value: "{enum_name_or_value}". The possible values are: "{enum_candidates}"'
+
+        super().__init__(message)
+
+
+class TestStepEnumSpecifierNotUnknownError(TestStepError):
+    """Raise when an enum value declared as unknown is in fact a known enum value from the definitions."""
+
+    def __init__(self, specified_value, enum_name):
+        message = f'The value "{specified_value}" is not unknown. It is the value of "{enum_name}"'
+        super().__init__(message)
+
+
+class TestStepEnumSpecifierWrongError(TestStepError):
+    """Raise when an enum value is specified for a given enum name but it does not match the enum value from the definitions."""
+
+    def __init__(self, specified_value, enum_name, enum_value):
+        message = f'The value "{specified_value}" is not the value of "{enum_name}({enum_value})"'
+        super().__init__(message)

scripts/py_matter_yamltests/matter_yamltests/parser.py

@@ -15,14 +15,16 @@
 
 import copy
 import logging
+import re
 from dataclasses import dataclass, field
 from enum import Enum, auto
 from typing import Optional
 
 from . import fixes
 from .constraints import get_constraints, is_typed_constraint
 from .definitions import SpecDefinitions
-from .errors import TestStepError, TestStepKeyError, TestStepValueNameError
+from .errors import (TestStepEnumError, TestStepEnumSpecifierNotUnknownError, TestStepEnumSpecifierWrongError, TestStepError,
+                     TestStepKeyError, TestStepValueNameError)
 from .pics_checker import PICSChecker
 from .yaml_loader import YamlLoader
 
@@ -38,6 +40,9 @@
     'SubscribeAll',
 ]
 
+# If True, enum values should use a valid name instead of a raw value
+STRICT_ENUM_VALUE_CHECK = False
+
 
 class UnknownPathQualifierError(TestStepError):
     """Raise when an attribute/command/event name is not found in the definitions."""
@@ -169,6 +174,99 @@ def _value_or_config(data, key, config):
     return data[key] if key in data else config.get(key)
 
 
+class EnumType:
+    def __init__(self, enum: Enum):
+        self.type = enum.name
+        self.base_type = enum.base_type
+
+        self._codes = {}
+        self.entries_by_name = {}
+        self.entries_by_code = {}
+        self._compute_entries(enum)
+
+    def translate(self, key: str, value) -> int:
+        if self._codes.get(key) is not None and self._codes.get(key) == value:
+            return self._codes.get(key)
+
+        if type(value) is str:
+            code = self._get_code_by_name(value)
+        else:
+            code = self._get_code_by_value(value)
+
+        if code is None:
+            raise TestStepEnumError(value, self.entries_by_name)
+
+        self._codes[key] = code
+        return code
+
+    def _get_code_by_name(self, value):
+        # For readability the name could sometimes be written as "enum_name(enum_code)" instead of "enum_name"
+        # In this case the enum_code should be checked to ensure that it is correct, unless enum_name is UnknownEnumValue
+        # in which case only invalid enum_code are allowed.
+        specified_name, specified_code = self._extract_name_and_code(value)
+        if specified_name not in self.entries_by_name:
+            return None
+
+        enum_code = self.entries_by_name.get(specified_name)
+        if specified_code is None or specified_code == enum_code:
+            return enum_code
+
+        if specified_name != f'{self.type}.UnknownEnumValue':
+            raise TestStepEnumSpecifierWrongError(
+                specified_code, specified_name, enum_code)
+
+        enum_name = self.entries_by_code.get(specified_code)
+        if enum_name:
+            raise TestStepEnumSpecifierNotUnknownError(value, enum_name)
+
+        return specified_code
+
+    def _get_code_by_value(self, value):
+        enum_name = self.entries_by_code.get(value)
+        if not enum_name:
+            return None
+
+        if STRICT_ENUM_VALUE_CHECK:
+            raise TestStepEnumError(value, self.entries_by_name)
+
+        return value
+
+    def _compute_entries(self, enum: Enum):
+        enum_codes = []
+        for enum_entry in enum.entries:
+            name = f'{self.type}.{enum_entry.name}'
+            code = enum_entry.code
+
+            self.entries_by_name[name] = code
+            self.entries_by_code[code] = name
+            enum_codes.append(code)
+
+        # search for the first invalid entry if any
+        max_code = 0xFF + 1
+        if self.base_type == 'enum16':
+            max_code = 0xFFFF + 1
+
+        for code in range(0, max_code):
+            if code not in enum_codes:
+                name = f'{self.type}.UnknownEnumValue'
+                self.entries_by_name[name] = code
+                self.entries_by_code[code] = name
+                break
+
+    def _extract_name_and_code(self, enum_name: str):
+        match = re.match(r"([\w.]+)(?:\((\w+)\))?", enum_name)
+        if match:
+            name = match.group(1)
+            code = int(match.group(2)) if match.group(2) else None
+            return name, code
+
+        return None, None
+
+    @staticmethod
+    def is_valid_type(target_type: str):
+        return target_type == 'enum8' or target_type == 'enum16'
+
+
 class _TestStepWithPlaceholders:
     '''A single YAML test parsed, as is, from YAML.
 
@@ -441,7 +539,11 @@ def _as_mapping(self, definitions, cluster_name, target_name):
         element = definitions.get_type_by_name(cluster_name, target_name)
 
         if hasattr(element, 'base_type'):
-            target_name = element.base_type.lower()
+            if EnumType.is_valid_type(element.base_type):
+                target_name = EnumType(element)
+            else:
+                target_name = element.base_type
+
         elif hasattr(element, 'fields'):
             target_name = {f.name: self._as_mapping(
                 definitions, cluster_name, f.data_type.name) for f in element.fields}
@@ -480,7 +582,11 @@ def _update_with_definition(self, container: dict, mapping_type):
 
                 if key == 'value':
                     value[key] = self._update_value_with_definition(
-                        item_value, mapping)
+                        value,
+                        key,
+                        item_value,
+                        mapping
+                    )
                 elif key == 'saveAs' and type(item_value) is str and item_value not in self._parsing_config_variable_storage:
                     self._parsing_config_variable_storage[item_value] = None
                 elif key == 'saveDataVersionAs' and type(item_value) is str and item_value not in self._parsing_config_variable_storage:
@@ -491,37 +597,80 @@ def _update_with_definition(self, container: dict, mapping_type):
                         # the the value type for the target field.
                         if is_typed_constraint(constraint):
                             value[key][constraint] = self._update_value_with_definition(
-                                constraint_value, mapping)
+                                item_value,
+                                constraint,
+                                constraint_value,
+                                mapping
+                            )
                 else:
                     # This key, value pair does not rely on cluster specifications.
                     pass
 
-    def _update_value_with_definition(self, value, mapping_type):
+    def _update_value_with_definition(self, container: dict, key: str, value, mapping_type):
+        """
+        Processes a given value based on a specified mapping type and returns the updated value.
+        This method does not modify the container in place; rather, it returns a new value that should be
+        used to update or process further as necessary.
+
+        The 'container' and 'key' parameters are primarily used for error tagging. If an error occurs
+        during the value processing, these parameters allow for the error to be precisely located and
+        reported, facilitating easier debugging and error tracking.
+
+        Parameters:
+          - container (dict): A dictionary that serves as a context for the operation. It is used for error
+            tagging if processing fails, by associating errors with specific locations within the data structure.
+          - key (str): The key related to the value being processed. It is used alongside 'container' to tag
+            errors, enabling precise identification of the error source.
+          - value: The value to be processed according to the mapping type.
+          - mapping_type: Dictates the processing or mapping logic to be applied to 'value'.
+
+        Returns:
+          The processed value, which is the result of applying the specified mapping type to the original 'value'.
+          This method does not update the 'container'; any necessary updates based on the processed value must
+          be handled outside this method.
+
+        Raises:
+          - TestStepError: If an error occurs during the processing of the value. The error includes details
+          from the 'container' and 'key' to facilitate error tracing and debugging.
+        """
+
         if not mapping_type:
             return value
 
         if type(value) is dict:
             rv = {}
-            for key in value:
+            for item_key in value:
                 # FabricIndex is a special case where the framework requires it to be passed even
                 # if it is not part of the requested arguments per spec and not part of the XML
                 # definition.
-                if key == 'FabricIndex' or key == 'fabricIndex':
-                    rv[key] = value[key]  # int64u
+                if item_key == 'FabricIndex' or item_key == 'fabricIndex':
+                    rv[item_key] = value[item_key]  # int64u
                 else:
-                    if not mapping_type.get(key):
-                        raise TestStepKeyError(value, key)
-                    mapping = mapping_type[key]
-                    rv[key] = self._update_value_with_definition(
-                        value[key], mapping)
+                    if not mapping_type.get(item_key):
+                        raise TestStepKeyError(value, item_key)
+                    mapping = mapping_type[item_key]
+                    rv[item_key] = self._update_value_with_definition(
+                        value,
+                        item_key,
+                        value[item_key],
+                        mapping
+                    )
             return rv
+
         if type(value) is list:
-            return [self._update_value_with_definition(entry, mapping_type) for entry in value]
+            return [self._update_value_with_definition(container, key, entry, mapping_type) for entry in value]
+
         # TODO currently unsure if the check of `value not in config` is sufficant. For
         # example let's say value = 'foo + 1' and map type is 'int64u', we would arguably do
         # the wrong thing below.
         if value is not None and value not in self._parsing_config_variable_storage:
-            if mapping_type == 'int64u' or mapping_type == 'int64s' or mapping_type == 'bitmap64' or mapping_type == 'epoch_us':
+            if type(mapping_type) is EnumType:
+                try:
+                    value = mapping_type.translate(key, value)
+                except (TestStepEnumError, TestStepEnumSpecifierNotUnknownError, TestStepEnumSpecifierWrongError) as e:
+                    e.tag_key_with_error(container, key)
+                    raise e
+            elif mapping_type == 'int64u' or mapping_type == 'int64s' or mapping_type == 'bitmap64' or mapping_type == 'epoch_us':
                 value = fixes.try_apply_float_to_integer_fix(value)
                 value = fixes.try_apply_yaml_cpp_longlong_limitation_fix(value)
                 value = fixes.try_apply_yaml_unrepresentable_integer_for_javascript_fixes(