03-Implemented changes for custom python test

.gitignore

@@ -12,4 +12,5 @@ test_environment.config
 SerialTests.lock
 test_db_creation.lock
 .sha_information
-test_collections/matter/sdk_tests/sdk_checkout
+test_collections/matter/sdk_tests/sdk_checkout
+test_collections/matter/sdk_tests/custom_python_tests_info.json

app/tests/conftest.py

@@ -14,8 +14,13 @@
 # limitations under the License.
 #
 import asyncio
+import contextlib
+import sys
+from importlib import import_module
+from pathlib import Path
 from typing import AsyncGenerator, Generator
 from unittest import mock
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 import pytest_asyncio
@@ -27,12 +32,18 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
 
+# isort: split - DO NOT MOVE THIS IMPORT, this should be before any app.*
+from .sdk_container_mock import mock_instance  # Mock is already setup when this imports
+
+# isort: split
+
 from app.core.config import settings
 from app.db.base_class import Base
 from app.db.init_db import create_app_database
 from app.main import app as main_app
 from app.test_engine import test_script_manager
 from app.test_engine.test_collection_discovery import discover_test_collections
+from test_collections.matter.sdk_tests.support.sdk_container import SDKContainer
 
 if settings.SQLALCHEMY_DATABASE_URI is None:
     raise ValueError("Database URI is missing")
@@ -123,3 +134,32 @@
 test_script_manager.test_script_manager.test_collections = discover_test_collections(
     disabled_collections=[]
 )
+
+
+@contextlib.contextmanager
+def use_real_sdk_container():
+    """Context manager to temporarily use the real SDKContainer"""
+    # Store the mock module
+    mock_module = sys.modules["test_collections.matter.sdk_tests.support.sdk_container"]
+
+    # Remove the mock from sys.modules to force reload
+    del sys.modules["test_collections.matter.sdk_tests.support.sdk_container"]
+
+    try:
+        # Import the real module
+        real_module = import_module(
+            "test_collections.matter.sdk_tests.support.sdk_container"
+        )
+        yield real_module
+    finally:
+        # Restore the mock module
+        sys.modules[
+            "test_collections.matter.sdk_tests.support.sdk_container"
+        ] = mock_module
+
+
+@pytest.fixture
+def real_sdk_container():
+    """Use the real SDKContainer in a test"""
+    with use_real_sdk_container() as real_module:
+        yield real_module.SDKContainer()

app/tests/sdk_container_mock.py

@@ -0,0 +1,99 @@
+#
+# Copyright (c) 2024 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.
+#
+
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, Mock
+
+# Constants from the original module
+DOCKER_LOGS_PATH = "/logs"
+DOCKER_PAA_CERTS_PATH = "/paa-root-certs"
+LOCAL_LOGS_PATH = Path("/var/tmp")
+LOCAL_PAA_CERTS_PATH = Path("/var/paa-root-certs")
+LOCAL_CREDENTIALS_DEVELOPMENT_PATH = Path("/var/credentials/development")
+DOCKER_CREDENTIALS_DEVELOPMENT_PATH = "/credentials/development"
+LOCAL_TEST_COLLECTIONS_PATH = (
+    "/home/ubuntu/certification-tool/backend/test_collections/matter"
+)
+LOCAL_PYTHON_TESTING_PATH = Path(
+    LOCAL_TEST_COLLECTIONS_PATH + "/sdk_tests/sdk_checkout/python_testing"
+)
+DOCKER_PYTHON_TESTING_PATH = "/root/python_testing"
+MAPPED_DATA_MODEL_VOLUME = "mapped_data_model_volume"
+DOCKER_DATA_MODEL_PATH = DOCKER_PYTHON_TESTING_PATH + "/data_model"
+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/scripts/sdk/matter_testing_infrastructure/chip/testing/test_harness_client.py"
+
+
+# Exception classes
+class SDKContainerNotRunning(Exception):
+    """Raised when we attempt to use the docker container, but it is not running"""
+
+
+class SDKContainerRetrieveExitCodeError(Exception):
+    """Raised when there's an error in the attempt to retrieve an execution's exit code"""
+
+
+# Create mock instance
+mock_instance = Mock()
+mock_instance.is_running.return_value = True
+mock_instance.pics_file_created = False
+mock_instance.send_command.return_value = Mock(exit_code=0, output="mocked output")
+mock_instance.start = AsyncMock()
+
+
+# Create mock SDKContainer class
+class MockSDKContainer:
+    def __new__(cls, *args, **kwargs):
+        return mock_instance
+
+
+# Store mock for access
+sys.mock_sdk_container = mock_instance
+
+# Create and setup mock module
+mock_module = type(
+    "mock_module",
+    (),
+    {
+        "SDKContainer": MockSDKContainer,
+        "DOCKER_LOGS_PATH": DOCKER_LOGS_PATH,
+        "DOCKER_PAA_CERTS_PATH": DOCKER_PAA_CERTS_PATH,
+        "LOCAL_LOGS_PATH": LOCAL_LOGS_PATH,
+        "LOCAL_PAA_CERTS_PATH": LOCAL_PAA_CERTS_PATH,
+        "LOCAL_CREDENTIALS_DEVELOPMENT_PATH": LOCAL_CREDENTIALS_DEVELOPMENT_PATH,
+        "DOCKER_CREDENTIALS_DEVELOPMENT_PATH": DOCKER_CREDENTIALS_DEVELOPMENT_PATH,
+        "LOCAL_TEST_COLLECTIONS_PATH": LOCAL_TEST_COLLECTIONS_PATH,
+        "LOCAL_PYTHON_TESTING_PATH": LOCAL_PYTHON_TESTING_PATH,
+        "DOCKER_PYTHON_TESTING_PATH": DOCKER_PYTHON_TESTING_PATH,
+        "MAPPED_DATA_MODEL_VOLUME": MAPPED_DATA_MODEL_VOLUME,
+        "DOCKER_DATA_MODEL_PATH": DOCKER_DATA_MODEL_PATH,
+        "LOCAL_RPC_PYTHON_TESTING_PATH": LOCAL_RPC_PYTHON_TESTING_PATH,
+        "DOCKER_RPC_PYTHON_TESTING_PATH": DOCKER_RPC_PYTHON_TESTING_PATH,
+        "SDKContainerNotRunning": SDKContainerNotRunning,
+        "SDKContainerRetrieveExitCodeError": SDKContainerRetrieveExitCodeError,
+        "__file__": "mocked_path",
+        "__loader__": None,
+        "__spec__": None,
+        "__name__": "test_collections.matter.sdk_tests.support.sdk_container",
+    },
+)()
+
+# Patch the module
+sys.modules["test_collections.matter.sdk_tests.support.sdk_container"] = mock_module

