Add python tests model classes

app/chip_tool/chip_tool.py

@@ -107,6 +107,19 @@
 XML_SPEC_DEFINITION_PATH = TEST_COLLECTION_SDK_CHECKOUT_PATH / Path(
     "sdk_runner/specifications/chip/"
 )
+# Python Testing Folder
+LOCAL_TEST_COLLECTIONS_PATH = "/home/ubuntu/certification-tool/backend/test_collections"
+LOCAL_PYTHON_TESTING_PATH = Path(
+    LOCAL_TEST_COLLECTIONS_PATH + "/sdk_tests/sdk_checkout/python_testing/scripts/sdk"
+)
+DOCKER_PYTHON_TESTING_PATH = "/root/python_testing"
+
+# RPC Client Running on SDK Container
+LOCAL_RPC_PYTHON_TESTING_PATH = Path(
+    LOCAL_TEST_COLLECTIONS_PATH + "/sdk_tests/support/python_testing/models/rpc_client/"
+    "test_harness_client.py"
+)
+DOCKER_RPC_PYTHON_TESTING_PATH = "/root/python_testing/test_harness_client.py"
 
 
 # Docker Network
@@ -168,6 +181,14 @@ class ChipTool(metaclass=Singleton):
                 "bind": DOCKER_CREDENTIALS_DEVELOPMENT_PATH,
                 "mode": "ro",
             },
+            LOCAL_PYTHON_TESTING_PATH: {
+                "bind": DOCKER_PYTHON_TESTING_PATH,
+                "mode": "rw",
+            },
+            LOCAL_RPC_PYTHON_TESTING_PATH: {
+                "bind": DOCKER_RPC_PYTHON_TESTING_PATH,
+                "mode": "rw",
+            },
         },
     }
 
@@ -203,6 +224,10 @@ def __init__(
             specifications_paths, self.pseudo_clusters
         )
 
+    @property
+    def pics_file_created(self) -> bool:
+        return self.__pics_file_created
+
     @property
     def node_id(self) -> int:
         """Node id is used to reference DUT during testing.
@@ -347,14 +372,11 @@ 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(self) -> None:
+        """
+        Creates the chip-tool container without any server running
+        (ChipTool or ChipApp).
         """
-
         if self.is_running():
             self.logger.info(
                 "chip-tool container already running, no need to start a new container"
@@ -363,30 +385,34 @@ async def start_container(
 
         # 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_server(
+        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.
+        """
+        await self.start_container()
+
         web_socket_config = WebSocketRunnerConfig()
         web_socket_config.server_address = self.__get_gateway_ip()
         self.__test_harness_runner = WebSocketRunner(config=web_socket_config)
@@ -661,32 +687,37 @@ def __trace_file_params(self, topic: str) -> str:
         path = Path(DOCKER_LOGS_PATH) / filename
         return f'--trace_file "{path}" --trace_decode 1'
 
-    def set_pics(self, pics: PICS) -> None:
-        """Sends command to chip tool to create pics file inside the container.
+    def set_pics(self, pics: PICS, in_container: bool) -> None:
+        """Sends command to create pics file.
 
         Args:
             pics (PICS): PICS that contains all the pics codes
+            in_container (bool): Whether the file should be created in the SDK container
+                                or not. YAML tests run directly in the backend and
+                                python tests run in the SDK container.
 
         Raises:
-            ChipToolNotRunning: Raises exception if chip tool is not running.
             PICSError: If creating PICS file inside the container fails.
