Add --app-ready-pattern option to python test arguments (project-chip#35871)

* Reuse Subprocess in run_python_test.py script

* Add --app-ready-pattern option to run_python_test.py

* Replace script-start-delay with app-ready-pattern

* Drop support for script-start-delay

* Use rmtree() instead of explicit "rm -rf" calls

* Update missed python.md section

* Silence output from tests

* Fix removing files

* Restyled by prettier-markdown

* Fix documentation builder warning

* Remove unused import

* Fix Metadata unit test

---------

Co-authored-by: Restyled.io <commits@restyled.io>

Author: arkq

docs/testing/python.md

@@ -50,12 +50,19 @@ Python tests located in src/python_testing
 # for details about the block below.
 #
 # === BEGIN CI TEST ARGUMENTS ===
-# test-runner-runs: run1
-# test-runner-run/run1/app: ${ALL_CLUSTERS_APP}
-# test-runner-run/run1/factoryreset: True
-# test-runner-run/run1/quiet: True
-# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
-# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# test-runner-runs:
+#   run1:
+#     app: ${ALL_CLUSTERS_APP}
+#     app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
+#     script-args: >
+#       --storage-path admin_storage.json
+#       --commissioning-method on-network
+#       --discriminator 1234
+#       --passcode 20202021
+#       --trace-to json:${TRACE_TEST_JSON}.json
+#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+#     factoryreset: true
+#     quiet: true
 # === END CI TEST ARGUMENTS ===
 
 class TC_MYTEST_1_1(MatterBaseTest):
@@ -669,10 +676,10 @@ for that run, e.g.:
 # test-runner-runs:
 #   run1:
 #     app: ${TYPE_OF_APP}
-#     factoryreset: <true|false>
-#     quiet: <true|false>
 #     app-args: <app_arguments>
 #     script-args: <script_arguments>
+#     factoryreset: <true|false>
+#     quiet: <true|false>
 # === END CI TEST ARGUMENTS ===
 ```
 
@@ -701,19 +708,18 @@ for that run, e.g.:
     -   Example:
         `--discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json`
 
+-   `app-ready-pattern`: Regular expression pattern to match against the output
+    of the application to determine when the application is ready. If this
+    parameter is specified, the test runner will not run the test script until
+    the pattern is found.
+
+    -   Example: `"Manual pairing code: \\[\\d+\\]"`
+
 -   `script-args`: Specifies the arguments to be passed to the test script.
 
     -   Example:
         `--storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto`
 
--   `script-start-delay`: Specifies the number of seconds to wait before
-    starting the test script. This parameter can be used to allow the
-    application to initialize itself properly before the test script will try to
-    commission it (e.g. in case if the application needs to be commissioned to
-    some other controller first). By default, the delay is 0 seconds.
-
-    -   Example: `10`
-
 This structured format ensures that all necessary configurations are clearly
 defined and easily understood, allowing for consistent and reliable test
 execution.

scripts/tests/run_python_test.py

@@ -15,58 +15,57 @@
 # limitations under the License.
 
 import datetime
+import glob
 import io
 import logging
 import os
 import os.path
-import queue
+import pathlib
 import re
 import shlex
-import signal
-import subprocess
 import sys
-import threading
 import time
-import typing
 
 import click
 import coloredlogs
 from chip.testing.metadata import Metadata, MetadataReader
+from chip.testing.tasks import Subprocess
 from colorama import Fore, Style
 
 DEFAULT_CHIP_ROOT = os.path.abspath(
     os.path.join(os.path.dirname(__file__), '..', '..'))
 
 MATTER_DEVELOPMENT_PAA_ROOT_CERTS = "credentials/development/paa-root-certs"
 
+TAG_PROCESS_APP = f"[{Fore.GREEN}APP {Style.RESET_ALL}]".encode()
+TAG_PROCESS_TEST = f"[{Fore.GREEN}TEST{Style.RESET_ALL}]".encode()
+TAG_STDOUT = f"[{Fore.YELLOW}STDOUT{Style.RESET_ALL}]".encode()
+TAG_STDERR = f"[{Fore.RED}STDERR{Style.RESET_ALL}]".encode()
 
-def EnqueueLogOutput(fp, tag, output_stream, q):
-    for line in iter(fp.readline, b''):
-        timestamp = time.time()
-        if len(line) > len('[1646290606.901990]') and line[0:1] == b'[':
-            try:
-                timestamp = float(line[1:18].decode())
-                line = line[19:]
-            except Exception:
-                pass
-        output_stream.write(
-            (f"[{datetime.datetime.fromtimestamp(timestamp).isoformat(sep=' ')}]").encode() + tag + line)
-        sys.stdout.flush()
-    fp.close()
+# RegExp which matches the timestamp in the output of CHIP application
+OUTPUT_TIMESTAMP_MATCH = re.compile(r'(?P<prefix>.*)\[(?P<ts>\d+\.\d+)\](?P<suffix>\[\d+:\d+\].*)'.encode())
 
 
-def RedirectQueueThread(fp, tag, stream_output, queue) -> threading.Thread:
-    log_queue_thread = threading.Thread(target=EnqueueLogOutput, args=(
-        fp, tag, stream_output, queue))
-    log_queue_thread.start()
-    return log_queue_thread
+def chip_output_extract_timestamp(line: bytes) -> (float, bytes):
+    """Try to extract timestamp from a CHIP application output line."""
+    if match := OUTPUT_TIMESTAMP_MATCH.match(line):
+        return float(match.group(2)), match.group(1) + match.group(3) + b'\n'
+    return time.time(), line
 
 
-def DumpProgramOutputToQueue(thread_list: typing.List[threading.Thread], tag: str, process: subprocess.Popen, stream_output, queue: queue.Queue):
-    thread_list.append(RedirectQueueThread(process.stdout,
-                                           (f"[{tag}][{Fore.YELLOW}STDOUT{Style.RESET_ALL}]").encode(), stream_output, queue))
-    thread_list.append(RedirectQueueThread(process.stderr,
-                                           (f"[{tag}][{Fore.RED}STDERR{Style.RESET_ALL}]").encode(), stream_output, queue))
+def process_chip_output(line: bytes, is_stderr: bool, process_tag: bytes = b"") -> bytes:
+    """Rewrite the output line to add the timestamp and the process tag."""
+    timestamp, line = chip_output_extract_timestamp(line)
+    timestamp = datetime.datetime.fromtimestamp(timestamp).isoformat(sep=' ')
+    return f"[{timestamp}]".encode() + process_tag + (TAG_STDERR if is_stderr else TAG_STDOUT) + line
+
+
+def process_chip_app_output(line, is_stderr):
+    return process_chip_output(line, is_stderr, TAG_PROCESS_APP)
+
+
+def process_test_script_output(line, is_stderr):
+    return process_chip_output(line, is_stderr, TAG_PROCESS_TEST)
 
 
 @click.command()
@@ -77,7 +76,9 @@ def DumpProgramOutputToQueue(thread_list: typing.List[threading.Thread], tag: st
 @click.option("--factoryreset-app-only", is_flag=True,
               help='Remove app config and repl configs (/tmp/chip* and /tmp/repl*) before running the tests, but not the controller config')
 @click.option("--app-args", type=str, default='',
-              help='The extra arguments passed to the device. Can use placholders like {SCRIPT_BASE_NAME}')
+              help='The extra arguments passed to the device. Can use placeholders like {SCRIPT_BASE_NAME}')
+@click.option("--app-ready-pattern", type=str, default=None,
+              help='Delay test script start until given regular expression pattern is found in the application output.')
 @click.option("--script", type=click.Path(exists=True), default=os.path.join(DEFAULT_CHIP_ROOT,
                                                                              'src',
                                                                              'controller',
@@ -87,14 +88,12 @@ def DumpProgramOutputToQueue(thread_list: typing.List[threading.Thread], tag: st
                                                                              'mobile-device-test.py'), help='Test script to use.')
 @click.option("--script-args", type=str, default='',
               help='Script arguments, can use placeholders like {SCRIPT_BASE_NAME}.')
-@click.option("--script-start-delay", type=int, default=0,
-              help='Delay in seconds before starting the script.')
 @click.option("--script-gdb", is_flag=True,
               help='Run script through gdb')
 @click.option("--quiet", is_flag=True, help="Do not print output from passing tests. Use this flag in CI to keep github log sizes manageable.")
 @click.option("--load-from-env", default=None, help="YAML file that contains values for environment variables.")
 def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: str,
-         script: str, script_args: str, script_start_delay: int, script_gdb: bool, quiet: bool, load_from_env):
+         app_ready_pattern: str, script: str, script_args: str, script_gdb: bool, quiet: bool, load_from_env):
     if load_from_env:
         reader = MetadataReader(load_from_env)
         runs = reader.parse_script(script)
@@ -105,10 +104,10 @@ def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: st
                 run="cmd-run",
                 app=app,
                 app_args=app_args,
+                app_ready_pattern=app_ready_pattern,
                 script_args=script_args,
-                script_start_delay=script_start_delay,
-                factoryreset=factoryreset,
-                factoryreset_app_only=factoryreset_app_only,
+                factory_reset=factoryreset,
+                factory_reset_app_only=factoryreset_app_only,
                 script_gdb=script_gdb,
                 quiet=quiet
             )
@@ -118,49 +117,38 @@ def main(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: st
         raise Exception(
             "No valid runs were found. Make sure you add runs to your file, see https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md document for reference/example.")
 
+    coloredlogs.install(level='INFO')
+
     for run in runs:
-        print(f"Executing {run.py_script_path.split('/')[-1]} {run.run}")
-        main_impl(run.app, run.factoryreset, run.factoryreset_app_only, run.app_args,
-                  run.py_script_path, run.script_args, run.script_start_delay, run.script_gdb, run.quiet)
+        logging.info("Executing %s %s", run.py_script_path.split('/')[-1], run.run)
+        main_impl(run.app, run.factory_reset, run.factory_reset_app_only, run.app_args or "",
+                  run.app_ready_pattern, run.py_script_path, run.script_args or "", run.script_gdb, run.quiet)
 
 
-def main_impl(app: str, factoryreset: bool, factoryreset_app_only: bool, app_args: str,
-              script: str, script_args: str, script_start_delay: int, script_gdb: bool, quiet: bool):
+def main_impl(app: str, factory_reset: bool, factory_reset_app_only: bool, app_args: str,
+              app_ready_pattern: str, script: str, script_args: str, script_gdb: bool, quiet: bool):
 
     app_args = app_args.replace('{SCRIPT_BASE_NAME}', os.path.splitext(os.path.basename(script))[0])
     script_args = script_args.replace('{SCRIPT_BASE_NAME}', os.path.splitext(os.path.basename(script))[0])
 
-    if factoryreset or factoryreset_app_only:
+    if factory_reset or factory_reset_app_only:
         # Remove native app config
-        retcode = subprocess.call("rm -rf /tmp/chip* /tmp/repl*", shell=True)
-        if retcode != 0:
-            raise Exception("Failed to remove /tmp/chip* for factory reset.")
+        for path in glob.glob('/tmp/chip*') + glob.glob('/tmp/repl*'):
+            pathlib.Path(path).unlink(missing_ok=True)
 
         # Remove native app KVS if that was used
-        kvs_match = re.search(r"--KVS (?P<kvs_path>[^ ]+)", app_args)
-        if kvs_match:
-            kvs_path_to_remove = kvs_match.group("kvs_path")
-            retcode = subprocess.call("rm -f %s" % kvs_path_to_remove, shell=True)
-            print("Trying to remove KVS path %s" % kvs_path_to_remove)
-            if retcode != 0:
-                raise Exception("Failed to remove %s for factory reset." % kvs_path_to_remove)
-
-    if factoryreset:
-        # Remove Python test admin storage if provided
-        storage_match = re.search(r"--storage-path (?P<storage_path>[^ ]+)", script_args)
-        if storage_match:
-            storage_path_to_remove = storage_match.group("storage_path")
-            retcode = subprocess.call("rm -f %s" % storage_path_to_remove, shell=True)
-            print("Trying to remove storage path %s" % storage_path_to_remove)
-            if retcode != 0:
-                raise Exception("Failed to remove %s for factory reset." % storage_path_to_remove)
+        if match := re.search(r"--KVS (?P<path>[^ ]+)", app_args):
+            logging.info("Removing KVS path: %s" % match.group("path"))
+            pathlib.Path(match.group("path")).unlink(missing_ok=True)
 
-    coloredlogs.install(level='INFO')
-
-    log_queue = queue.Queue()
-    log_cooking_threads = []
+    if factory_reset:
+        # Remove Python test admin storage if provided
+        if match := re.search(r"--storage-path (?P<path>[^ ]+)", script_args):
+            logging.info("Removing storage path: %s" % match.group("path"))
+            pathlib.Path(match.group("path")).unlink(missing_ok=True)
 
     app_process = None
+    app_exit_code = 0
     app_pid = 0
 
     stream_output = sys.stdout.buffer
@@ -171,16 +159,15 @@ def main_impl(app: str, factoryreset: bool, factoryreset_app_only: bool, app_arg
         if not os.path.exists(app):
             if app is None:
                 raise FileNotFoundError(f"{app} not found")
-        app_args = [app] + shlex.split(app_args)
-        logging.info(f"Execute: {app_args}")
-        app_process = subprocess.Popen(
-            app_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE, bufsize=0)
-        app_process.stdin.close()
-        app_pid = app_process.pid
-        DumpProgramOutputToQueue(
-            log_cooking_threads, Fore.GREEN + "APP " + Style.RESET_ALL, app_process, stream_output, log_queue)
-
-    time.sleep(script_start_delay)
+        if app_ready_pattern:
+            app_ready_pattern = re.compile(app_ready_pattern.encode())
+        app_process = Subprocess(app, *shlex.split(app_args),
+                                 output_cb=process_chip_app_output,
+                                 f_stdout=stream_output,
+                                 f_stderr=stream_output)
+        app_process.start(expected_output=app_ready_pattern, timeout=30)
+        app_process.p.stdin.close()
+        app_pid = app_process.p.pid
 
     script_command = [script, "--paa-trust-store-path", os.path.join(DEFAULT_CHIP_ROOT, MATTER_DEVELOPMENT_PAA_ROOT_CERTS),
                       '--log-format', '%(message)s', "--app-pid", str(app_pid)] + shlex.split(script_args)
@@ -198,31 +185,24 @@ def main_impl(app: str, factoryreset: bool, factoryreset_app_only: bool, app_arg
 
     final_script_command = [i.replace('|', ' ') for i in script_command]
 
-    logging.info(f"Execute: {final_script_command}")
-    test_script_process = subprocess.Popen(
-        final_script_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
-    test_script_process.stdin.close()
-    DumpProgramOutputToQueue(log_cooking_threads, Fore.GREEN + "TEST" + Style.RESET_ALL,
-                             test_script_process, stream_output, log_queue)
-
+    test_script_process = Subprocess(final_script_command[0], *final_script_command[1:],
+                                     output_cb=process_test_script_output,
+                                     f_stdout=stream_output,
+                                     f_stderr=stream_output)
+    test_script_process.start()
+    test_script_process.p.stdin.close()
     test_script_exit_code = test_script_process.wait()
 
     if test_script_exit_code != 0:
-        logging.error("Test script exited with error %r" % test_script_exit_code)
+        logging.error("Test script exited with returncode %d" % test_script_exit_code)
 
-    test_app_exit_code = 0
     if app_process:
-        logging.warning("Stopping app with SIGINT")
-        app_process.send_signal(signal.SIGINT.value)
-        test_app_exit_code = app_process.wait()
-
-    # There are some logs not cooked, so we wait until we have processed all logs.
-    # This procedure should be very fast since the related processes are finished.
-    for thread in log_cooking_threads:
-        thread.join()
+        logging.info("Stopping app with SIGTERM")
+        app_process.terminate()
+        app_exit_code = app_process.returncode
 
     # We expect both app and test script should exit with 0
-    exit_code = test_script_exit_code if test_script_exit_code != 0 else test_app_exit_code
+    exit_code = test_script_exit_code or app_exit_code
 
     if quiet:
         if exit_code:

src/controller/python/test/test_scripts/mobile-device-test.py

@@ -23,12 +23,18 @@
 # for details about the block below.
 #
 # === BEGIN CI TEST ARGUMENTS ===
-# test-runner-runs: run1
-# test-runner-run/run1/app: ${ALL_CLUSTERS_APP}
-# test-runner-run/run1/factoryreset: True
-# test-runner-run/run1/quiet: True
-# test-runner-run/run1/app-args: --trace-to json:${TRACE_APP}.json
-# test-runner-run/run1/script-args: --log-level INFO -t 3600 --disable-test ClusterObjectTests.TestTimedRequestTimeout --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# test-runner-runs:
+#   run1:
+#     app: ${ALL_CLUSTERS_APP}
+#     app-args: --trace-to json:${TRACE_APP}.json
+#     script-args: >
+#       --log-level INFO
+#       --timeout 3600
+#       --disable-test ClusterObjectTests.TestTimedRequestTimeout
+#       --trace-to json:${TRACE_TEST_JSON}.json
+#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+#     factoryreset: true
+#     quiet: true
 # === END CI TEST ARGUMENTS ===
 
 import asyncio

src/python_testing/TCP_Tests.py

@@ -15,12 +15,19 @@
 #    limitations under the License.
 #
 # === BEGIN CI TEST ARGUMENTS ===
-# test-runner-runs: run1
-# test-runner-run/run1/app: ${ALL_CLUSTERS_APP}
-# test-runner-run/run1/factoryreset: True
-# test-runner-run/run1/quiet: True
-# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
-# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+# test-runner-runs:
+#   run1:
+#     app: ${ALL_CLUSTERS_APP}
+#     app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
+#     script-args: >
+#       --storage-path admin_storage.json
+#       --commissioning-method on-network
+#       --discriminator 1234
+#       --passcode 20202021
+#       --trace-to json:${TRACE_TEST_JSON}.json
+#       --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
+#     factoryreset: true
+#     quiet: true
 # === END CI TEST ARGUMENTS ===
 #
 import chip.clusters as Clusters