Move MockTestRunner to chip testing package (#37170)

* Move MockTestRunner to chip testing package

* Fix script dir path

* Remove unused import

src/python_testing/matter_testing_infrastructure/BUILD.gn

@@ -43,6 +43,7 @@ pw_python_package("chip-testing-module") {
     "chip/testing/matter_testing.py",
     "chip/testing/metadata.py",
     "chip/testing/pics.py",
+    "chip/testing/runner.py",
     "chip/testing/spec_parsing.py",
     "chip/testing/taglist_and_topology_test.py",
     "chip/testing/tasks.py",

src/python_testing/test_testing/MockTestRunner.py → src/python_testing/matter_testing_infrastructure/chip/testing/runner.py

@@ -17,7 +17,6 @@
 
 import asyncio
 import importlib
-import os
 import sys
 from pathlib import Path
 from unittest.mock import MagicMock
@@ -33,11 +32,12 @@ async def __call__(self, *args, **kwargs):
 
 class MockTestRunner():
 
-    def __init__(self, filename: str, classname: str, test: str, endpoint: int = None, pics: dict[str, bool] = None, paa_trust_store_path=None):
+    def __init__(self, abs_filename: str, classname: str, test: str, endpoint: int = None,
+                 pics: dict[str, bool] = None, paa_trust_store_path=None):
         self.kvs_storage = 'kvs_admin.json'
         self.config = MatterTestConfig(endpoint=endpoint, paa_trust_store_path=paa_trust_store_path,
                                        pics=pics, storage_path=self.kvs_storage)
-        self.set_test(filename, classname, test)
+        self.set_test(abs_filename, classname, test)
 
         self.set_test_config(self.config)
 
@@ -48,17 +48,16 @@ def __init__(self, filename: str, classname: str, test: str, endpoint: int = Non
             catTags=self.config.controller_cat_tags
         )
 
-    def set_test(self, filename: str, classname: str, test: str):
+    def set_test(self, abs_filename: str, classname: str, test: str):
         self.test = test
         self.config.tests = [self.test]
 
-        module_name = Path(os.path.basename(filename)).stem
-
         try:
-            module = importlib.import_module(module_name)
+            filename_path = Path(abs_filename)
+            module = importlib.import_module(filename_path.stem)
         except ModuleNotFoundError:
-            sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
-            module = importlib.import_module(module_name)
+            sys.path.append(str(filename_path.parent.resolve()))
+            module = importlib.import_module(filename_path.stem)
 
         self.test_class = getattr(module, classname)
 
@@ -72,7 +71,7 @@ def set_test_config(self, test_config: MatterTestConfig = MatterTestConfig()):
     def Shutdown(self):
         self.stack.Shutdown()
 
-    def run_test_with_mock_read(self,  read_cache: Attribute.AsyncReadTransaction.ReadResponse, hooks=None):
+    def run_test_with_mock_read(self, read_cache: Attribute.AsyncReadTransaction.ReadResponse, hooks=None):
         self.default_controller.Read = AsyncMock(return_value=read_cache)
         # This doesn't need to do anything since we are overriding the read anyway
         self.default_controller.FindOrEstablishPASESession = AsyncMock(return_value=None)

src/python_testing/test_testing/TestDecorators.py

@@ -26,14 +26,15 @@
 # You will get step_* calls as appropriate in between the test_start and test_stop calls if the test is not skipped.
 
 import sys
+from pathlib import Path
 from typing import Optional
 
 import chip.clusters as Clusters
 from chip.clusters import Attribute
 from chip.testing.matter_testing import (MatterBaseTest, MatterTestConfig, async_test_body, has_attribute, has_cluster, has_feature,
                                          run_if_endpoint_matches, run_on_singleton_matching_endpoint, should_run_test_on_endpoint)
+from chip.testing.runner import MockTestRunner
 from mobly import asserts
-from MockTestRunner import MockTestRunner
 
 
 def get_clusters(endpoints: list[int]) -> Attribute.AsyncReadTransaction.ReadResponse:
@@ -216,7 +217,8 @@ async def test_no_run_on_singleton_matching_endpoint(self):
 def main():
     failures = []
     hooks = DecoratorTestRunnerHooks()
-    test_runner = MockTestRunner('TestDecorators.py', 'TestDecorators', 'test_checkers')
+    test_runner = MockTestRunner(Path(__file__).parent / 'TestDecorators.py',
+                                 'TestDecorators', 'test_checkers')
     read_resp = get_clusters([0, 1])
     ok = test_runner.run_test_with_mock_read(read_resp, hooks)
     if not ok:

src/python_testing/test_testing/common_icdm_data.py

@@ -18,10 +18,11 @@
 
 import string
 from dataclasses import dataclass
+from pathlib import Path
 
 import chip.clusters as Clusters
 from chip.clusters import Attribute
-from MockTestRunner import MockTestRunner
+from chip.testing.runner import MockTestRunner
 
 c = Clusters.IcdManagement
 attr = c.Attributes
@@ -54,8 +55,8 @@ def test_spec_to_attribute_cache(test_icdm: ICDMData) -> Attribute.AsyncReadTran
 
 
 def run_tests(pics, label, test_cases, test_name):
-    test_runner = MockTestRunner(
-        label, label, test_name, 0, pics)
+    test_runner = MockTestRunner(Path(__file__).parent / f"../{label}.py",
+                                 label, test_name, 0, pics)
     failures = []
     for idx, t in enumerate(test_cases):
         ok = test_runner.run_test_with_mock_read(

src/python_testing/test_testing/test_IDM_10_4.py

@@ -16,13 +16,13 @@
 #    limitations under the License.
 #
 
-import os
 import sys
+from pathlib import Path
 
 import chip.clusters as Clusters
 from chip.clusters import Attribute
 from chip.testing.pics import parse_pics_xml
-from MockTestRunner import MockTestRunner
+from chip.testing.runner import MockTestRunner
 
 # Reachable attribute is off in the pics file
 # MaxPathsPerInvoke is not include in the pics file
@@ -80,10 +80,12 @@ def create_read(include_reachable: bool = False, include_max_paths: bool = False
 
 def main():
     # TODO: add the same test for commands and features
-    script_dir = os.path.dirname(os.path.realpath(__file__))
-    with open(f'{script_dir}/example_pics_xml_basic_info.xml') as f:
+
+    script_dir = Path(__file__).resolve().parent
+    with open(script_dir / 'example_pics_xml_basic_info.xml') as f:
         pics = parse_pics_xml(f.read())
-    test_runner = MockTestRunner('TC_pics_checker.py', 'TC_PICS_Checker', 'test_TC_IDM_10_4', 0, pics)
+    test_runner = MockTestRunner(script_dir / '../TC_pics_checker.py',
+                                 'TC_PICS_Checker', 'test_TC_IDM_10_4', 0, pics)
     failures = []
 
     # Success, include vendor ID, which IS in the pics file, and neither of the incorrect ones

src/python_testing/test_testing/test_TC_CCNTL_2_2.py

@@ -19,16 +19,16 @@
 import asyncio
 import base64
 import os
-import pathlib
 import sys
 import typing
+from pathlib import Path
 
 import chip.clusters as Clusters
 import click
 from chip import ChipDeviceCtrl
 from chip.clusters import Attribute
 from chip.interaction_model import InteractionModelError, Status
-from MockTestRunner import AsyncMock, MockTestRunner
+from chip.testing.runner import AsyncMock, MockTestRunner
 
 try:
     from chip.testing.matter_testing import MatterTestConfig, get_default_paa_trust_store, run_tests_no_exit
@@ -175,13 +175,14 @@ def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, dynamic_eve
 @click.command()
 @click.argument('th_server_app', type=click.Path(exists=True))
 def main(th_server_app: str):
-    root = os.path.abspath(os.path.join(pathlib.Path(__file__).resolve().parent, '..', '..', '..'))
+    root = os.path.abspath(os.path.join(Path(__file__).resolve().parent, '..', '..', '..'))
     print(f'root = {root}')
     paa_path = get_default_paa_trust_store(root)
     print(f'paa = {paa_path}')
 
     pics = {"PICS_SDK_CI_ONLY": True}
-    test_runner = MyMock('TC_CCTRL_2_2', 'TC_CCTRL_2_2', 'test_TC_CCTRL_2_2', paa_trust_store_path=paa_path, pics=pics)
+    test_runner = MyMock(Path(__file__).parent / '../TC_CCTRL_2_2.py',
+                         'TC_CCTRL_2_2', 'test_TC_CCTRL_2_2', paa_trust_store_path=paa_path, pics=pics)
     config = MatterTestConfig()
     config.global_test_params = {'th_server_app_path': th_server_app}
     test_runner.set_test_config(config)

src/python_testing/test_testing/test_TC_DGGEN_3_2.py

@@ -18,10 +18,11 @@
 
 import sys
 from dataclasses import dataclass
+from pathlib import Path
 
 import chip.clusters as Clusters
 from chip.clusters import Attribute
-from MockTestRunner import MockTestRunner
+from chip.testing.runner import MockTestRunner
 
 
 @dataclass
@@ -51,7 +52,8 @@ def test_spec_to_attribute_cache(test_spec: TestSpec) -> Attribute.AsyncReadTran
 
 
 def main():
-    test_runner = MockTestRunner('TC_DGGEN_3_2', 'TC_DGGEN_3_2', 'test_TC_DGGEN_3_2', 0)
+    test_runner = MockTestRunner(Path(__file__).parent / '../TC_DGGEN_3_2.py',
+                                 'TC_DGGEN_3_2', 'test_TC_DGGEN_3_2', 0)
     failures = []
     for idx, t in enumerate(TEST_CASES):
         ok = test_runner.run_test_with_mock_read(test_spec_to_attribute_cache(t)) == t.expect_pass

src/python_testing/test_testing/test_TC_MCORE_FS_1_1.py

@@ -19,16 +19,16 @@
 import asyncio
 import base64
 import os
-import pathlib
 import sys
 import typing
+from pathlib import Path
 
 import chip.clusters as Clusters
 import click
 from chip import ChipDeviceCtrl
 from chip.clusters import Attribute
 from chip.interaction_model import InteractionModelError, Status
-from MockTestRunner import AsyncMock, MockTestRunner
+from chip.testing.runner import AsyncMock, MockTestRunner
 
 try:
     from chip.testing.matter_testing import MatterTestConfig, get_default_paa_trust_store, run_tests_no_exit
@@ -146,13 +146,14 @@ def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, dynamic_eve
 @click.command()
 @click.argument('th_server_app', type=click.Path(exists=True))
 def main(th_server_app: str):
-    root = os.path.abspath(os.path.join(pathlib.Path(__file__).resolve().parent, '..', '..', '..'))
+    root = os.path.abspath(os.path.join(Path(__file__).resolve().parent, '..', '..', '..'))
     print(f'root = {root}')
     paa_path = get_default_paa_trust_store(root)
     print(f'paa = {paa_path}')
 
     pics = {"PICS_SDK_CI_ONLY": True}
-    test_runner = MyMock('TC_MCORE_FS_1_1', 'TC_MCORE_FS_1_1', 'test_TC_MCORE_FS_1_1', paa_trust_store_path=paa_path, pics=pics)
+    test_runner = MyMock(Path(__file__).parent / '../TC_MCORE_FS_1_1.py',
+                         'TC_MCORE_FS_1_1', 'test_TC_MCORE_FS_1_1', paa_trust_store_path=paa_path, pics=pics)
     config = MatterTestConfig()
     config.user_params = {'th_server_app_path': th_server_app}
     test_runner.set_test_config(config)

src/python_testing/test_testing/test_TC_SC_7_1.py

@@ -17,12 +17,13 @@
 #
 
 import sys
+from pathlib import Path
 from random import randbytes
 
 import chip.clusters as Clusters
 from chip.clusters import Attribute
 from chip.testing.matter_testing import MatterTestConfig
-from MockTestRunner import MockTestRunner
+from chip.testing.runner import MockTestRunner
 
 
 def read_trusted_root(filled: bool) -> Attribute.AsyncReadTransaction.ReadResponse:
@@ -43,7 +44,8 @@ def main():
     manual_3333_20202021 = '31693312339'
     manual_2222_20202024 = '20055212333'
 
-    test_runner = MockTestRunner('TC_SC_7_1', 'TC_SC_7_1', 'test_TC_SC_7_1', 0)
+    test_runner = MockTestRunner(Path(__file__).parent / '../TC_SC_7_1.py',
+                                 'TC_SC_7_1', 'test_TC_SC_7_1', 0)
     failures = []
 
     # Tests with no code specified should fail

src/python_testing/test_testing/test_TC_TMP_2_1.py

@@ -19,11 +19,12 @@
 import sys
 import typing
 from dataclasses import dataclass
+from pathlib import Path
 
 import chip.clusters as Clusters
 from chip.clusters import Attribute
 from chip.clusters.Types import NullValue
-from MockTestRunner import MockTestRunner
+from chip.testing.runner import MockTestRunner
 
 
 @dataclass
@@ -160,7 +161,8 @@ def test_spec_to_attribute_cache(test_spec: TestSpec) -> Attribute.AsyncReadTran
 
 
 def main():
-    test_runner = MockTestRunner('TC_TMP_2_1', 'TC_TMP_2_1', 'test_TC_TMP_2_1', 1)
+    test_runner = MockTestRunner(Path(__file__).parent / '../TC_TMP_2_1.py',
+                                 'TC_TMP_2_1', 'test_TC_TMP_2_1', 1)
     failures = []
     for idx, t in enumerate(TEST_CASES):
         ok = test_runner.run_test_with_mock_read(test_spec_to_attribute_cache(t)) == t.expect_pass