Rename metadata_parser module to chip.testing (#35615)

* Rename metadata_parser module to chip.testing

* MCORE_FS_1_4: Use Subprocess from chip.testing

* MCORE_FS_1_3: Use Subprocess from chip.testing

* Cleanup unused arguments

* Restyled by isort

* Fix test case to work on CI

BUILD.gn

@@ -81,7 +81,7 @@ if (current_toolchain != "${dir_pw_toolchain}/default:default") {
     "//examples/common/pigweed/rpc_console/py:chip_rpc",
     "//integrations/mobly:chip_mobly",
     "//scripts/py_matter_yamltests:matter_yamltests",
-    "//src/python_testing/matter_testing_infrastructure:metadata_parser",
+    "//src/python_testing/matter_testing_infrastructure:chip-testing",
   ]
 
   pw_python_venv("matter_build_venv") {
@@ -122,7 +122,7 @@ if (current_toolchain != "${dir_pw_toolchain}/default:default") {
       deps = [
         "${chip_root}/scripts:matter_yamltests_distribution.wheel",
         "${chip_root}/src/controller/python:chip-repl",
-        "${chip_root}/src/python_testing/matter_testing_infrastructure:metadata_parser.wheel",
+        "${chip_root}/src/python_testing/matter_testing_infrastructure:chip-testing.wheel",
       ]
     }
   }
@@ -249,7 +249,7 @@ if (current_toolchain != "${dir_pw_toolchain}/default:default") {
         "//scripts/py_matter_idl:matter_idl.tests",
         "//scripts/py_matter_yamltests:matter_yamltests.tests",
         "//src:tests_run",
-        "//src/python_testing/matter_testing_infrastructure:metadata_parser.tests",
+        "//src/python_testing/matter_testing_infrastructure:chip-testing.tests",
       ]
 
       if (current_os == "linux" || current_os == "mac") {

examples/fabric-admin/scripts/fabric-sync-app.py

@@ -79,12 +79,11 @@ async def forward_stdin(f_out: asyncio.StreamWriter):
 
 class Subprocess:
 
-    def __init__(self, tag: str, program: str, *args, stdout_cb=None):
+    def __init__(self, tag: str, program: str, *args):
         self.event = asyncio.Event()
         self.tag = tag.encode()
         self.program = program
         self.args = args
-        self.stdout_cb = stdout_cb
         self.expected_output = None
 
     def _check_output(self, line: bytes):
@@ -122,8 +121,7 @@ def terminate(self):
         self.p.terminate()
 
 
-async def run_admin(program, stdout_cb=None, storage_dir=None,
-                    rpc_admin_port=None, rpc_bridge_port=None,
+async def run_admin(program, storage_dir=None, rpc_admin_port=None, rpc_bridge_port=None,
                     paa_trust_store_path=None, commissioner_name=None,
                     commissioner_node_id=None, commissioner_vendor_id=None):
     args = []
@@ -141,8 +139,7 @@ async def run_admin(program, stdout_cb=None, storage_dir=None,
         args.extend(["--commissioner-nodeid", str(commissioner_node_id)])
     if commissioner_vendor_id is not None:
         args.extend(["--commissioner-vendor-id", str(commissioner_vendor_id)])
-    p = Subprocess("[FS-ADMIN]", program, "interactive", "start", *args,
-                   stdout_cb=stdout_cb)
+    p = Subprocess("[FS-ADMIN]", program, "interactive", "start", *args)
     await p.run()
     return p
 

scripts/build_python.sh

@@ -198,7 +198,7 @@ ninja -C "$OUTPUT_ROOT" python_wheels
 WHEEL=("$OUTPUT_ROOT"/controller/python/chip*.whl)
 
 # Add the matter_testing_infrastructure wheel
-WHEEL+=("$OUTPUT_ROOT"/python/obj/src/python_testing/matter_testing_infrastructure/metadata_parser._build_wheel/metadata_parser-*.whl)
+WHEEL+=("$OUTPUT_ROOT"/python/obj/src/python_testing/matter_testing_infrastructure/chip-testing._build_wheel/chip_testing-*.whl)
 
 if [ -n "$extra_packages" ]; then
     WHEEL+=("$extra_packages")

scripts/tests/run_python_test.py

@@ -31,8 +31,8 @@
 
 import click
 import coloredlogs
+from chip.testing.metadata import Metadata, MetadataReader
 from colorama import Fore, Style
-from metadata_parser.metadata import Metadata, MetadataReader
 
 DEFAULT_CHIP_ROOT = os.path.abspath(
     os.path.join(os.path.dirname(__file__), '..', '..'))

src/python_testing/TC_MCORE_FS_1_3.py

@@ -36,67 +36,33 @@
 import logging
 import os
 import random
-import subprocess
-import sys
 import tempfile
-import threading
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import Status
+from chip.testing.tasks import Subprocess
 from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches
 from mobly import asserts
 
 
-# TODO: Make this class more generic. Issue #35348
-class Subprocess(threading.Thread):
-
-    def __init__(self, args: list = [], tag="", **kw):
-        super().__init__(**kw)
-        self.tag = f"[{tag}] " if tag else ""
-        self.args = args
-
-    def forward_f(self, f_in, f_out):
-        while True:
-            line = f_in.readline()
-            if not line:
-                break
-            f_out.write(f"{self.tag}{line}")
-            f_out.flush()
-
-    def run(self):
-        logging.info("RUN: %s", " ".join(self.args))
-        self.p = subprocess.Popen(self.args, errors="ignore", stdin=subprocess.PIPE,
-                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        # Forward stdout and stderr with a tag attached.
-        t1 = threading.Thread(target=self.forward_f, args=[self.p.stdout, sys.stdout])
-        t1.start()
-        t2 = threading.Thread(target=self.forward_f, args=[self.p.stderr, sys.stderr])
-        t2.start()
-        # Wait for the process to finish.
-        self.p.wait()
-        t1.join()
-        t2.join()
-
-    def stop(self):
-        self.p.terminate()
-        self.join()
-
-
 class AppServer:
 
     def __init__(self, app, storage_dir, port=None, discriminator=None, passcode=None):
-
-        args = [app]
-        args.extend(["--KVS", tempfile.mkstemp(dir=storage_dir, prefix="kvs-app-")[1]])
+        args = [
+            "--KVS", tempfile.mkstemp(dir=storage_dir, prefix="kvs-app-")[1],
+        ]
         args.extend(['--secured-device-port', str(port)])
         args.extend(["--discriminator", str(discriminator)])
         args.extend(["--passcode", str(passcode)])
-        self.app = Subprocess(args, tag="SERVER")
-        self.app.start()
+        self.app = Subprocess(app, *args, prefix="[SERVER]")
+
+    def start(self):
+        # Start process and block until it prints the expected output.
+        self.app.start(expected_output="Server initialization complete")
 
-    def stop(self):
-        self.app.stop()
+    def terminate(self):
+        self.app.terminate()
 
 
 class TC_MCORE_FS_1_3(MatterBaseTest):
@@ -134,10 +100,11 @@ def setup_class(self):
             port=self.th_server_port,
             discriminator=self.th_server_discriminator,
             passcode=self.th_server_passcode)
+        self.th_server.start()
 
     def teardown_class(self):
         if self.th_server is not None:
-            self.th_server.stop()
+            self.th_server.terminate()
         if self.storage is not None:
             self.storage.cleanup()
         super().teardown_class()

src/python_testing/TC_MCORE_FS_1_4.py

@@ -36,86 +36,32 @@
 import logging
 import os
 import random
-import subprocess
-import sys
 import tempfile
-import threading
 
 import chip.clusters as Clusters
 from chip import ChipDeviceCtrl
 from chip.interaction_model import Status
+from chip.testing.tasks import Subprocess
 from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches
 from mobly import asserts
 
-# TODO: Make this class more generic. Issue #35348
-
-
-class Subprocess(threading.Thread):
-
-    def __init__(self, args: list = [], stdout_cb=None, tag="", **kw):
-        super().__init__(**kw)
-        self.tag = f"[{tag}] " if tag else ""
-        self.stdout_cb = stdout_cb
-        self.args = args
-
-    def forward_f(self, f_in, f_out):
-        while True:
-            line = f_in.readline()
-            if not line:
-                break
-            f_out.write(f"{self.tag}{line}")
-            f_out.flush()
-            if self.stdout_cb is not None:
-                self.stdout_cb(line)
-
-    def run(self):
-        logging.info("RUN: %s", " ".join(self.args))
-        self.p = subprocess.Popen(self.args, errors="ignore", stdin=subprocess.PIPE,
-                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        # Forward stdout and stderr with a tag attached.
-        forwarding_stdout_thread = threading.Thread(target=self.forward_f, args=[self.p.stdout, sys.stdout])
-        forwarding_stdout_thread.start()
-        forwarding_stderr_thread = threading.Thread(target=self.forward_f, args=[self.p.stderr, sys.stderr])
-        forwarding_stderr_thread.start()
-        # Wait for the process to finish.
-        self.p.wait()
-        forwarding_stdout_thread.join()
-        forwarding_stderr_thread.join()
-
-    def stop(self):
-        self.p.terminate()
-        self.join()
-
 
 class FabricSyncApp:
 
-    def _process_admin_output(self, line):
-        if self.wait_for_text_text is not None and self.wait_for_text_text in line:
-            self.wait_for_text_event.set()
-
-    def wait_for_text(self, timeout=30):
-        if not self.wait_for_text_event.wait(timeout=timeout):
-            raise Exception(f"Timeout waiting for text: {self.wait_for_text_text}")
-        self.wait_for_text_event.clear()
-        self.wait_for_text_text = None
-
     def __init__(self, fabric_sync_app_path, fabric_admin_app_path, fabric_bridge_app_path,
                  storage_dir, fabric_name=None, node_id=None, vendor_id=None,
                  paa_trust_store_path=None, bridge_port=None, bridge_discriminator=None,
                  bridge_passcode=None):
-
-        self.wait_for_text_event = threading.Event()
-        self.wait_for_text_text = None
-
-        args = [fabric_sync_app_path]
-        args.append(f"--app-admin={fabric_admin_app_path}")
-        args.append(f"--app-bridge={fabric_bridge_app_path}")
-        # Override default ports, so it will be possible to run
-        # our TH_FSA alongside the DUT_FSA during CI testing.
-        args.append("--app-admin-rpc-port=44000")
-        args.append("--app-bridge-rpc-port=44001")
-        # Keep the storage directory in a temporary location.
-        args.append(f"--storage-dir={storage_dir}")
+        args = [
+            f"--app-admin={fabric_admin_app_path}",
+            f"--app-bridge={fabric_bridge_app_path}",
+            # Override default ports, so it will be possible to run
+            # our TH_FSA alongside the DUT_FSA during CI testing.
+            "--app-admin-rpc-port=44000",
+            "--app-bridge-rpc-port=44001",
+            # Keep the storage directory in a temporary location.
+            f"--storage-dir={storage_dir}",
+        ]
         if paa_trust_store_path is not None:
             args.append(f"--paa-trust-store-path={paa_trust_store_path}")
         if fabric_name is not None:
@@ -127,55 +73,38 @@ def __init__(self, fabric_sync_app_path, fabric_admin_app_path, fabric_bridge_ap
         args.append(f"--discriminator={bridge_discriminator}")
         args.append(f"--passcode={bridge_passcode}")
 
-        self.fabric_sync_app = Subprocess(args, stdout_cb=self._process_admin_output)
-        self.wait_for_text_text = "Successfully opened pairing window on the device"
-        self.fabric_sync_app.start()
+        self.fabric_sync_app = Subprocess(fabric_sync_app_path, *args)
 
-        # Wait for the fabric-sync-app to be ready.
-        self.wait_for_text()
+    def start(self):
+        # Start process and block until it prints the expected output.
+        self.fabric_sync_app.start(expected_output="Successfully opened pairing window on the device")
 
-    def commission_on_network(self, node_id, setup_pin_code=None, filter_type=None, filter=None):
-        self.wait_for_text_text = f"Commissioning complete for node ID {node_id:#018x}: success"
-        # Send the commissioning command to the admin.
-        self.fabric_sync_app.p.stdin.write(f"pairing onnetwork {node_id} {setup_pin_code}\n")
-        self.fabric_sync_app.p.stdin.flush()
-        # Wait for success message.
-        self.wait_for_text()
+    def terminate(self):
+        self.fabric_sync_app.terminate()
 
-    def stop(self):
-        self.fabric_sync_app.stop()
+    def commission_on_network(self, node_id, setup_pin_code=None, filter_type=None, filter=None):
+        self.fabric_sync_app.send(
+            f"pairing onnetwork {node_id} {setup_pin_code}",
+            expected_output=f"Commissioning complete for node ID {node_id:#018x}: success")
 
 
 class AppServer:
 
-    def _process_admin_output(self, line):
-        if self.wait_for_text_text is not None and self.wait_for_text_text in line:
-            self.wait_for_text_event.set()
-
-    def wait_for_text(self, timeout=30):
-        if not self.wait_for_text_event.wait(timeout=timeout):
-            raise Exception(f"Timeout waiting for text: {self.wait_for_text_text}")
-        self.wait_for_text_event.clear()
-        self.wait_for_text_text = None
-
     def __init__(self, app, storage_dir, port=None, discriminator=None, passcode=None):
-        self.wait_for_text_event = threading.Event()
-        self.wait_for_text_text = None
-
-        args = [app]
-        args.extend(["--KVS", tempfile.mkstemp(dir=storage_dir, prefix="kvs-app-")[1]])
+        args = [
+            "--KVS", tempfile.mkstemp(dir=storage_dir, prefix="kvs-app-")[1],
+        ]
         args.extend(['--secured-device-port', str(port)])
         args.extend(["--discriminator", str(discriminator)])
         args.extend(["--passcode", str(passcode)])
-        self.app = Subprocess(args, stdout_cb=self._process_admin_output, tag="SERVER")
-        self.wait_for_text_text = "Server initialization complete"
-        self.app.start()
+        self.app = Subprocess(app, *args, prefix="[SERVER]")
 
-        # Wait for the server-app to be ready.
-        self.wait_for_text()
+    def start(self):
+        # Start process and block until it prints the expected output.
+        self.app.start(expected_output="Server initialization complete")
 
-    def stop(self):
-        self.app.stop()
+    def terminate(self):
+        self.app.terminate()
 
 
 class TC_MCORE_FS_1_4(MatterBaseTest):
@@ -237,6 +166,7 @@ def setup_class(self):
             bridge_discriminator=self.th_fsa_bridge_discriminator,
             bridge_passcode=self.th_fsa_bridge_passcode,
             vendor_id=0xFFF1)
+        self.th_fsa_controller.start()
 
         # Get the named pipe path for the DUT_FSA app input from the user params.
         dut_fsa_stdin_pipe = self.user_params.get("dut_fsa_stdin_pipe", None)
@@ -254,12 +184,13 @@ def setup_class(self):
             port=self.th_server_port,
             discriminator=self.th_server_discriminator,
             passcode=self.th_server_passcode)
+        self.th_server.start()
 
     def teardown_class(self):
         if self.th_fsa_controller is not None:
-            self.th_fsa_controller.stop()
+            self.th_fsa_controller.terminate()
         if self.th_server is not None:
-            self.th_server.stop()
+            self.th_server.terminate()
         if self.storage is not None:
             self.storage.cleanup()
         super().teardown_class()

src/python_testing/matter_testing_infrastructure/BUILD.gn

@@ -18,7 +18,8 @@ import("//build_overrides/chip.gni")
 import("//build_overrides/pigweed.gni")
 import("$dir_pw_build/python.gni")
 
-pw_python_package("metadata_parser") {
+# Python package for CHIP testing support.
+pw_python_package("chip-testing") {
   setup = [
     "setup.py",
     "setup.cfg",
@@ -28,9 +29,13 @@ pw_python_package("metadata_parser") {
   inputs = [ "env_test.yaml" ]
 
   sources = [
-    "metadata_parser/__init__.py",
-    "metadata_parser/metadata.py",
+    "chip/testing/__init__.py",
+    "chip/testing/metadata.py",
+    "chip/testing/tasks.py",
   ]
 
-  tests = [ "metadata_parser/test_metadata.py" ]
+  tests = [
+    "chip/testing/test_metadata.py",
+    "chip/testing/test_tasks.py",
+  ]
 }