Have CI use yamltests python parser to run using chip-repl (#24295)

* Have CI use yamltests python parser to run using chip-repl

This is done only for linux platform.

.github/workflows/tests.yaml

@@ -184,6 +184,7 @@ jobs:
             - name: Build Apps
               timeout-minutes: 45
               run: |
+                  scripts/run_in_build_env.sh './scripts/build_python.sh --install_wheel build-env'
                   ./scripts/run_in_build_env.sh \
                      "./scripts/build/build_examples.py \
                         --target linux-x64-chip-tool${CHIP_TOOL_VARIANT}-${BUILD_VARIANT} \
@@ -212,6 +213,23 @@ jobs:
                      --tv-app ./out/linux-x64-tv-app-${BUILD_VARIANT}/chip-tv-app \
                      --bridge-app ./out/linux-x64-bridge-${BUILD_VARIANT}/chip-bridge-app \
                   "
+            - name: Run Tests using chip-repl
+              timeout-minutes: 5
+              run: |
+                  ./scripts/run_in_build_env.sh \
+                  "./scripts/tests/run_test_suite.py \
+                     --chip-tool ./out/linux-x64-chip-tool${CHIP_TOOL_VARIANT}-${BUILD_VARIANT}/chip-tool \
+                     --run-yamltests-with-chip-repl \
+                     run \
+                     --iterations 1 \
+                     --test-timeout-seconds 120 \
+                     --all-clusters-app ./out/linux-x64-all-clusters-${BUILD_VARIANT}/chip-all-clusters-app \
+                     --lock-app ./out/linux-x64-lock-${BUILD_VARIANT}/chip-lock-app \
+                     --ota-provider-app ./out/linux-x64-ota-provider-${BUILD_VARIANT}/chip-ota-provider-app \
+                     --ota-requestor-app ./out/linux-x64-ota-requestor-${BUILD_VARIANT}/chip-ota-requestor-app \
+                     --tv-app ./out/linux-x64-tv-app-${BUILD_VARIANT}/chip-tv-app \
+                     --bridge-app ./out/linux-x64-bridge-${BUILD_VARIANT}/chip-bridge-app \
+                  "
             - name: Uploading core files
               uses: actions/upload-artifact@v3
               if: ${{ failure() && !env.ACT }}

scripts/tests/chiptest/__init__.py

@@ -14,11 +14,31 @@
 #    limitations under the License.
 #
 
+import os
 import subprocess
+from pathlib import Path
 
 from . import linux, runner
 from .test_definition import ApplicationPaths, TestDefinition, TestTarget
 
+_DEFAULT_CHIP_ROOT = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+_YAML_TEST_SUITE_PATH = os.path.abspath(
+    os.path.join(_DEFAULT_CHIP_ROOT, "src/app/tests/suites"))
+
+
+def _FindYamlTestPath(name: str):
+    yaml_test_suite_path = Path(_YAML_TEST_SUITE_PATH)
+    if not yaml_test_suite_path.exists():
+        raise FileNotFoundError(f"Expected directory {_YAML_TEST_SUITE_PATH} to exist")
+    for path in yaml_test_suite_path.rglob(name):
+        if not path.is_file():
+            continue
+        if path.name != name:
+            continue
+        return str(path)
+    return None
+
 
 def target_for_name(name: str):
     if name.startswith("TV_") or name.startswith("Test_TC_MC_") or name.startswith("Test_TC_LOWPOWER_") or name.startswith("Test_TC_KEYPADINPUT_") or name.startswith("Test_TC_APPLAUNCHER_") or name.startswith("Test_TC_MEDIAINPUT_") or name.startswith("Test_TC_WAKEONLAN_") or name.startswith("Test_TC_CHANNEL_") or name.startswith("Test_TC_MEDIAPLAYBACK_") or name.startswith("Test_TC_AUDIOOUTPUT_") or name.startswith("Test_TC_TGTNAV_") or name.startswith("Test_TC_APBSC_") or name.startswith("Test_TC_CONTENTLAUNCHER_") or name.startswith("Test_TC_ALOGIN_"):
@@ -53,7 +73,29 @@ def tests_with_command(chip_tool: str, is_manual: bool):
         )
 
 
