Merge pull request #1 from ccruzagralopes/add_support_for_PICS_in_pyt…

…hon_tests

Add support for step PICS in python tests

app/chip_tool/chip_tool.py

@@ -224,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.
@@ -683,32 +687,36 @@ 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 container or
+                                not. If false, the file is created on the host.
 
         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

@@ -60,7 +60,7 @@ async def setup(self) -> None:
 
         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

@@ -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_server(test_type)
+        await chip_tool.start_container()
 
     mock_create_container.assert_not_called()
     mock_start_chip_server.assert_not_called()
@@ -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}"'
     )
 
@@ -454,7 +453,7 @@ async def test_set_pics() -> None:
     ) as mock_run:
         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:

app/tests/python_tests/test_python_test_suite.py

@@ -20,9 +20,11 @@
 
 import pytest
 
-from app.chip_tool.chip_tool import ChipToolTestType
+from app.chip_tool.chip_tool import ChipTool
 from app.models.test_suite_execution import TestSuiteExecution
+from app.schemas import PICS
 from app.test_engine.logger import test_engine_logger
+from app.tests.utils.test_pics_data import create_random_pics
 from test_collections.sdk_tests.support.python_testing.models.test_suite import (
     PythonTestSuite,
     SuiteType,
@@ -60,6 +62,8 @@ def test_python_test_suite_python_version() -> None:
 @pytest.mark.asyncio
 async def test_suite_setup_log_python_version() -> None:
     """Test that test suite python version is logged to test engine logger in setup."""
+    chip_tool: ChipTool = ChipTool()
+
     for type in list(SuiteType):
         python_test_version = "best_version"
         # Create a subclass of PythonTestSuite
@@ -69,17 +73,80 @@ async def test_suite_setup_log_python_version() -> None:
 
         suite_instance = suite_class(TestSuiteExecution())
 
-        # We're patching ChipToolSuite.setup to avoid starting chip-tool
         with mock.patch.object(
             target=test_engine_logger, attribute="info"
-        ) as logger_info, mock.patch(
-            "app.chip_tool.test_suite.ChipToolSuite.setup"
-        ) as _:
+        ) as logger_info, mock.patch.object(
+            target=chip_tool, attribute="start_container"
+        ), mock.patch(
+            target="test_collections.sdk_tests.support.python_testing.models.test_suite"
+            ".PythonTestSuite.pics",
+            new_callable=PICS,
+        ):
             await suite_instance.setup()
             logger_info.assert_called()
             logger_info.assert_any_call(f"Python Test Version: {python_test_version}")
 
 