-
         """
         # List of default PICS which needs to set specifically in TH are added here.
         # These PICS are applicable for CI / Chip tool testing purposes only.
         # These PICS are unknown / not visible to external users.
 
         pics_codes = self.__pics_file_content(pics) + "\n".join(DEFAULT_PICS)
-        cmd = f"{SHELL_PATH} {SHELL_OPTION} "
-        cmd = cmd + f"\"{ECHO_COMMAND} '{pics_codes}\n' > {PICS_FILE_PATH}\""
-        self.logger.info(f"Sending command: {cmd}")
-        result = subprocess.run(cmd, shell=True)
 
-        # When streaming logs, the exit code is not directly available.
-        # By storing the execution id, the exit code can be fetched from docker later.
-        self.__last_exec_id = str(result.returncode)
+        prefix = f"{SHELL_PATH} {SHELL_OPTION}"
+        cmd = f"\"{ECHO_COMMAND} '{pics_codes}' > {PICS_FILE_PATH}\""
+
+        if in_container:
+            exec_result = self.send_command(cmd, prefix=prefix)
+            success = exec_result.exit_code == 0
+        else:
+            full_cmd = f"{prefix} {cmd}"
+            self.logger.info(f"Sending command: {full_cmd}")
+            result = subprocess.run(full_cmd, shell=True)
+            success = result.returncode == 0
 
-        if result.returncode != 0:
+        if not success:
             raise PICSError("Creating PICS file failed")
 
         self.__pics_file_created = True

app/chip_tool/test_suite.py

@@ -54,13 +54,13 @@ async def setup(self) -> None:
         logger.info("Setting up chip_tool")
         # Use test engine logger to log all events to test run.
         self.chip_tool.logger = logger
-        await self.chip_tool.start_container(
+        await self.chip_tool.start_server(
             self.test_type, self.config.dut_config.chip_tool_use_paa_certs
         )
 
         if len(self.pics.clusters) > 0:
             logger.info("Create PICS file for DUT")
-            self.chip_tool.set_pics(pics=self.pics)
+            self.chip_tool.set_pics(pics=self.pics, in_container=False)
         else:
             # Disable sending "-PICS" option when running chip-tool
             self.chip_tool.reset_pics_state()

app/tests/chip_tool/test_chip_tool.py

@@ -69,7 +69,7 @@ async def test_start_container() -> None:
     ) as mock_create_container, mock.patch.object(
         target=chip_tool, attribute="start_chip_server"
     ) as mock_start_chip_server:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
     mock_create_container.assert_called_once_with(docker_image, ChipTool.run_parameters)
     mock_start_chip_server.assert_awaited_once_with(test_type, False)
@@ -98,7 +98,7 @@ async def test_start_container_using_paa_certs() -> None:
     ) as mock_create_container, mock.patch.object(
         target=chip_tool, attribute="start_chip_server"
     ) as mock_start_chip_server:
-        await chip_tool.start_container(test_type, use_paa_certs=True)
+        await chip_tool.start_server(test_type, use_paa_certs=True)
 
     mock_create_container.assert_called_once_with(docker_image, ChipTool.run_parameters)
     mock_start_chip_server.assert_awaited_once_with(test_type, True)
@@ -111,7 +111,6 @@ async def test_start_container_using_paa_certs() -> None:
 @pytest.mark.asyncio
 async def test_not_start_container_when_running() -> None:
     chip_tool = ChipTool()
-    test_type = ChipToolTestType.CHIP_TOOL
 
     with mock.patch.object(
         target=chip_tool, attribute="is_running", return_value=True
@@ -120,7 +119,7 @@ async def test_not_start_container_when_running() -> None:
     ) as mock_create_container, mock.patch.object(
         target=chip_tool, attribute="start_chip_server"
     ) as mock_start_chip_server:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_container()
 
     mock_create_container.assert_not_called()
     mock_start_chip_server.assert_not_called()
@@ -359,8 +358,8 @@ async def test_destroy_container_running() -> None:
     ), mock.patch.object(
         target=chip_tool, attribute="start_chip_server"
     ):
-        await chip_tool.start_container(test_type)
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
+        await chip_tool.start_server(test_type)
 
         assert chip_tool._ChipTool__chip_tool_container is not None
 
@@ -403,7 +402,7 @@ async def test_destroy_container_once() -> None:
     ), mock.patch.object(
         target=chip_tool, attribute="start_chip_server"
     ):
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         await chip_tool.destroy_device()
         await chip_tool.destroy_device()
@@ -432,7 +431,7 @@ async def test_set_pics() -> None:
         "PICS_USER_PROMPT=1"
     )
     expected_command = (
-        f"{SHELL_PATH} {SHELL_OPTION} \"echo '{expected_pics_data}\n' "
+        f"{SHELL_PATH} {SHELL_OPTION} \"echo '{expected_pics_data}' "
         f'> {PICS_FILE_PATH}"'
     )
 
@@ -452,9 +451,9 @@ async def test_set_pics() -> None:
         target="app.chip_tool.chip_tool.subprocess.run",
         return_value=CompletedProcess(expected_command, 0),
     ) as mock_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
-        chip_tool.set_pics(pics)
+        chip_tool.set_pics(pics, in_container=False)
 
     mock_run.assert_called_once_with(expected_command, shell=True)
     assert chip_tool._ChipTool__pics_file_created is True
@@ -473,7 +472,7 @@ def test_set_pics_with_error() -> None:
         target="app.chip_tool.chip_tool.subprocess.run",
         return_value=CompletedProcess("", 1),
     ), pytest.raises(PICSError):
-        chip_tool.set_pics(pics)
+        chip_tool.set_pics(pics, in_container=False)
         assert chip_tool._ChipTool__pics_file_created is False
 
     # clean up:
@@ -505,7 +504,7 @@ async def test_send_command_default_prefix() -> None:
         target="app.chip_tool.chip_tool.exec_run_in_container",
         return_value=mock_result,
     ) as mock_exec_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         result = chip_tool.send_command(cmd, prefix=chip_tool_prefix)
 
@@ -548,7 +547,7 @@ async def test_send_command_custom_prefix() -> None:
         target="app.chip_tool.chip_tool.exec_run_in_container",
         return_value=mock_result,
     ) as mock_exec_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         result = chip_tool.send_command(cmd, prefix=chip_tool_prefix)
 
@@ -592,7 +591,7 @@ async def test_run_test_default_config() -> None:
         target="app.chip_tool.chip_tool.WebSocketRunner.run",
         return_value=True,
     ) as mock_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         await chip_tool.run_test(
             test_step_interface=TestRunnerHooks(),
@@ -649,7 +648,7 @@ async def test_run_test_custom_timeout() -> None:
         target="app.chip_tool.chip_tool.WebSocketRunner.run",
         return_value=True,
     ) as mock_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         await chip_tool.run_test(
             test_step_interface=TestRunnerHooks(),
@@ -700,7 +699,7 @@ async def test_run_test_with_custom_parameter() -> None:
         target="app.chip_tool.chip_tool.WebSocketRunner.run",
         return_value=True,
     ) as mock_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         await chip_tool.run_test(
             test_step_interface=TestRunnerHooks(),
@@ -752,7 +751,7 @@ async def test_run_test_with_endpoint_parameter() -> None:
         target="app.chip_tool.chip_tool.WebSocketRunner.run",
         return_value=True,
     ) as mock_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         await chip_tool.run_test(
             test_step_interface=TestRunnerHooks(),
@@ -803,7 +802,7 @@ async def test_run_test_with_nodeID_and_cluster_parameters() -> None:
         target="app.chip_tool.chip_tool.WebSocketRunner.run",
         return_value=True,
     ) as mock_run:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         await chip_tool.run_test(
             test_step_interface=TestRunnerHooks(),
@@ -862,7 +861,7 @@ async def test_pairing_on_network_command_params() -> None:
         attribute="send_websocket_command",
         return_value='{"results": []}',
     ) as mock_send_websocket_command:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         # Send on-network pairing command
         result = await chip_tool.pairing_on_network(
@@ -913,7 +912,7 @@ async def test_pairing_ble_wifi_command_params() -> None:
         attribute="send_websocket_command",
         return_value='{"results": []}',
     ) as mock_send_websocket_command:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         # Send BLE-WIFI pairing command
         result = await chip_tool.pairing_ble_wifi(
@@ -968,7 +967,7 @@ async def test_pairing_ble_thread_command_params() -> None:
         return_value='{"results": []}',
         # '{  "results": [{ "error": "FAILURE" }]
     ) as mock_send_websocket_command:
-        await chip_tool.start_container(test_type)
+        await chip_tool.start_server(test_type)
 
         # Send BLE-THREAD pairing command
         result = await chip_tool.pairing_ble_thread(