-def AllTests(chip_tool: str):
+# TODO We will move away from hardcoded list of yamltests to run all file when yamltests
+# parser/runner reaches parity with the code gen version.
+def _hardcoded_python_yaml_tests():
+    currently_supported_yaml_tests = ["TestConstraints.yaml"]
+
+    for name in currently_supported_yaml_tests:
+        yaml_test_path = _FindYamlTestPath(name)
+        if not yaml_test_path:
+            raise FileNotFoundError(f"Could not find YAML test {name}")
+
+        target = target_for_name(name)
+
+        yield TestDefinition(
+            run_name=yaml_test_path, name=name, target=target, is_manual=False, use_chip_repl_yaml_tester=True
+        )
+
+
+def AllTests(chip_tool: str, run_yamltests_with_chip_repl: bool):
+    if run_yamltests_with_chip_repl:
+        for test in _hardcoded_python_yaml_tests():
+            yield test
+        return
+
     for test in tests_with_command(chip_tool, is_manual=False):
         yield test
 

scripts/tests/chiptest/linux.py

@@ -127,6 +127,35 @@ def CreateNamespacesForAppTest():
         logging.warn("Some addresses look to still be tentative")
 
 
+def RemoveNamespaceForAppTest():
+    """
+    Removes namespaces for a tool and app binaries previously created to simulate an
+    isolated network. This tears down what was created in CreateNamespacesForAppTest.
+    """
+    COMMANDS = [
+        "ip link set dev eth-ci down",
+        "ip link set dev eth-ci-switch down",
+        "ip addr del 10.10.10.5/24 dev eth-ci",
+
+        "ip link set br1 down",
+        "ip link delete br1",
+
+        "ip link delete eth-ci-switch",
+        "ip link delete eth-tool-switch",
+        "ip link delete eth-app-switch",
+
+        "ip netns del tool",
+        "ip netns del app",
+    ]
+
+    for command in COMMANDS:
+        logging.debug("Executing '%s'" % command)
+        if os.system(command) != 0:
+            breakpoint()
+            logging.error("Failed to execute '%s'" % command)
+            sys.exit(1)
+
+
 def PrepareNamespacesForTestExecution(in_unshare: bool):
     if not in_unshare:
         EnsureNetworkNamespaceAvailability()
@@ -136,6 +165,10 @@ def PrepareNamespacesForTestExecution(in_unshare: bool):
     CreateNamespacesForAppTest()
 
 
+def ShutdownNamespaceForTestExecution():
+    RemoveNamespaceForAppTest()
+
+
 def PathsWithNetworkNamespaces(paths: ApplicationPaths) -> ApplicationPaths:
     """
     Returns a copy of paths with updated command arrays to invoke the
@@ -149,4 +182,5 @@ def PathsWithNetworkNamespaces(paths: ApplicationPaths) -> ApplicationPaths:
         ota_requestor_app='ip netns exec app'.split() + paths.ota_requestor_app,
         tv_app='ip netns exec app'.split() + paths.tv_app,
         bridge_app='ip netns exec app'.split() + paths.bridge_app,
+        chip_repl_yaml_tester_cmd='ip netns exec tool'.split() + paths.chip_repl_yaml_tester_cmd,
     )

scripts/tests/chiptest/test_definition.py

@@ -165,9 +165,10 @@ class ApplicationPaths:
     ota_requestor_app: typing.List[str]
     tv_app: typing.List[str]
     bridge_app: typing.List[str]
+    chip_repl_yaml_tester_cmd: typing.List[str]
 
     def items(self):
-        return [self.chip_tool, self.all_clusters_app, self.lock_app, self.ota_provider_app, self.ota_requestor_app, self.tv_app, self.bridge_app]
+        return [self.chip_tool, self.all_clusters_app, self.lock_app, self.ota_provider_app, self.ota_requestor_app, self.tv_app, self.bridge_app, self.chip_repl_yaml_tester_cmd]
 
 
 @dataclass
@@ -215,6 +216,7 @@ class TestDefinition:
     run_name: str
     target: TestTarget
     is_manual: bool
+    use_chip_repl_yaml_tester: bool = False
 
     def Run(self, runner, apps_register, paths: ApplicationPaths, pics_file: str, timeout_seconds: typing.Optional[int], dry_run=False):
         """
@@ -238,8 +240,8 @@ def Run(self, runner, apps_register, paths: ApplicationPaths, pics_file: str, ti
                                 "don't know which application to run")
 
             for path in paths.items():
-                # Do not add chip-tool to the register
-                if path == paths.chip_tool:
+                # Do not add chip-tool or chip-repl-yaml-tester-cmd to the register
+                if path == paths.chip_tool or path == paths.chip_repl_yaml_tester_cmd:
                     continue
 
                 # For the app indicated by self.target, give it the 'default' key to add to the register
