Run multiple python tests in one execution

app/test_engine/logger.py

@@ -22,5 +22,9 @@
 CHIPTOOL_LEVEL = "CHIPTOOL"
 test_engine_logger.level(CHIPTOOL_LEVEL, no=21, icon="🤖", color="<cyan>")
 
+# Add custom logger for python tests
+PYTHON_TEST_LEVEL = "PYTHON_TEST"
+test_engine_logger.level(PYTHON_TEST_LEVEL, no=22, icon="🐍", color="<cyan>")
+
 # Log format for logs that come from CHIP. Arguments: module (e.g. "DMG"), message
 CHIP_LOG_FORMAT = "CHIP:{} {}"

test_collections/sdk_tests/support/python_testing/models/python_test_parser.py

@@ -73,23 +73,31 @@ def __parse_test_case_from_class(
     pics_method_name = "pics_" + tc_name
 
     methods = [m for m in class_.body if isinstance(m, ast.FunctionDef)]
+
+    tc_desc = tc_name
+    tc_steps = []
+    tc_pics = []
+
     try:
         desc_method = next(m for m in methods if desc_method_name in m.name)
         tc_desc = desc_method.body[BODY_INDEX].value.value  # type: ignore
+    except StopIteration:  # Raised when `next` doesn't find a matching method
+        pass
 
+    try:
         steps_method = next(m for m in methods if steps_method_name in m.name)
         tc_steps = __retrieve_steps(steps_method)
-    except StopIteration as si:  # Raised when `next` doesn't find a matching method
-        raise PythonParserException(
-            f"{path} did not contain valid definition for {tc_name}"
-        ) from si
+    except StopIteration:  # Raised when `next` doesn't find a matching method
+        # If the python test does not implement the steps template method,
+        # the test case will be presented in UI and the whole test case will be
+        # executed as one step
+        pass
 
-    # PICS method is optional
     try:
         pics_method = next(m for m in methods if pics_method_name in m.name)
         tc_pics = __retrieve_pics(pics_method)
     except StopIteration:  # Raised when `next` doesn't find a matching method
-        tc_pics = []
+        pass
 
     return PythonTest(
         name=tc_name,

test_collections/sdk_tests/support/python_testing/models/rpc_client/test_harness_client.py

@@ -21,7 +21,12 @@
 import sys
 from multiprocessing.managers import BaseManager
 
-import matter_testing_support
+from matter_testing_support import (
+    CommissionDeviceTest,
+    MatterTestConfig,
+    parse_matter_test_args,
+    run_tests,
+)
 
 
 class TestRunnerHooks:
@@ -31,14 +36,17 @@ class TestRunnerHooks:
 def main() -> None:
     # Load python_testing as a module. This folder is where all python script is located
     sys.path.append("/root/python_testing")
-    test_name = sys.argv[1]
 
-    config_options = sys.argv[2:]
-    config = matter_testing_support.parse_matter_test_args(config_options)
+    test_args = sys.argv[2:]
+    config = parse_matter_test_args(test_args)
 
-    if config is None:
-        raise ValueError(f"Not a valid test id: {test_name}")
+    if sys.argv[1] == "commission":
+        commission(config)
+    else:
+        run_test(test_name=sys.argv[1], config=config)
 
+
+def run_test(test_name: str, config: MatterTestConfig) -> None:
     module = importlib.import_module(test_name)
     TestClassReference = getattr(module, test_name)
 
@@ -47,7 +55,12 @@ def main() -> None:
     manager.connect()
     test_runner_hooks = manager.TestRunnerHooks()  # shared object proxy # type: ignore
 
-    matter_testing_support.run_tests(TestClassReference, config, test_runner_hooks)
+    run_tests(TestClassReference, config, test_runner_hooks)
+
+
+def commission(config: MatterTestConfig) -> None:
+    config.commission_only = True
+    run_tests(CommissionDeviceTest, config, None)
 
 
 if __name__ == "__main__":

test_collections/sdk_tests/support/python_testing/models/test_case.py

@@ -19,7 +19,6 @@
 from typing import Any, Type, TypeVar
 
 from app.models import TestCaseExecution
-from app.schemas.test_environment_config import DutPairingModeEnum
 from app.test_engine.logger import test_engine_logger as logger
 from app.test_engine.models import TestCase, TestStep
 from test_collections.sdk_tests.support.chip_tool.chip_tool import (
@@ -32,15 +31,11 @@
     SDKPythonTestResultBase,
     SDKPythonTestRunnerHooks,
 )
+from .utils import EXECUTABLE, RUNNER_CLASS_PATH, generate_command_arguments
 
 # Custom type variable used to annotate the factory method in PythonTestCase.
 T = TypeVar("T", bound="PythonTestCase")
 
-# Command line params
-RUNNER_CLASS = "test_harness_client.py"
-RUNNER_CLASS_PATH = "/root/python_testing/"
-EXECUTABLE = "python3"
-
 
 class PythonTestCase(TestCase):
     """Base class for all Python Test based test cases.
@@ -61,6 +56,15 @@ def __init__(self, test_case_execution: TestCaseExecution) -> None:
         self.__runned = 0
         self.test_stop_called = False
 
+    def next_step(self) -> None:
+        # Python tests that don't follow the template only have the default step "Start
+        # Python test", but inside the file there can be more than one test case, so the
+        # hooks steps methods will continue to be called
+        if len(self.test_steps) == 1:
+            return
+
+        super().next_step()
+
     def start(self, count: int) -> None:
         pass
 
@@ -146,58 +150,6 @@ async def setup(self) -> None:
     async def cleanup(self) -> None:
         logger.info("Test Cleanup")
 
-    def __generate_command_arguments(self) -> list:
-        # All valid arguments for python test
-        valid_args = [
-            "ble_interface_id",
-            "commissioning_method",
-            "controller_node_id",
-            "discriminator",
-            "endpoint",
-            "logs_path",
-            "PICS",
-            "paa_trust_store_path",
-            "timeout",
-            "trace_to",
-            "int_arg",
-            "float_arg",
-            "string_arg",
-            "json_arg",
-            "hex_arg",
-            "bool_arg",
-            "storage_path",
-            "passcode",
-            "dut-node-id",
-        ]
-
-        dut_config = self.project.config.dut_config
-        test_parameters = self.project.config.test_parameters
-
-        pairing_mode = (
-            "on-network"
-            if dut_config.pairing_mode == DutPairingModeEnum.ON_NETWORK.value
-            else dut_config.pairing_mode
-        )
-
-        arguments = []
-        # Retrieve arguments from dut_config
-        arguments.append(f"--discriminator {dut_config.discriminator}")
-        arguments.append(f"--passcode {dut_config.setup_code}")
-        arguments.append(f"--commissioning-method {pairing_mode}")
-
-        # Retrieve arguments from test_parameters
-        if test_parameters:
-            for name, value in test_parameters.items():
-                if name in valid_args:
-                    if str(value) != "":
-                        arguments.append(f"--{name.replace('_','-')} {str(value)}")
-                    else:
-                        arguments.append(f"--{name.replace('_','-')} " "")
-                else:
-                    logger.warning(f"Argument {name} is not valid")
-
-        return arguments
-
     async def execute(self) -> None:
         try:
             logger.info("Running Python Test: " + self.python_test.name)
@@ -207,12 +159,14 @@ async def execute(self) -> None:
             manager.start()
             test_runner_hooks = manager.TestRunnerHooks()  # type: ignore
 
-            runner_class = RUNNER_CLASS_PATH + RUNNER_CLASS
-            command = [f"{runner_class} {self.python_test.name}"]
+            command = [f"{RUNNER_CLASS_PATH} {self.python_test.name}"]
 
             # Generate the command argument by getting the test_parameters from
             # project configuration
-            command_arguments = self.__generate_command_arguments()
+            # comissioning method is omitted because it's handled by the test suite
+            command_arguments = generate_command_arguments(
+                config=self.config, omit_commissioning_method=True
+            )
             command.extend(command_arguments)
 
             if self.chip_tool.pics_file_created:

test_collections/sdk_tests/support/python_testing/models/test_suite.py

@@ -14,12 +14,19 @@
 # limitations under the License.
 #
 from enum import Enum
-from typing import Type, TypeVar
+from typing import Generator, Type, TypeVar, cast
 
 from app.test_engine.logger import test_engine_logger as logger
 from app.test_engine.models import TestSuite
 from test_collections.sdk_tests.support.chip_tool import ChipTool
 
+from .utils import (
+    EXECUTABLE,
+    RUNNER_CLASS_PATH,
+    generate_command_arguments,
+    handle_logs,
+)
+
 
 class SuiteType(Enum):
     AUTOMATED = 1
@@ -74,8 +81,25 @@ async def setup(self) -> None:
         else:
             self.chip_tool.reset_pics_state()
 
+        logger.info("Commission DUT")
+        self.commission_device()
+
     async def cleanup(self) -> None:
         logger.info("Suite Cleanup")
 
         logger.info("Stopping SDK container")
         await self.chip_tool.destroy_device()
+
+    def commission_device(self) -> None:
+        command = [f"{RUNNER_CLASS_PATH} commission"]
+        command_arguments = generate_command_arguments(config=self.config)
+        command.extend(command_arguments)
+
+        exec_result = self.chip_tool.send_command(
+            command,
+            prefix=EXECUTABLE,
+            is_stream=True,
+            is_socket=False,
+        )
+
+        handle_logs(cast(Generator, exec_result.output), logger)

test_collections/sdk_tests/support/python_testing/models/utils.py

@@ -0,0 +1,95 @@
+#
+# Copyright (c) 2023 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.
+#
+
+from __future__ import annotations
+
+from typing import Generator
+
+import loguru
+
+from app.schemas.test_environment_config import TestEnvironmentConfig
+from app.test_engine.logger import PYTHON_TEST_LEVEL
+from app.test_engine.logger import test_engine_logger as logger
+
+# Command line params
+RUNNER_CLASS_PATH = "/root/python_testing/test_harness_client.py"
+EXECUTABLE = "python3"
+
+
+def generate_command_arguments(
+    config: TestEnvironmentConfig, omit_commissioning_method: bool = False
+) -> list:
+    # All valid arguments for python test
+    valid_args = [
+        "ble_interface_id",
+        "commissioning_method",
+        "controller_node_id",
+        "discriminator",
+        "endpoint",
+        "logs_path",
+        "PICS",
+        "paa_trust_store_path",
+        "timeout",
+        "trace_to",
+        "int_arg",
+        "float_arg",
+        "string_arg",
+        "json_arg",
+        "hex_arg",
+        "bool_arg",
+        "storage_path",
+        "passcode",
+        "dut_node_id",
+        "qr_code",
+        "manual_code",
+    ]
+
+    dut_config = config.dut_config
+    test_parameters = config.test_parameters
+
+    pairing_mode = (
+        "on-network"
+        if dut_config.pairing_mode == "onnetwork"
+        else dut_config.pairing_mode
+    )
+
+    arguments = []
+    # Retrieve arguments from dut_config
+    arguments.append(f"--discriminator {dut_config.discriminator}")
+    arguments.append(f"--passcode {dut_config.setup_code}")
+    if not omit_commissioning_method:
+        arguments.append(f"--commissioning-method {pairing_mode}")
+
+    # Retrieve arguments from test_parameters
+    if test_parameters:
+        for name, value in test_parameters.items():
+            if name in valid_args:
+                if str(value) != "":
+                    arguments.append(f"--{name.replace('_','-')} {str(value)}")
+                else:
+                    arguments.append(f"--{name.replace('_','-')} " "")
+            else:
+                logger.warning(f"Argument {name} is not valid")
+
+    return arguments
+
+
+def handle_logs(log_generator: Generator, logger: loguru.Logger) -> None:
+    for chunk in log_generator:
+        decoded_log = chunk.decode().strip()
+        log_lines = decoded_log.splitlines()
+        for line in log_lines:
+            logger.log(PYTHON_TEST_LEVEL, line)