Yaml Test Runner: Add support for Python constraints

docs/testing/yaml_schema.md

@@ -62,6 +62,7 @@ YAML schema
 |      hasMasksClear |list||
 |      notValue |NoneType,bool,int,float,list,dict|Y|
 |      anyOf |list||
+|      python |str|Y|
 |    saveAs |str||
 |    saveDataVersschemaionAs |str||
 |  saveResponseAs |str||

scripts/py_matter_yamltests/matter_yamltests/constraints.py

@@ -15,13 +15,22 @@
 #    limitations under the License.
 #
 
+import ast
+import builtins
+import inspect
 import math
 import re
 import string
 from abc import ABC, abstractmethod
 from typing import List
 
 from .errors import TestStepError
+from .fixes import fix_typed_yaml_value
+
+
+def print_to_log(*objects, sep=' ', end=None):
+    # Try to fit in with the test logger output format
+    print('\n\t\t    ' + sep.join(str(arg) for arg in objects))
 
 
 class ConstraintParseError(Exception):
@@ -121,6 +130,11 @@ def __init__(self, context, reason):
         super().__init__(context, 'anyOf', reason)
 
 
+class ConstraintPythonError(ConstraintCheckError):
+    def __init__(self, context, reason):
+        super().__init__(context, 'python', reason)
+
+
 class BaseConstraint(ABC):
     '''Constraint Interface'''
 
@@ -130,7 +144,7 @@ def __init__(self, context, types: list, is_null_allowed: bool = False):
         self._is_null_allowed = is_null_allowed
         self._context = context
 
-    def validate(self, value, value_type_name):
+    def validate(self, value, value_type_name, runtime_variables):
         if value is None and self._is_null_allowed:
             return
 
@@ -194,6 +208,8 @@ def _raise_error(self, reason):
             raise ConstraintNotValueError(self._context, reason)
         elif isinstance(self, _ConstraintAnyOf):
             raise ConstraintAnyOfError(self._context, reason)
+        elif isinstance(self, _ConstraintPython):
+            raise ConstraintPythonError(self._context, reason)
         else:
             # This should not happens.
             raise ConstraintParseError('Unknown constraint instance.')
@@ -204,7 +220,7 @@ def __init__(self, context, has_value):
         super().__init__(context, types=[])
         self._has_value = has_value
 
-    def validate(self, value, value_type_name):
+    def validate(self, value, value_type_name, runtime_variables):
         # We are overriding the BaseConstraint of validate since has value is a special case where
         # we might not be expecting a value at all, but the basic null check in BaseConstraint
         # is not what we want.
@@ -824,6 +840,46 @@ def get_reason(self, value, value_type_name) -> str:
         return f'The response value "{value}" is not a value from {self._any_of}.'
 
 
+class _ConstraintPython(BaseConstraint):
+    def __init__(self, context, source: str):
+        super().__init__(context, types=[], is_null_allowed=False)
+
+        # Parse the source as the body of a function
+        if '\n' not in source:  # treat single line code like a lambda
+            source = 'return (' + source + ')\n'
+        parsed = ast.parse(source)
+        module = ast.parse('def _func(value): pass')
+        module.body[0].body = parsed.body  # inject parsed body
+        self._ast = module
+
+    def validate(self, value, value_type_name, runtime_variables):
+        # Build a global scope that includes all runtime variables
+        scope = {name: fix_typed_yaml_value(value) for name, value in runtime_variables.items()}
+        scope['__builtins__'] = self.BUILTINS
+        # Execute the module AST and extract the defined function
+        exec(compile(self._ast, '<string>', 'exec'), scope)
+        func = scope['_func']
+        # Call the function to validate the value
+        try:
+            valid = func(value)
+        except Exception as ex:
+            self._raise_error(f'Python constraint {type(ex).__name__}: {ex}')
+        if type(valid) is not bool:
+            self._raise_error("Python constraint TypeError: must return a bool")
+        if not valid:
+            self._raise_error(f'The response value "{value}" is not valid')
+
+    def check_response(self, value, value_type_name) -> bool: pass  # unused
+    def get_reason(self, value, value_type_name) -> str: pass  # unused
+
+    # Explicitly list allowed functions / constants, avoid things like exec, eval, import. Classes are generally safe.
+    ALLOWED_BUILTINS = ['True', 'False', 'None', 'abs', 'all', 'any', 'ascii', 'bin', 'chr', 'divmod', 'enumerate', 'filter', 'format',
+                        'hex', 'isinstance', 'issubclass', 'iter', 'len', 'max', 'min', 'next', 'oct', 'ord', 'pow', 'repr', 'round', 'sorted', 'sum']
+    BUILTINS = (dict(inspect.getmembers(builtins, inspect.isclass)) |
+                {name: getattr(builtins, name) for name in ALLOWED_BUILTINS} |
+                {'print': print_to_log})
+
+
 def get_constraints(constraints: dict) -> List[BaseConstraint]:
     _constraints = []
     context = constraints
@@ -879,6 +935,9 @@ def get_constraints(constraints: dict) -> List[BaseConstraint]:
         elif 'anyOf' == constraint:
             _constraints.append(_ConstraintAnyOf(
                 context, constraint_value))