@@ -278,7 +280,12 @@ def Run(self, runner, apps_register, paths: ApplicationPaths, pics_file: str, ti
             if dry_run:
                 logging.info(" ".join(pairing_cmd))
                 logging.info(" ".join(test_cmd))
-
+            elif self.use_chip_repl_yaml_tester:
+                chip_repl_yaml_tester_cmd = paths.chip_repl_yaml_tester_cmd
+                python_cmd = chip_repl_yaml_tester_cmd + \
+                    ['--setup-code', app.setupCode] + ['--yaml-path', self.run_name]
+                runner.RunSubprocess(python_cmd, name='CHIP_REPL_YAML_TESTER',
+                                     dependencies=[apps_register], timeout_seconds=timeout_seconds)
             else:
                 runner.RunSubprocess(pairing_cmd,
                                      name='PAIR', dependencies=[apps_register])

scripts/tests/chiptest/yamltest_with_chip_repl_tester.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env -S python3 -B
+
+#    Copyright (c) 2022 Project CHIP Authors
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+
+import functools
+import glob
+import os
+import tempfile
+
+# isort: off
+
+from chip import ChipDeviceCtrl  # Needed before chip.FabricAdmin
+import chip.FabricAdmin  # Needed before chip.CertificateAuthority
+
+# isort: on
+
+import chip.CertificateAuthority
+import chip.native
+import click
+from chip.ChipStack import *
+from chip.yaml.runner import ReplTestRunner
+from matter_yamltests.definitions import ParseSource, SpecDefinitions
+from matter_yamltests.parser import TestParser
+
+_DEFAULT_CHIP_ROOT = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+_CLUSTER_XML_DIRECTORY_PATH = os.path.abspath(
+    os.path.join(_DEFAULT_CHIP_ROOT, "src/app/zap-templates/zcl/data-model/"))
+
+
+def _sort_with_global_attribute_first(a, b):
+    if a.endswith('global-attributes.xml'):
+        return -1
+    elif b.endswith('global-attributes.xml'):
+        return 1
+    elif a > b:
+        return 1
+    elif a == b:
+        return 0
+    elif a < b:
+        return -1
+
+
+@click.command()
+@click.option(
+    '--setup-code',
+    default=None,
+    help='setup-code')
+@click.option(
+    '--yaml-path',
+    default=None,
+    help='yaml-path')
+@click.option(
+    '--node-id',
+    default=0x12344321,
+    help='Node ID to use when commissioning device')
+def main(setup_code, yaml_path, node_id):
+    # Setting up python environment for running YAML CI tests using python parser.
+    with tempfile.NamedTemporaryFile() as chip_stack_storage:
+        chip.native.Init()
+        chip_stack = ChipStack(chip_stack_storage.name)
+        certificate_authority_manager = chip.CertificateAuthority.CertificateAuthorityManager(
+            chip_stack, chipStack.GetStorageManager())
+        certificate_authority_manager.LoadAuthoritiesFromStorage()
+
+        if len(certificate_authority_manager.activeCaList) == 0:
+            ca = certificate_authority_manager.NewCertificateAuthority()
+            ca.NewFabricAdmin(vendorId=0xFFF1, fabricId=1)
+        elif len(certificate_authority_manager.activeCaList[0].adminList) == 0:
+            certificate_authority_manager.activeCaList[0].NewFabricAdmin(vendorId=0xFFF1, fabricId=1)
+
+        ca_list = certificate_authority_manager.activeCaList
+
+        # Creating and commissioning to a single controller to match what is currently done when
+        # running.
+        dev_ctrl = ca_list[0].adminList[0].NewController()
+        dev_ctrl.CommissionWithCode(setup_code, node_id)
+
+        # Creating Cluster definition.
+        cluster_xml_filenames = glob.glob(_CLUSTER_XML_DIRECTORY_PATH + '/*/*.xml', recursive=False)
+        cluster_xml_filenames.sort(key=functools.cmp_to_key(_sort_with_global_attribute_first))
+        sources = [ParseSource(source=name) for name in cluster_xml_filenames]
+        clusters_definitions = SpecDefinitions(sources)
+
+        # Parsing YAML test and setting up chip-repl yamltests runner.
+        yaml = TestParser(yaml_path, None, clusters_definitions)
+        runner = ReplTestRunner(clusters_definitions, certificate_authority_manager)
+
+        # Executing and validating test
+        for test_step in yaml.tests:
+            test_action = runner.encode(test_step)
+            # TODO if test_action is None we should see if it is a pseudo cluster.
+            if test_action is not None:
+                response = runner.execute(test_action)
+                decoded_response = runner.decode(response)
+                post_processing_result = test_step.post_process_response(decoded_response)
+                if not post_processing_result.is_success():
+                    # TODO figure out how we error out here
+                    pass
+
+        runner.shutdown()
+        # Tearing down chip stack. If not done in the correct order test will fail.
+        certificate_authority_manager.Shutdown()
+        chip_stack.Shutdown()
+
+
+if __name__ == '__main__':
+    main()

scripts/tests/run_test_suite.py

@@ -105,12 +105,17 @@ class RunContext:
     default=False,
     help='Internal flag for running inside a unshared environment'
 )