test_collections/matter/sdk_tests/support/python_testing/list_python_tests_classes.py

@@ -36,8 +36,15 @@
 PYTHON_SCRIPTS_PATH = PYTHON_TESTING_PATH / "scripts/sdk"
 PYTHON_SCRIPTS_FOLDER = SDKTestFolder(path=PYTHON_SCRIPTS_PATH, filename_pattern="TC*")
 
+CUSTOM_PYTHON_SCRIPTS_PATH = PYTHON_TESTING_PATH / "scripts/custom"
+CUSTOM_PYTHON_SCRIPTS_FOLDER = SDKTestFolder(
+    path=CUSTOM_PYTHON_SCRIPTS_PATH, filename_pattern="TC*"
+)
+
 CONTAINER_TH_CLIENT_EXEC = "python3 /root/python_testing/scripts/sdk/matter_testing_infrastructure/chip/testing/test_harness_client.py"  # noqa
 
+sdk_container: SDKContainer = SDKContainer()
+
 
 def base_test_classes(module: ast.Module) -> list[ast.ClassDef]:
     """Find classes that inherit from MatterBaseTest.
@@ -59,9 +66,9 @@
     ]
 
 
-def get_command_list() -> list:
+def get_command_list(test_folder: SDKTestFolder) -> list:
     python_script_commands = []
-    python_test_files = PYTHON_SCRIPTS_FOLDER.file_paths(extension=".py")
+    python_test_files = test_folder.file_paths(extension=".py")
     python_test_files.sort()
 
     for python_test_file in python_test_files:
@@ -79,7 +86,10 @@
     return python_script_commands
 
 
-async def proccess_commands_sdk_container(commands: list) -> None:
+async def proccess_commands_sdk_container(
+    commands: list,
+    json_output_file: Path,
+) -> None:
     complete_json = []
     errors_found: list[str] = []
     test_function_count = 0
@@ -114,20 +124,18 @@
 
     sdk_container.destroy()
 
-    complete_json_path = COMPLETE_JSON_OUTPUT_FILE_FOLDER / "python_tests_info.json"
-
     # complete_json.append({"sdk_sha": matter_settings.SDK_SHA})
     # Create a wrapper object with sdk_sha at root level
     json_output = {"sdk_sha": matter_settings.SDK_SHA, "tests": complete_json}
 
-    with open(complete_json_path, "w") as json_file:
+    with open(json_output_file, "w") as json_file:
         json.dump(json_output, json_file, indent=4, sort_keys=True)
         json_file.close()
 
     print("###########################################################################")
     print("###############################   REPORT   ################################")
     print("###########################################################################")
-    print(f">>>>>>>> Output JSON file: {complete_json_path}")
+    print(f">>>>>>>> Output JSON file: {json_output_file}")
     print(f">>>>>>>> Total of test functions: {test_function_count}")
     print(
         (
@@ -143,10 +151,16 @@
     print("###########################################################################")
 
 
-if __name__ == "__main__":
-    python_scripts_command_list = get_command_list()
-    loop = asyncio.get_event_loop()
-    loop.run_until_complete(
-        proccess_commands_sdk_container(python_scripts_command_list)
+async def generate_python_test_json_file(
+    test_folder: SDKTestFolder = PYTHON_SCRIPTS_FOLDER,
+    json_output_file=COMPLETE_JSON_OUTPUT_FILE_FOLDER / "python_tests_info.json",
+) -> None:
+    python_scripts_command_list = get_command_list(test_folder=test_folder)
+
+    await proccess_commands_sdk_container(
+        python_scripts_command_list, json_output_file=json_output_file
     )
-    loop.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(generate_python_test_json_file())

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

@@ -60,10 +60,13 @@ def parse_python_script(path: Path) -> list[PythonTest]:
     Example: file TC_ACE_1_3.py has the methods test_TC_ACE_1_3, desc_TC_ACE_1_3,
         pics_TC_ACE_1_3 and steps_TC_ACE_1_3.
     """
