Add python tests model classes

app/chip_tool/chip_tool.py

@@ -107,6 +107,17 @@
 XML_SPEC_DEFINITION_PATH = TEST_COLLECTION_SDK_CHECKOUT_PATH / Path(
     "sdk_runner/specifications/chip/"
 )
+# Python Testing Folder
+# LOCAL_PYTHON_TESTING_PATH = TEST_COLLECTION_SDK_CHECKOUT_PATH / Path(
+#     "python_testing/scripts/sdk"
+# )
+# DOCKER_PYTHON_TESTING_PATH = "/root/python_testing"
+
+# RPC Client Running on SDK Container
+LOCAL_RPC_PYTHON_TESTING_PATH = BACKEND_ROOT / Path(
+    "test_collections/sdk_tests/support/python_testing/models/rpc_client/"
+)
+DOCKER_RPC_PYTHON_TESTING_PATH = "/root/python_testing/rpc_client"
 
 
 # Docker Network
@@ -132,6 +143,7 @@ class ChipToolUnknownTestType(Exception):
 class ChipToolTestType(str, Enum):
     CHIP_TOOL = "chip-tool"
     CHIP_APP = "chip-app"
+    PYTHON_TEST = "python-test"
 
 
 class ChipTool(metaclass=Singleton):
@@ -168,6 +180,10 @@ class ChipTool(metaclass=Singleton):
                 "bind": DOCKER_CREDENTIALS_DEVELOPMENT_PATH,
                 "mode": "ro",
             },
+            LOCAL_RPC_PYTHON_TESTING_PATH: {
+                "bind": DOCKER_RPC_PYTHON_TESTING_PATH,
+                "mode": "rw",
+            },
         },
     }
 
@@ -347,46 +363,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(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"
-            )
-            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_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.
+        """
+        if self.is_running():
+            self.logger.info(
+                "chip-tool container already running, no need to start a new container"
+            )
+            return
+
+        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)
@@ -590,7 +607,7 @@ async def run_test(
 
         if test_type == ChipToolTestType.CHIP_TOOL:
             test_path = f"{YAML_TESTS_PATH}/{test_id}.yaml"
-        else:
+        elif test_type == ChipToolTestType.CHIP_APP:
             test_path = f"{YAML_TESTS_PATH}/{test_id}_Simulated.yaml"
 
         parser_config = TestParserConfig(pics_path, self.specifications, test_options)

app/chip_tool/test_suite.py

@@ -54,7 +54,7 @@
         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
         )
 
@@ -280,3 +280,4 @@
                     f"Received unknown prompt option for \
                         decommissioning step: {prompt_response.response}"
                 )
+

app/tests/chip_tool/test_chip_tool.py

@@ -69,7 +69,7 @@
     ) 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 @@
     ) 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)
@@ -120,7 +120,7 @@
     ) 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_not_called()
     mock_start_chip_server.assert_not_called()
@@ -359,8 +359,8 @@
     ), 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 +403,7 @@
     ), 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()
@@ -452,7 +452,7 @@
         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)
 
@@ -505,7 +505,7 @@
         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 +548,7 @@
         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 +592,7 @@
         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 +649,7 @@
         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 +700,7 @@
         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 +752,7 @@
         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 +803,7 @@
         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 +862,7 @@
         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 +913,7 @@
         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 +968,7 @@
         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(
@@ -988,3 +988,4 @@
     # clean up:
     chip_tool._ChipTool__chip_tool_container = None
     settings.CHIP_TOOL_TRACE = original_trace_setting_value
+

app/tests/python_tests/test_python_folder.py

@@ -0,0 +1,58 @@
+#
+# 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.
+#
+# flake8: noqa
+from pathlib import Path
+from unittest import mock
+
+from test_collections.sdk_tests.support.models.sdk_test_folder import SDKTestFolder
+
+test_python_path = Path("/test/python")
+
+
+def test_python_folder_version() -> None:
+    version_file_content = "python_test_version"
+
+    # We mock open to read version_file_content and Path exists to ignore that we're
+    # testing with a fake path
+    with mock.patch(
+        "test_collections.sdk_tests.support.models.sdk_test_folder.open",
+        new=mock.mock_open(read_data=version_file_content),
+    ), mock.patch.object(target=Path, attribute="exists", return_value=True) as _:
+        python_test_folder = SDKTestFolder(test_python_path)
+
+        assert python_test_folder.version == version_file_content
+
+
+def test_python_folder_version_missing() -> None:
+    expected_version = "Unknown"
+    with mock.patch.object(target=Path, attribute="exists", return_value=False) as _:
+        python_folder = SDKTestFolder(test_python_path)
+        assert python_folder.version == expected_version
+
+
+def test_python_folder_filename_pattern() -> None:
+    """Test SDKTestFolder will search for files with filename pattern."""
+    with mock.patch.object(target=Path, attribute="glob") as path_glob:
+        # Default file_name_pattern: *
+        python_folder = SDKTestFolder(test_python_path)
+        _ = python_folder.file_paths(extension=".py")
+        path_glob.assert_called_once_with("*.py")
+
+        path_glob.reset_mock()
+        pattern = "TC_*"
+        python_test_folder = SDKTestFolder(test_python_path, filename_pattern=pattern)
+        _ = python_test_folder.file_paths(extension=".py")
+        path_glob.assert_called_once_with(f"{pattern}.py")