+@click.option(
+    '--run-yamltests-with-chip-repl',
+    default=False,
+    is_flag=True,
+    help='Run YAML tests using chip-repl based python parser only')
 @click.option(
     '--chip-tool',
     help='Binary path of chip tool app to use to run the test')
 @click.pass_context
 def main(context, dry_run, log_level, target, target_glob, target_skip_glob,
-         no_log_timestamps, root, internal_inside_unshare, chip_tool):
+         no_log_timestamps, root, internal_inside_unshare, run_yamltests_with_chip_repl, chip_tool):
     # Ensures somewhat pretty logging of what is going on
     log_fmt = '%(asctime)s.%(msecs)03d %(levelname)-7s %(message)s'
     if no_log_timestamps:
@@ -121,7 +126,7 @@ def main(context, dry_run, log_level, target, target_glob, target_skip_glob,
         chip_tool = FindBinaryPath('chip-tool')
 
     # Figures out selected test that match the given name(s)
-    all_tests = [test for test in chiptest.AllTests(chip_tool)]
+    all_tests = [test for test in chiptest.AllTests(chip_tool, run_yamltests_with_chip_repl)]
 
     # Default to only non-manual tests unless explicit targets are specified.
     tests = list(filter(lambda test: not test.is_manual, all_tests))
@@ -189,6 +194,9 @@ def cmd_list(context):
 @click.option(
     '--bridge-app',
     help='what bridge app to use')
+@click.option(
+    '--chip-repl-yaml-tester',
+    help='what python script to use for running yaml tests using chip-repl as controller')
 @click.option(
     '--pics-file',
     type=click.Path(exists=True),
@@ -200,7 +208,7 @@ def cmd_list(context):
     type=int,
     help='If provided, fail if a test runs for longer than this time')
 @click.pass_context
-def cmd_run(context, iterations, all_clusters_app, lock_app, ota_provider_app, ota_requestor_app, tv_app, bridge_app, pics_file, test_timeout_seconds):
+def cmd_run(context, iterations, all_clusters_app, lock_app, ota_provider_app, ota_requestor_app, tv_app, bridge_app, chip_repl_yaml_tester, pics_file, test_timeout_seconds):
     runner = chiptest.runner.Runner()
 
     if all_clusters_app is None:
@@ -221,6 +229,9 @@ def cmd_run(context, iterations, all_clusters_app, lock_app, ota_provider_app, o
     if bridge_app is None:
         bridge_app = FindBinaryPath('chip-bridge-app')
 
+    if chip_repl_yaml_tester is None:
+        chip_repl_yaml_tester = FindBinaryPath('yamltest_with_chip_repl_tester.py')
+
     # Command execution requires an array
     paths = chiptest.ApplicationPaths(
         chip_tool=[context.obj.chip_tool],
@@ -229,7 +240,8 @@ def cmd_run(context, iterations, all_clusters_app, lock_app, ota_provider_app, o
         ota_provider_app=[ota_provider_app],
         ota_requestor_app=[ota_requestor_app],
         tv_app=[tv_app],
-        bridge_app=[bridge_app]
+        bridge_app=[bridge_app],
+        chip_repl_yaml_tester_cmd=['python3'] + [chip_repl_yaml_tester]
     )
 
     if sys.platform == 'linux':
@@ -262,6 +274,8 @@ def cmd_run(context, iterations, all_clusters_app, lock_app, ota_provider_app, o
                 sys.exit(2)
 
     apps_register.uninit()
+    if sys.platform == 'linux':
+        chiptest.linux.ShutdownNamespaceForTestExecution()
 
 
 # On linux, allow an execution shell to be prepared