Fix code violation and add code to chip_tool for python runner

app/chip_tool/chip_tool.py

@@ -266,9 +266,6 @@ async def start_chip_server(
         elif test_type == ChipToolTestType.CHIP_APP:
             prefix = CHIP_APP_EXE
             command = ["--interactive", "--port 9002"]
-        elif test_type == ChipToolTestType.PYTHON_TEST:
-            # TODO - To be implemented
-            pass
         else:
             raise ChipToolUnknownTestType(f"Unsupported Test Type: {test_type}")
 
@@ -349,46 +346,47 @@ def __get_gateway_ip(self) -> str:
             .get(DOCKER_GATEWAY_KEY, "")
         )
 
-    async def start_container(
-        self, test_type: ChipToolTestType, use_paa_certs: bool = False
-    ) -> None:
-        """Creates the chip-tool container.
-
-        Returns only when the container is created and all chip-tool services start.
+    async def start_container_no_server(self) -> None:
+        """
+        Creates the chip-tool container without any service running
+        (ChipTool or ChipApp).
         """
-
-        if self.is_running():
-            self.logger.info(
-                "chip-tool container already running, no need to start a new container"
-            )
-            return
-
         # Ensure there's no existing container running using the same name.
         self.__destroy_existing_container()
-
         # Async return when the container is running
         self.__chip_tool_container = await container_manager.create_container(
             self.image_tag, self.run_parameters
         )
-
         # Reset any previous states
         self.__last_exec_id = None
         self.__pics_file_created = False
-
         # Generate new random node id for the DUT
         self.__reset_node_id()
         self.logger.info(f"New Node Id generated: {hex(self.node_id)}")
-
         self.logger.info(
             f"""
             chip-tool started: {self.container_name}
             with configuration: {self.run_parameters}
             """
         )
-
         # Server started is false after spinning up a new container.
         self.__server_started = False
 
+    async def start_container(
+        self, test_type: ChipToolTestType, use_paa_certs: bool = False
+    ) -> None:
+        """Creates the chip-tool container.
+
+        Returns only when the container is created and all chip-tool services start.
+        """
+        if self.is_running():
+            self.logger.info(
+                "chip-tool container already running, no need to start a new container"
+            )
+            return
+
+        await self.start_container_no_server()
+
         web_socket_config = WebSocketRunnerConfig()
         web_socket_config.server_address = self.__get_gateway_ip()
         self.__test_harness_runner = WebSocketRunner(config=web_socket_config)
@@ -602,9 +600,6 @@ async def run_test(
             adapter = ChipToolAdapter.Adapter(parser_config.definitions)
         elif test_type == ChipToolTestType.CHIP_APP:
             adapter = ChipAppAdapter.Adapter(parser_config.definitions)
-        elif test_type == ChipToolTestType.PYTHON_TEST:
-            # TODO - run_test() for python_test to be implemented
-            pass
         else:
             raise ChipToolUnknownTestType(f"Unsupported Test Type: {test_type}")
 

app/chip_tool/test_case.py

@@ -225,6 +225,7 @@ async def setup(self) -> None:
             )
 
         self.chip_tool = ChipTool()
+        await self.chip_tool.start_container_no_server()
 
         # Use test engine logger to log all events to test run.
         self.chip_tool.logger = test_engine_logger

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

@@ -15,7 +15,7 @@
 #
 import ast
 from pathlib import Path
-from typing import List, Tuple
+from typing import List
 
 from .python_test_models import PythonTest, PythonTestStep, PythonTestType
 

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

@@ -15,45 +15,52 @@
 #
 import os
 import re
-from multiprocessing import Process
 from multiprocessing.managers import BaseManager
 from typing import Any, Type, TypeVar
 
-# TODO check if this should be changed to  SDK python test specific entries
 from matter_chip_tool_adapter.decoder import MatterLog
-from matter_yamltests.hooks import TestRunnerHooks
 
-from app.chip_tool.chip_tool import ChipToolTestType
+from app.chip_tool.chip_tool import ChipTool, ChipToolTestType
 from app.chip_tool.test_case import ChipToolTest
+from app.models import TestCaseExecution
 from app.test_engine.logger import (
     CHIP_LOG_FORMAT,
     CHIPTOOL_LEVEL,
     logger,
     test_engine_logger,
 )
-from app.test_engine.models import ManualVerificationTestStep, TestCase, TestStep
+from app.test_engine.models import TestCase, TestStep
 
 from .python_test_models import PythonTest
 from .python_testing_hooks_proxy import SDKPythonTestRunnerHooks
 
 # Custom type variable used to annotate the factory method in PythonTestCase.
 T = TypeVar("T", bound="PythonTestCase")
 
+# Command line params
+RUNNER_CLASS = "external_runner.py"
+RUNNER_CLASS_PATH = "./testing2/"
+EXECUTABLE = "python3"
+
 
 class PythonTestCase(TestCase):
     """Base class for all Python Test based test cases.
 
     This class provides a class factory that will dynamically declare a new sub-class
     based on the test-type the Python test is expressing.
 