+        elif 'python' == constraint:
+            _constraints.append(_ConstraintPython(
+                context, constraint_value))
         else:
             raise ConstraintParseError(f'Unknown constraint type:{constraint}')
 
@@ -904,9 +963,14 @@ def is_typed_constraint(constraint: str):
         'hasMasksClear': False,
         'notValue': True,
         'anyOf': True,
+        'python': False,
     }
 
     is_typed = constraints.get(constraint)
     if is_typed is None:
         raise ConstraintParseError(f'Unknown constraint type:{constraint}')
     return is_typed
+
+
+def is_variable_aware_constraint(constraint: str):
+    return constraint == 'python'

scripts/py_matter_yamltests/matter_yamltests/fixes.py

@@ -102,6 +102,28 @@ def convert_yaml_octet_string_to_bytes(s: str) -> bytes:
     return binascii.unhexlify(accumulated_hex)
 
 
+def fix_typed_yaml_value(value):
+    """Applies fixups to typed runtime variables if necessary."""
+    if type(value) is dict:
+        mapping_type = value.get('type')
+        default_value = value.get('defaultValue')
+        if mapping_type is not None and default_value is not None:
+            value = default_value
+            if mapping_type == 'int64u' or mapping_type == 'int64s' or mapping_type == 'bitmap64' or mapping_type == 'epoch_us':
+                value = try_apply_float_to_integer_fix(value)
+                value = try_apply_yaml_cpp_longlong_limitation_fix(value)
+                value = try_apply_yaml_unrepresentable_integer_for_javascript_fixes(value)
+            elif mapping_type == 'single' or mapping_type == 'double':
+                value = try_apply_yaml_float_written_as_strings(value)
+            elif isinstance(value, float) and mapping_type != 'single' and mapping_type != 'double':
+                value = try_apply_float_to_integer_fix(value)
+            elif mapping_type == 'octet_string' or mapping_type == 'long_octet_string':
+                value = convert_yaml_octet_string_to_bytes(value)
+            elif mapping_type == 'boolean':
+                value = bool(value)
+    return value
+
+
 def add_yaml_support_for_scientific_notation_without_dot(loader):
     regular_expression = re.compile(u'''^(?:
      [-+]?(?:[0-9][0-9_]*)\\.[0-9_]*(?:[eE][-+]?[0-9]+)?

scripts/py_matter_yamltests/matter_yamltests/parser.py

@@ -21,7 +21,7 @@
 from typing import Optional
 
 from . import fixes
-from .constraints import get_constraints, is_typed_constraint
+from .constraints import get_constraints, is_typed_constraint, is_variable_aware_constraint
 from .definitions import SpecDefinitions
 from .errors import (TestStepEnumError, TestStepEnumSpecifierNotUnknownError, TestStepEnumSpecifierWrongError, TestStepError,
                      TestStepKeyError, TestStepValueNameError)
@@ -1157,7 +1157,7 @@ def _response_constraints_validation(self, expected_response, received_response,
 
             for constraint in constraints:
                 try:
-                    constraint.validate(received_value, response_type_name)
+                    constraint.validate(received_value, response_type_name, self._runtime_config_variable_storage)
                     result.success(check_type, error_success)
                 except TestStepError as e:
                     e.update_context(expected_response, self.step_index)
@@ -1204,6 +1204,8 @@ def _update_placeholder_values(self, containers):
 
                 if 'constraints' in item:
                     for constraint, constraint_value in item['constraints'].items():
+                        if is_variable_aware_constraint(constraint):
+                            continue
                         values[idx]['constraints'][constraint] = self._config_variable_substitution(
                             constraint_value)
 

scripts/py_matter_yamltests/matter_yamltests/yaml_loader.py

@@ -106,7 +106,8 @@
     'hasMasksSet': list,
     'hasMasksClear': list,
     'notValue': (type(None), bool, str, int, float, list, dict),
-    'anyOf': list
+    'anyOf': list,
+    'python': str,
 }
 
 # Note: this is not used in the loader, just provided for information in the schema tree

src/app/tests/suites/TestConstraints.yaml

@@ -19,6 +19,10 @@ config:
     cluster: "Unit Testing"
     endpoint: 1
 
+    AnOctetString:
+        type: octet_string
+        defaultValue: hex:deafbeef
+
 tests:
     - label: "Wait for the commissioned device to be retrieved"
       cluster: "DelayCommands"
@@ -52,6 +56,22 @@ tests:
           constraints:
               excludes: [0, 5]
 
+    - label: "Read attribute LIST With Python constraint"
+      command: "readAttribute"
+      attribute: "list_int8u"
+      response:
+          constraints:
+              python: len(value) == 4 and len(AnOctetString) == 4
+
+    - label: "Read attribute LIST With multi-line Python constraint"
+      command: "readAttribute"
+      attribute: "list_int8u"
+      response:
+          constraints:
+              python: |
+                  print("Hello from Python")
+                  return True
+
     - label: "Write attribute LIST Back to Default Value"
       command: "writeAttribute"
       attribute: "list_int8u"