+    python_tests: list[PythonTest] = []
+
     with open(path, "r") as json_file:
         parsed_scripts = json.load(json_file)
 
-    python_tests: list[PythonTest] = []
+    if len(parsed_scripts) == 0:
+        return python_tests
 
     for script_info in parsed_scripts["tests"]:
         test_function = script_info["function"]

test_collections/matter/sdk_tests/support/python_testing/sdk_python_tests.py

@@ -14,11 +14,17 @@
 # limitations under the License.
 #
 
+import asyncio
 from pathlib import Path
 from typing import Optional
 
 from ..models.sdk_test_folder import SDKTestFolder
 from ..paths import SDK_CHECKOUT_PATH, SDK_TESTS_ROOT
+from .list_python_tests_classes import (
+    COMPLETE_JSON_OUTPUT_FILE_FOLDER,
+    CUSTOM_PYTHON_SCRIPTS_FOLDER,
+    generate_python_test_json_file,
+)
 from .models.python_test_models import PythonTest, PythonTestType
 from .models.python_test_parser import parse_python_script
 from .models.test_declarations import (
@@ -50,6 +56,7 @@
 )
 
 PYTHON_TESTS_PARSED_FILE = SDK_TESTS_ROOT / "python_tests_info.json"
+CUSTOM_PYTHON_TESTS_PARSED_FILE = SDK_TESTS_ROOT / "custom_python_tests_info.json"
 
 
 def _init_test_suites(
@@ -170,26 +177,32 @@ def sdk_mandatory_python_test_collection(
 
 def custom_python_test_collection(
     python_test_folder: SDKTestFolder = CUSTOM_PYTHON_TEST_FOLDER,
+    tests_file_path: Path = CUSTOM_PYTHON_TESTS_PARSED_FILE,
 ) -> Optional[PythonCollectionDeclaration]:
+    asyncio.run(
+        generate_python_test_json_file(
+            test_folder=CUSTOM_PYTHON_SCRIPTS_FOLDER,
+            json_output_file=COMPLETE_JSON_OUTPUT_FILE_FOLDER
+            / "custom_python_tests_info.json",
+        )
+    )
+
     """Declare a new collection of test suites."""
-    return None
-    # TODO: Implement custom python test collection
-    # collection = PythonCollectionDeclaration(
-    #     name="Custom SDK Python Tests", folder=python_test_folder
-    # )
-
-    # suites = __parse_python_tests(
-    #     python_test_version="custom",
-    #     mandatory=False,
-    # )
-
-    # for suite in suites:
-    #     if not suite.test_cases:
-    #         continue
-    #     suite.sort_test_cases()
-    #     collection.add_test_suite(suite)
-
-    # if not collection.test_suites:
-    #     return None
-
-    # return collection
+    collection = PythonCollectionDeclaration(
+        name="Custom SDK Python Tests", folder=python_test_folder
+    )
+
+    suites = __parse_python_tests(
+        python_test_version="custom", mandatory=False, tests_file_path=tests_file_path
+    )
+
+    for suite in suites:
+        if not suite.test_cases:
+            continue
+        suite.sort_test_cases()
+        collection.add_test_suite(suite)
+
+    if not collection.test_suites:
+        return None
+
+    return collection