Yaml Test Runner: Add support for Python constraints

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(f'Python constraint validation did not 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,24 @@ 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"