[chiptool.py] Add --trace_file, --trace_decode, --node_id, --continue…

…OnFailure, --timeout, --delayInMs supports (#26333)

scripts/py_matter_yamltests/matter_yamltests/parser.py

@@ -962,8 +962,8 @@ def __init__(self, test_file: str, parser_config: TestParserConfig = TestParserC
  yaml_loader = YamlLoader()
  filename, name, pics, config, tests = yaml_loader.load(test_file)
 
- self.__apply_config_override(config, parser_config.config_override)
  self.__apply_legacy_config(config)
+ self.__apply_config_override(config, parser_config.config_override)
 
  self.filename = filename
  self.name = name
@@ -974,12 +974,21 @@ def __init__(self, test_file: str, parser_config: TestParserConfig = TestParserC
  PICSChecker(parser_config.pics),
  tests
  )
+ self.timeout = config['timeout']
 
  def __apply_config_override(self, config, config_override):
  for key, value in config_override.items():
- if value is None:
+ if value is None or not key in config:
  continue
 
+ if type(value) is str:
+ if key == 'timeout' or key == 'endpoint':
+ value = int(value)
+ elif key == 'nodeId' and value.startswith('0x'):
+ value = int(value, 16)
+ elif key == 'nodeId':
+ value = int(value)
+
  if isinstance(config[key], dict) and 'defaultValue' in config[key]:
  config[key]['defaultValue'] = value
  else:

scripts/py_matter_yamltests/matter_yamltests/runner.py

@@ -43,10 +43,14 @@ class TestRunnerOptions:
  stop running the tests if a step index matches
  the number. This is mostly useful when running
  a single test file and for debugging purposes.
+
+ delay_in_ms: If set to any value that is not zero the runner will
+ wait for the given time between steps.
  """
  stop_on_error: bool = True
  stop_on_warning: bool = False
  stop_at_number: int = -1
+ delay_in_ms: int = 0
 
 
 @dataclass
@@ -139,7 +143,9 @@ def run(self, parser_builder_config: TestParserBuilderConfig, runner_config: Tes
  if not parser or not runner_config:
  continue
 
- result = asyncio.run(self._run(parser, runner_config))
+ loop = asyncio.get_event_loop()
+ result = loop.run_until_complete(asyncio.wait_for(
+ self._run(parser, runner_config), parser.timeout))
  if isinstance(result, Exception):
  raise (result)
  elif not result:
@@ -200,6 +206,9 @@ async def _run(self, parser: TestParser, config: TestRunnerConfig):
  if (idx + 1) == config.options.stop_at_number:
  break
 
+ if config.options.delay_in_ms:
+ await asyncio.sleep(config.options.delay_in_ms / 1000)
+
  hooks.test_stop(round(test_duration))
 
  except Exception as exception:

scripts/tests/yaml/chiptool.py

@@ -79,6 +79,14 @@ def chiptool_runner_options(f):
  help='Name of a websocket server to run at launch.')(f)
  f = click.option('--server_arguments', type=str, default='interactive server',
  help='Optional arguments to pass to the websocket server at launch.')(f)
+ f = click.option('--trace_file', type=click.Path(), default=None,
+ help='Optional file path to save the tracing output to.')(f)
+ f = click.option('--trace_decode', type=bool, default=True,
+ help='Decode the tracing ouput to a human readable format.')(f)
+ f = click.option('--delay-in-ms', '--delayInMs', type=int, default=0, show_default=True,
+ help='Add a delay between each test suite steps.')(f)
+ f = click.option('--continueOnFailure', type=bool, default=False, show_default=True,
+ help='Do not stop running the test suite on first error.')(f)
  f = click.option('--PICS', type=click.Path(exists=True), show_default=True, default=_DEFAULT_PICS_FILE,
  help='Path to the PICS file to use.')(f)
  return f
@@ -88,13 +96,36 @@ def chiptool_runner_options(f):
 CONTEXT_SETTINGS['default_map']['chiptool']['use_test_harness_log_format'] = True
 
 
+def maybe_update_server_arguments(ctx):
+ if ctx.params['trace_file']:
+ ctx.params['server_arguments'] += ' --trace_file {}'.format(ctx.params['trace_file'])
+
+ if ctx.params['trace_decode']:
+ ctx.params['server_arguments'] += ' --trace_decode 1'
+
+ del ctx.params['trace_file']
+ del ctx.params['trace_decode']
+
+ return ctx.params['server_arguments']
+
+
+def maybe_update_stop_on_error(ctx):
+ if ctx.params['continueonfailure']:
+ ctx.params['stop_on_error'] = False
+
+ del ctx.params['continueonfailure']
+
+
 @click.command(context_settings=CONTEXT_SETTINGS)
 @click.argument('commands', nargs=-1)
 @chiptool_runner_options
 @click.pass_context
-def chiptool_py(ctx, commands: list[str], server_path: str, server_name: str, server_arguments: str, pics: str):
+def chiptool_py(ctx, commands: list[str], server_path: str, server_name: str, server_arguments: str, trace_file: str, trace_decode: bool, delay_in_ms: int, continueonfailure: bool, pics: str):
  success = False
 
+ server_arguments = maybe_update_server_arguments(ctx)
+ maybe_update_stop_on_error(ctx)
+
  if len(commands) > 1 and commands[0] == 'tests':
  success = send_yaml_command(commands[1], server_path, server_arguments, pics, commands[2:])
  else:

scripts/tests/yaml/runner.py

@@ -73,6 +73,8 @@ def test_runner_options(f):
  help='Show additional logs provided by the adapter on error.')(f)
  f = click.option('--use_test_harness_log_format', type=bool, default=False, show_default=True,
  help='Use the test harness log format.')(f)
+ f = click.option('--delay-in-ms', type=int, default=0, show_default=True,
+ help='Add a delay between test suite steps.')(f)
  return f
 
 
@@ -263,10 +265,10 @@ def dry_run(parser_group: ParserGroup):
 @runner_base.command()
 @test_runner_options
 @pass_parser_group
-def run(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_warning: bool, stop_at_number: int, show_adapter_logs: bool, show_adapter_logs_on_error: bool, use_test_harness_log_format: bool):
+def run(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_warning: bool, stop_at_number: int, show_adapter_logs: bool, show_adapter_logs_on_error: bool, use_test_harness_log_format: bool, delay_in_ms: int):
  """Run the test suite."""
  adapter = __import__(adapter, fromlist=[None]).Adapter(parser_group.builder_config.parser_config.definitions)
- runner_options = TestRunnerOptions(stop_on_error, stop_on_warning, stop_at_number)
+ runner_options = TestRunnerOptions(stop_on_error, stop_on_warning, stop_at_number, delay_in_ms)
  runner_hooks = TestRunnerLogger(show_adapter_logs, show_adapter_logs_on_error, use_test_harness_log_format)
  runner_config = TestRunnerConfig(adapter, parser_group.pseudo_clusters, runner_options, runner_hooks)
 
@@ -278,10 +280,10 @@ def run(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_wa
 @test_runner_options
 @websocket_runner_options
 @pass_parser_group
-def websocket(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_warning: bool, stop_at_number: int, show_adapter_logs: bool, show_adapter_logs_on_error: bool, use_test_harness_log_format: bool, server_address: str, server_port: int, server_path: str, server_name: str, server_arguments: str):
+def websocket(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_warning: bool, stop_at_number: int, show_adapter_logs: bool, show_adapter_logs_on_error: bool, use_test_harness_log_format: bool, delay_in_ms: int, server_address: str, server_port: int, server_path: str, server_name: str, server_arguments: str):
  """Run the test suite using websockets."""
  adapter = __import__(adapter, fromlist=[None]).Adapter(parser_group.builder_config.parser_config.definitions)
- runner_options = TestRunnerOptions(stop_on_error, stop_on_warning, stop_at_number)
+ runner_options = TestRunnerOptions(stop_on_error, stop_on_warning, stop_at_number, delay_in_ms)
  runner_hooks = TestRunnerLogger(show_adapter_logs, show_adapter_logs_on_error, use_test_harness_log_format)
  runner_config = TestRunnerConfig(adapter, parser_group.pseudo_clusters, runner_options, runner_hooks)
 
@@ -301,10 +303,10 @@ def websocket(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop
 @test_runner_options
 @chip_repl_runner_options
 @pass_parser_group
-def chip_repl(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_warning: bool, stop_at_number: int, show_adapter_logs: bool, show_adapter_logs_on_error: bool, use_test_harness_log_format: bool, runner: str, repl_storage_path: str, commission_on_network_dut: bool):
+def chip_repl(parser_group: ParserGroup, adapter: str, stop_on_error: bool, stop_on_warning: bool, stop_at_number: int, show_adapter_logs: bool, show_adapter_logs_on_error: bool, use_test_harness_log_format: bool, delay_in_ms: int, runner: str, repl_storage_path: str, commission_on_network_dut: bool):
  """Run the test suite using chip-repl."""
  adapter = __import__(adapter, fromlist=[None]).Adapter(parser_group.builder_config.parser_config.definitions)
- runner_options = TestRunnerOptions(stop_on_error, stop_on_warning, stop_at_number)
+ runner_options = TestRunnerOptions(stop_on_error, stop_on_warning, stop_at_number, delay_in_ms)
  runner_hooks = TestRunnerLogger(show_adapter_logs, show_adapter_logs_on_error, use_test_harness_log_format)
  runner_config = TestRunnerConfig(adapter, parser_group.pseudo_clusters, runner_options, runner_hooks)