Automated implementation of initially 1 python test to use Metadata script as an intermediate argument holder

.github/workflows/tests.yaml

@@ -475,6 +475,16 @@ jobs:
                       build \
                       --copy-artifacts-to objdir-clone \
                    "
+            - name: Generate an argument environment file
+              run: |
+                  echo -n "" >/tmp/test_env.yaml
+                  echo "ALL_CLUSTERS_APP: out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app" >> /tmp/test_env.yaml
+                  echo "CHIP_LOCK_APP: out/linux-x64-lock-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-lock-app" >> /tmp/test_env.yaml
+                  echo "ENERGY_MANAGEMENT_APP: out/linux-x64-energy-management-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-energy-management-app" >> /tmp/test_env.yaml
+                  echo "TRACE_APP: out/trace_data/app-{SCRIPT_BASE_NAME}" >> /tmp/test_env.yaml
+                  echo "TRACE_TEST_JSON: out/trace_data/test-{SCRIPT_BASE_NAME}" >> /tmp/test_env.yaml
+                  echo "TRACE_TEST_PERFETTO: out/trace_data/test-{SCRIPT_BASE_NAME}" >> /tmp/test_env.yaml
+
             - name: Run Tests
               run: |
                   mkdir -p out/trace_data
@@ -516,7 +526,7 @@ jobs:
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app --factoryreset --quiet --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json --enable-key 000102030405060708090a0b0c0d0e0f" --script "src/python_testing/TC_IDM_1_4.py" --script-args "--hex-arg PIXIT.DGGEN.TEST_EVENT_TRIGGER_KEY:000102030405060708090a0b0c0d0e0f --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app  --factoryreset --quiet --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_PWRTL_2_1.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app  --factoryreset --quiet --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_RR_1_1.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
-                  scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app  --factoryreset --quiet --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_SC_3_6.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
+                  scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_SC_3_6.py'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app  --factoryreset --quiet --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_TIMESYNC_2_1.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app  --factoryreset --quiet --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_TIMESYNC_2_10.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'
                   scripts/run_in_python_env.sh out/venv './scripts/tests/run_python_test.py --app out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app  --factoryreset --quiet --app-args "--discriminator 1234 --KVS kvs1 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json" --script "src/python_testing/TC_TIMESYNC_2_11.py" --script-args "--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --PICS src/app/tests/suites/certification/ci-pics-values --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto"'

scripts/tests/py/metadata.py

@@ -15,51 +15,64 @@
 
 import re
 from dataclasses import dataclass
-from typing import Dict, List, Optional
+from typing import Any, Dict, List
 
 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 +83,30 @@ 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.
-        """
-
-        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]:
+         Dictionary where each key represent a particular argument and its value represent
+         the value for that argument defined in the test script.
         """
-        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
+        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
 
     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/py/test_metadata.py

@@ -12,39 +12,60 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
 import tempfile
 import unittest
-from os import path
-from typing import List
 
 from metadata import Metadata, MetadataReader
 
 
 class TestMetadataReader(unittest.TestCase):
 
-    def setUp(self):
-        # build the reader object
-        self.reader = MetadataReader(path.join(path.dirname(__file__), "env_test.yaml"))
+    test_file_content = ''' 
+    # test-runner-runs: run1 
+    # test-runner-run/run1/app: ${ALL_CLUSTERS_APP}
+    # test-runner-run/run1/app-args: --discriminator 1234 --trace-to json:${TRACE_APP}.json
+    # test-runner-run/run1/script-args: --commissioning-method on-network --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+    # test-runner-run/run1/factoryreset: True
+    # test-runner-run/run1/quiet: True
+    '''
 
-    def assertMetadataParse(self, file_content: str, expected: List[Metadata]):
-        with tempfile.NamedTemporaryFile(mode='w', delete=False) as fp:
+    env_file_content = '''
+    ALL_CLUSTERS_APP: out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app
+    CHIP_LOCK_APP: out/linux-x64-lock-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-lock-app
+    ENERGY_MANAGEMENT_APP: out/linux-x64-energy-management-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-energy-management-app
+    TRACE_APP: out/trace_data/app-{SCRIPT_BASE_NAME}
+    TRACE_TEST_JSON: out/trace_data/test-{SCRIPT_BASE_NAME}
+    TRACE_TEST_PERFETTO: out/trace_data/test-{SCRIPT_BASE_NAME}
+    '''
+
+    expected_metadata = Metadata(
+        script_args="--commissioning-method on-network --trace-to json:out/trace_data/test-{SCRIPT_BASE_NAME}.json --trace-to perfetto:out/trace_data/test-{SCRIPT_BASE_NAME}.perfetto",
+        py_script_path="",
+        app_args="--discriminator 1234 --trace-to json:out/trace_data/app-{SCRIPT_BASE_NAME}.json",
+        run="run1",
+        app="out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app",
+        factoryreset=True,
+        quiet=True
+    )
+
+    def generate_temp_file(self, directory: str, file_content: str) -> str:
+        fd, temp_file_path = tempfile.mkstemp(dir=directory)
+        with os.fdopen(fd, 'w') as fp:
             fp.write(file_content)
-            fp.close()
-            for e in expected:
-                e.py_script_path = fp.name
-            actual = self.reader.parse_script(fp.name)
-            self.assertEqual(actual, expected)
-
-    def test_parse_single_run(self):
-        self.assertMetadataParse(''' 
-            # test-runner-runs: run1 
-            # test-runner-run/run1: app/all-clusters discriminator passcode
-            ''',
-                                 [
-                                     Metadata(app="out/linux-x64-all-clusters-ipv6only-no-ble-no-wifi-tsan-clang-test/chip-all-clusters-app",
-                                              discriminator=1234, run="run1", passcode=20202021)
-                                 ]
-                                 )
+        return temp_file_path
+
+    def test_run_arg_generation(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_file = self.generate_temp_file(temp_dir, self.test_file_content)
+            env_file = self.generate_temp_file(temp_dir, self.env_file_content)
+
+            reader = MetadataReader(env_file)
+            self.maxDiff = None
+
+            self.expected_metadata.py_script_path = temp_file
+            actual = reader.parse_script(temp_file)[0]
+            self.assertEqual(self.expected_metadata, actual)
 
 
 if __name__ == "__main__":