From a59bac690275aae4992ec338e20d3cf4dd133e2e Mon Sep 17 00:00:00 2001 From: Vatsal Ghelani Date: Wed, 12 Jun 2024 17:04:56 -0400 Subject: [PATCH] Automated implementation of 1 python test to use Metadata script as an intermediate argument holder --- scripts/tests/py/metadata.py | 186 +++++++++++-------------------- scripts/tests/run_python_test.py | 33 +++++- src/python_testing/TC_SC_3_6.py | 7 ++ 3 files changed, 105 insertions(+), 121 deletions(-) diff --git a/scripts/tests/py/metadata.py b/scripts/tests/py/metadata.py index 0445b3a31b4a6e..2f758c629a313e 100644 --- a/scripts/tests/py/metadata.py +++ b/scripts/tests/py/metadata.py @@ -15,51 +15,63 @@ import re from dataclasses import dataclass -from typing import Dict, List, Optional - +from typing import Dict, List, Optional, Any import yaml @dataclass class Metadata: - py_script_path: Optional[str] = None - run: Optional[str] = None - app: Optional[str] = None - discriminator: Optional[str] = None - passcode: Optional[str] = None - - def copy_from_dict(self, attr_dict: Dict[str, str]) -> None: + py_script_path: str + run: str + app: str + app_args: str + script_args: str + factoryreset: bool = False + factoryreset_app_only: bool = False + script_gdb: bool = False + quiet: bool = True + + def copy_from_dict(self, attr_dict: Dict[str, Any]) -> None: """ Sets the value of the attributes from a dictionary. Attributes: attr_dict: - Dictionary that stores attributes value that should - be transferred to this class. + Dictionary that stores attributes value that should + be transferred to this class. """ - if "app" in attr_dict: self.app = attr_dict["app"] if "run" in attr_dict: self.run = attr_dict["run"] - if "discriminator" in attr_dict: - self.discriminator = attr_dict["discriminator"] + if "app-args" in attr_dict: + self.app_args = attr_dict["app-args"] - if "passcode" in attr_dict: - self.passcode = attr_dict["passcode"] + if "script-args" in attr_dict: + self.script_args = attr_dict["script-args"] if "py_script_path" in attr_dict: self.py_script_path = attr_dict["py_script_path"] - # TODO - set other attributes as well + if "factoryreset" in attr_dict: + self.factoryreset = bool(attr_dict["factoryreset"]) + + if "factoryreset_app_only" in attr_dict: + self.factoryreset_app_only = bool(attr_dict["factoryreset_app_only"]) + + if "script_gdb" in attr_dict: + self.script_gdb = bool(attr_dict["script_gdb"]) + + if "quiet" in attr_dict: + self.quiet = bool(attr_dict["quiet"]) class MetadataReader: """ - A class to parse run arguments from the test scripts and + A class to parse run arguments from the test scripts and resolve them to environment specific values. """ @@ -70,97 +82,31 @@ def __init__(self, env_yaml_file_path: str): Parameters: env_yaml_file_path: - Path to the environment file that contains the YAML configuration. + Path to the environment file that contains the YAML configuration. """ with open(env_yaml_file_path) as stream: - self.env = yaml.safe_load(stream) + self.env: Dict[str, str] = yaml.safe_load(stream) def __resolve_env_vals__(self, metadata_dict: Dict[str, str]) -> None: """ Resolves the argument defined in the test script to environment values. For example, if a test script defines "all_clusters" as the value for app name, we will check the environment configuration to see what raw value is - assocaited with the "all_cluster" variable and set the value for "app" option + associated with the "all_cluster" variable and set the value for "app" option to this raw value. Parameter: metadata_dict: - Dictionary where each key represent a particular argument and its value represent - the value for that argument defined in the test script. + Dictionary where each key represent a particular argument and its value represent + the value for that argument defined in the test script. """ + for arg, arg_val in metadata_dict.items(): + # We do not expect to recurse (like ${FOO_${BAR}}) so just expand once + for name, value in self.env.items(): + arg_val = arg_val.replace(f'${{{name}}}', value) + metadata_dict[arg] = arg_val - for run_arg, run_arg_val in metadata_dict.items(): - - if not type(run_arg_val) == str or run_arg == "run": - metadata_dict[run_arg] = run_arg_val - continue - - if run_arg_val is None: - continue - - sub_args = run_arg_val.split('/') - - if len(sub_args) not in [1, 2]: - err = """The argument is not in the correct format. - The argument must follow the format of arg1 or arg1/arg2. - For example, arg1 represents the argument type and optionally arg2 - represents a specific variable defined in the environment file whose - value should be used as the argument value. If arg2 is not specified, - we will just use the first value associated with arg1 in the environment file.""" - raise Exception(err) - - if len(sub_args) == 1: - run_arg_val = self.env.get(sub_args[0]) - - elif len(sub_args) == 2: - run_arg_val = self.env.get(sub_args[0]).get(sub_args[1]) - - # if a argument has been specified in the comment header - # but can't be found in the env file, consider it to be - # boolean value. - if run_arg_val is None: - run_arg_val = True - - metadata_dict[run_arg] = run_arg_val - - def __read_args__(self, run_args_lines: List[str]) -> Dict[str, str]: - """ - Parses a list of lines and extracts argument - values from it. - - Parameters: - - run_args_lines: - Line in test script header that contains run argument definition. - Each line will contain a list of run arguments separated by a space. - Line below is one example of what the run argument line will look like: - "app/all-clusters discriminator KVS storage-path" - - In this case the line defines that app, discriminator, KVS, and storage-path - are the arguments that should be used with this run. - - An argument can be defined multiple times in the same line or in different lines. - The last definition will override any previous definition. For example, - "KVS/kvs1 KVS/kvs2 KVS/kvs3" line will lead to KVS value of kvs3. - """ - metadata_dict = {} - - for run_line in run_args_lines: - for run_arg_word in run_line.strip().split(): - ''' - We expect the run arg to be defined in one of the - following two formats: - 1. run_arg - 2. run_arg/run_arg_val - - Examples: "discriminator" and "app/all_clusters" - - ''' - run_arg = run_arg_word.split('/', 1)[0] - metadata_dict[run_arg] = run_arg_word - - return metadata_dict def parse_script(self, py_script_path: str) -> List[Metadata]: """ @@ -171,47 +117,51 @@ def parse_script(self, py_script_path: str) -> List[Metadata]: Parameter: py_script_path: - path to the python test script + path to the python test script Return: List[Metadata] - List of Metadata object where each Metadata element represents - the run arguments associated with a particular run defined in - the script file. + List of Metadata object where each Metadata element represents + the run arguments associated with a particular run defined in + the script file. """ runs_def_ptrn = re.compile(r'^\s*#\s*test-runner-runs:\s*(.*)$') - args_def_ptrn = re.compile(r'^\s*#\s*test-runner-run/([a-zA-Z0-9_]+):\s*(.*)$') + arg_def_ptrn = re.compile(r'^\s*#\s*test-runner-run/([a-zA-Z0-9_]+)/([a-zA-Z0-9_\-]+):\s*(.*)$') - runs_arg_lines: Dict[str, List[str]] = {} - runs_metadata = [] + runs_arg_lines: Dict[str, Dict[str, str]] = {} + runs_metadata: List[Metadata] = [] with open(py_script_path, 'r', encoding='utf8') as py_script: for line in py_script.readlines(): - runs_match = runs_def_ptrn.match(line.strip()) - args_match = args_def_ptrn.match(line.strip()) + args_match = arg_def_ptrn.match(line.strip()) if runs_match: for run in runs_match.group(1).strip().split(): - runs_arg_lines[run] = [] + runs_arg_lines[run] = {} + runs_arg_lines[run]['run'] = run + runs_arg_lines[run]['py_script_path'] = py_script_path elif args_match: - runs_arg_lines[args_match.group(1)].append(args_match.group(2)) - - for run, lines in runs_arg_lines.items(): - metadata_dict = self.__read_args__(lines) - self.__resolve_env_vals__(metadata_dict) - - # store the run value and script location in the - # metadata object - metadata_dict['py_script_path'] = py_script_path - metadata_dict['run'] = run - - metadata = Metadata() - - metadata.copy_from_dict(metadata_dict) + runs_arg_lines[args_match.group(1)][args_match.group(2)] = args_match.group(3) + + for run, attr in runs_arg_lines.items(): + self.__resolve_env_vals__(attr) + + metadata = Metadata( + py_script_path=attr.get("py_script_path", ""), + run=attr.get("run", ""), + app=attr.get("app", ""), + app_args=attr.get("app_args", ""), + script_args=attr.get("script_args", ""), + factoryreset=bool(attr.get("factoryreset", False)), + factoryreset_app_only=bool(attr.get("factoryreset_app_only", False)), + script_gdb=bool(attr.get("script_gdb", False)), + quiet=bool(attr.get("quiet", True)) + ) + metadata.copy_from_dict(attr) runs_metadata.append(metadata) return runs_metadata diff --git a/scripts/tests/run_python_test.py b/scripts/tests/run_python_test.py index 5d2afc31ae21b7..496287df572a0b 100755 --- a/scripts/tests/run_python_test.py +++ b/scripts/tests/run_python_test.py @@ -32,6 +32,7 @@ import click import coloredlogs from colorama import Fore, Style +from py.metadata import MetadataReader, Metadata DEFAULT_CHIP_ROOT = os.path.abspath( os.path.join(os.path.dirname(__file__), '..', '..')) @@ -89,7 +90,33 @@ def DumpProgramOutputToQueue(thread_list: typing.List[threading.Thread], tag: st @click.option("--script-gdb", is_flag=True, help='Run script through gdb') @click.option("--quiet", is_flag=True, help="Do not print output from passing tests. Use this flag in CI to keep github log sizes manageable.") -def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: str, script: str, script_args: str, script_gdb: bool, quiet: bool): +@click.option("--load-from-env", default=None, help="YAML file that contains values for environment variables.") +def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: str, script: str, script_args: str, script_gdb: bool, quiet: bool, load_from_env): + if load_from_env: + reader = MetadataReader(load_from_env) + runs = reader.parse_script(script) + else: + runs = [ + Metadata( + py_script_path=script, + run="cmd-run", + app=app, + app_args=app_args, + script_args=script_args, + factoryreset=factoryreset, + factoryreset_app_only=factoryreset_app_only, + script_gdb=script_gdb, + quiet=quiet + ) + ] + + for run in runs: + print(f"Executing run: {run.py_script_path}") + main_impl(run.app, run.factoryreset, run.factoryreset_app_only, run.app_args, run.py_script_path, run.script_args, run.script_gdb, run.quiet) + + +def main_impl(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: str, script: str, script_args: str, script_gdb: bool, quiet: bool): + app_args = app_args.replace('{SCRIPT_BASE_NAME}', os.path.splitext(os.path.basename(script))[0]) script_args = script_args.replace('{SCRIPT_BASE_NAME}', os.path.splitext(os.path.basename(script))[0]) @@ -192,8 +219,8 @@ def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: st else: logging.info("Test completed successfully") - sys.exit(exit_code) - + if exit_code != 0: + sys.exit(exit_code) if __name__ == '__main__': main(auto_envvar_prefix='CHIP') diff --git a/src/python_testing/TC_SC_3_6.py b/src/python_testing/TC_SC_3_6.py index ec09d4bb8e815e..2363afdc1e3c0a 100644 --- a/src/python_testing/TC_SC_3_6.py +++ b/src/python_testing/TC_SC_3_6.py @@ -15,6 +15,13 @@ # limitations under the License. # +# test-runner-runs: run1 +# test-runner-run/run1/app: ${ALL_CLUSTERS_APP} +# test-runner-run/run1/factoryreset: True +# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json +# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto + + import asyncio import logging import queue