+@pytest.mark.asyncio
+async def test_suite_setup_without_pics() -> None:
+    chip_tool: ChipTool = ChipTool()
+
+    for type in list(SuiteType):
+        python_test_version = "best_version"
+        # Create a subclass of PythonTestSuite
+        suite_class: Type[PythonTestSuite] = PythonTestSuite.class_factory(
+            suite_type=type, name="SomeSuite", python_test_version=python_test_version
+        )
+
+        suite_instance = suite_class(TestSuiteExecution())
+
+        with mock.patch(
+            "app.chip_tool.test_suite.ChipToolSuite.setup"
+        ), mock.patch.object(target=chip_tool, attribute="start_container"), mock.patch(
+            target="test_collections.sdk_tests.support.python_testing.models.test_suite"
+            ".PythonTestSuite.pics",
+            new_callable=PICS,
+        ), mock.patch.object(
+            target=chip_tool, attribute="set_pics"
+        ) as mock_set_pics, mock.patch.object(
+            target=chip_tool, attribute="reset_pics_state"
+        ) as mock_reset_pics_state:
+            await suite_instance.setup()
+
+        mock_set_pics.assert_not_called()
+        mock_reset_pics_state.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_suite_setup_with_pics() -> None:
+    chip_tool: ChipTool = ChipTool()
+
+    for type in list(SuiteType):
+        python_test_version = "best_version"
+        # Create a subclass of PythonTestSuite
+        suite_class: Type[PythonTestSuite] = PythonTestSuite.class_factory(
+            suite_type=type, name="SomeSuite", python_test_version=python_test_version
+        )
+
+        suite_instance = suite_class(TestSuiteExecution())
+
+        with mock.patch(
+            "app.chip_tool.test_suite.ChipToolSuite.setup"
+        ), mock.patch.object(target=chip_tool, attribute="start_container"), mock.patch(
+            target="test_collections.sdk_tests.support.python_testing.models.test_suite"
+            ".PythonTestSuite.pics",
+            new_callable=create_random_pics,
+        ), mock.patch.object(
+            target=chip_tool, attribute="set_pics"
+        ) as mock_set_pics, mock.patch.object(
+            target=chip_tool, attribute="reset_pics_state"
+        ) as mock_reset_pics_state:
+            await suite_instance.setup()
+
+        mock_set_pics.assert_called_once()
+        mock_reset_pics_state.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_chip_tool_suite_setup() -> None:
     """Test that PythonTestSuite.setup is called when PythonChipToolsSuite.setup is called.

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

@@ -23,39 +23,18 @@
 
 import matter_testing_support
 
-try:
-    from matter_yamltests.hooks import TestRunnerHooks
-except:
 
-    class TestRunnerHooks:
-        pass
-
-
-MATTER_DEVELOPMENT_PAA_ROOT_CERTS = "/paa-root-certs"
-
-# Pre-computed param list for each Python Test as defined in Verification Steps.
-test_params = {
-    "TC_ACE_1_3": matter_testing_support.MatterTestConfig(
-        tests=["test_TC_ACE_1_3"],
-        commissioning_method="on-network",
-        discriminators=[3840],
-        setup_passcodes=[20202021],
-        dut_node_ids=[0x12344321],
-        paa_trust_store_path=MATTER_DEVELOPMENT_PAA_ROOT_CERTS,
-        storage_path="/root/admin_storage.json",
-    )
-}
+class TestRunnerHooks:
+    pass
 
 
 def main() -> None:
     sys.path.append("/root/python_testing")
 
-    if len(sys.argv) != 2:
-        raise Exception("Python test id should be provided as the only parameter.")
-
     test_name = sys.argv[1]
 
-    config = test_params.get(test_name)
+    config_options = sys.argv[2:]
+    config = matter_testing_support.parse_matter_test_args(config_options)
 
     if config is None:
         raise ValueError(f"Not a valid test id: {test_name}")

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

@@ -17,9 +17,9 @@
 from multiprocessing.managers import BaseManager
 from typing import Any, Type, TypeVar
 
-from app.chip_tool.chip_tool import ChipTool
+from app.chip_tool.chip_tool import PICS_FILE_PATH, ChipTool
 from app.models import TestCaseExecution
-from app.test_engine.logger import logger, test_engine_logger
+from app.test_engine.logger import test_engine_logger as logger
 from app.test_engine.models import TestCase, TestStep
 
 from .python_test_models import PythonTest
@@ -49,7 +49,7 @@ class PythonTestCase(TestCase):
 
     def __init__(self, test_case_execution: TestCaseExecution) -> None:
         super().__init__(test_case_execution=test_case_execution)
-        self.chip_tool: ChipTool
+        self.chip_tool: ChipTool = ChipTool(logger)
         self.__runned = 0
         self.test_stop_called = False
 
@@ -68,9 +68,7 @@ def test_stop(self, exception: Exception, duration: int) -> None:
         self.current_test_step.mark_as_completed()
 
     def step_skipped(self, name: str, expression: str) -> None:
-        self.current_test_step.mark_as_not_applicable(
-            f"Test step skipped: {name}. {expression} == False"
-        )
+        self.current_test_step.mark_as_not_applicable("Test step skipped")
         self.next_step()
 
     def step_start(self, name: str) -> None:
@@ -94,16 +92,6 @@ def pics(cls) -> set[str]:
         """Test Case level PICS. Read directly from parsed Python Test."""
         return cls.python_test.PICS
 
-    async def setup(self) -> None:
-        """Override Setup to log Python Test version."""
-        test_engine_logger.info(f"Python Test Version: {self.python_test_version}")
-        try:
-            self.chip_tool = ChipTool()
-            await self.chip_tool.start_container()
-            assert self.chip_tool.is_running()
-        except NotImplementedError:
-            pass
-
     @classmethod
     def class_factory(cls, test: PythonTest, python_test_version: str) -> Type[T]:
         """class factory method for PythonTestCase."""
@@ -123,17 +111,35 @@ def class_factory(cls, test: PythonTest, python_test_version: str) -> Type[T]:
             },
         )
 
+    async def setup(self) -> None:
+        logger.info("Test Setup")
+
+    async def cleanup(self) -> None:
+        logger.info("Test Cleanup")
+
     async def execute(self) -> None:
         try:
             logger.info("Running Python Test: " + self.metadata["title"])
+
             BaseManager.register("TestRunnerHooks", SDKPythonTestRunnerHooks)
             manager = BaseManager(address=("0.0.0.0", 50000), authkey=b"abc")
             manager.start()
             test_runner_hooks = manager.TestRunnerHooks()  # type: ignore
+
             runner_class = RUNNER_CLASS_PATH + RUNNER_CLASS
+            command = (
+                f"{runner_class} {self.metadata['title']}"
+                " --commissioning-method on-network --discriminator 3840 --passcode"
+                " 20202021 --storage-path /root/admin_storage.json"
+                " --paa-trust-store-path /paa-root-certs"
+            )
+
+            if self.chip_tool.pics_file_created:
+                command += f" --PICS {PICS_FILE_PATH}"
+
             # TODO Ignoring stream from docker execution
             self.chip_tool.send_command(
-                f"{runner_class} {self.metadata['title']}",
+                command,
                 prefix=EXECUTABLE,
                 is_stream=True,
                 is_socket=False,
@@ -162,9 +168,6 @@ def __call_function_from_name(self, obj, func_name, kwargs) -> None:  # type: ig
             raise TypeError(f"{func_name} is not callable")
         func(**kwargs)
 
-    async def cleanup(self) -> None:
-        logger.info("Test Cleanup")
-
     def create_test_steps(self) -> None:
         self.test_steps = [TestStep("Start Python test")]
         for step in self.python_test.steps: