Automated implementation of 1 python test to use Metadata script as a…

…n intermediate argument holder

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

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')

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