-    The PythonTest will be stored as a class property that will be used at run-time in all
-    instances of such subclass.
+    The PythonTest will be stored as a class property that will be used at run-time
+    in all instances of such subclass.
     """
 
     python_test: PythonTest
     python_test_version: str
     test_finished: bool
 
+    def __init__(self, test_case_execution: TestCaseExecution) -> None:
+        super().__init__(test_case_execution=test_case_execution)
+        self.chip_tool: ChipTool
+
     def reset(self) -> None:
         self.start_called = False
         self.stop_called = False
@@ -123,7 +130,7 @@ def __report_failures(self, logger: Any, request: TestStep, received: Any) -> No
             # It is expected the runner to return a PostProcessResponseResult,
             # but in case of returning a different type
             self.current_test_step.append_failure(
-                "Test Step Failure: \n " f"Expected: '<Empty>' \n Received:  '<Empty>'"
+                "Test Step Failure: \n Expected: '<Empty>' \n Received:  '<Empty>'"
             )
             return
 
@@ -147,7 +154,8 @@ def pics(cls) -> set[str]:
 
     @classmethod
     def default_test_parameters(cls) -> dict[str, Any]:
-        """Python Testing config dict, sometimes have a nested dict with type and default value.
+        """Python Testing config dict, sometimes have a nested dict with type and
+        default value.
         Only defaultValue is used in this case.
         """
         parameters = {}
@@ -164,6 +172,9 @@ async def setup(self) -> None:
         test_engine_logger.info(f"Python Test Version: {self.python_test_version}")
         try:
             await super().setup()
+            self.chip_tool = ChipTool()
+            await self.chip_tool.start_container_no_server()
+            assert self.chip_tool.is_running()
         except NotImplementedError:
             pass
 
@@ -221,37 +232,22 @@ def run_command(self, cmd: str) -> None:
     async def execute(self) -> None:
         try:
             logger.info("Running Python Test: " + self.metadata["title"])
-
-            # NOTE that this aproach invalidates  parallel  execution since test_case_instance object is shared  by the class
-            # TODO: Same approach could work from TestCase side: create test_case_instance inside PythonTestCase to avoid using SDKPythonTestRunnerHooks
-
             BaseManager.register("TestRunnerHooks", SDKPythonTestRunnerHooks)
             manager = BaseManager(address=("0.0.0.0", 50000), authkey=b"abc")
             manager.start()
-
             test_runner_hooks = manager.TestRunnerHooks()  # type: ignore
-
-            command = (
-                "docker run -it --network host --privileged"
-                " -v /var/paa-root-certs:/root/paa-root-certs"
-                " -v /var/run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket:rw"
-                " -v /home/ubuntu/chip-certification-tool/backend/sdk_content/python_testing2:/root/python_testing2"
-                " connectedhomeip/chip-cert-bins:19771ed7101321d68b87d05201d42d00adf5368f"
-                " python3 python_testing2/hello_external_runner.py "
-                f" {self.metadata['title']}"
+            runner_class = RUNNER_CLASS_PATH + RUNNER_CLASS
+            result = self.chip_tool.send_command(
+                f"{runner_class} {self.metadata['title']}", prefix=EXECUTABLE
             )
-            # Start the command in a new process
-            p = Process(target=self.run_command, args=(command,))
-            p.start()
-
             while ((update := test_runner_hooks.updates_test()) is not None) or (
                 not test_runner_hooks.finished()
             ):
                 if not update:
                     continue
 
                 def handle_update(update: dict) -> None:
-                    def call_function(obj, func_name: str, kwargs) -> None:  # type: ignore
+                    def call_function(obj, func_name, kwargs) -> None:  # type: ignore
                         func = getattr(obj, func_name, None)
                         if not func:
                             raise AttributeError(
@@ -266,12 +262,12 @@ def call_function(obj, func_name: str, kwargs) -> None:  # type: ignore
                         call_function(self, func_name, kwargs)
 
                 handle_update(update)
-
         finally:
             pass
 
     async def cleanup(self) -> None:
         logger.info("Test Cleanup")
+        self.chip_tool.destroy_device()
 
 
 class PythonChipToolTestCase(PythonTestCase, ChipToolTest):