Create test script to verify that the Linux tv-casting-app is able to discover the Linux tv-app.

.github/workflows/examples-linux-tv-casting-app.yaml

@@ -17,7 +17,7 @@ name: Test TV Casting Example
 on:
     push:
         branches-ignore:
-            - 'dependabot/**'
+            - "dependabot/**"
     pull_request:
     merge_group:
 
@@ -63,6 +63,12 @@ jobs:
                   ./scripts/run_in_build_env.sh \
                     "scripts/examples/gn_build_example.sh examples/tv-casting-app/linux/ out/tv-casting-app"
 
+            - name: Test casting from Linux tv-casting-app to Linux tv-app
+              run: |
+                  ./scripts/run_in_build_env.sh \
+                    "python3 ./scripts/tests/run_tv_casting_test.py"
+              timeout-minutes: 1
+
             - name: Uploading Size Reports
               uses: ./.github/actions/upload-size-reports
               if: ${{ !env.ACT }}

scripts/tests/run_tv_casting_test.py

@@ -0,0 +1,225 @@
+#!/usr/bin/env -S python3 -B
+
+# 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 logging
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from typing import List, Optional, TextIO, Tuple
+
+import click
+
+# Configure logging format.
+logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
+
+# The maximum amount of time to wait for the Linux tv-app to start before timeout.
+TV_APP_MAX_START_WAIT_SEC = 2
+
+# File names of logs for the Linux tv-casting-app and the Linux tv-app.
+LINUX_TV_APP_LOGS = 'Linux-tv-app-logs.txt'
+LINUX_TV_CASTING_APP_LOGS = 'Linux-tv-casting-app-logs.txt'
+
+# Values that identify the Linux tv-app and are noted in the 'Device Configuration' in the Linux tv-app output
+# as well as under the 'Discovered Commissioner' details in the Linux tv-casting-app output.
+VENDOR_ID = 0xFFF1   # Spec 7.20.2.1 MEI code: test vendor IDs are 0xFFF1 to 0xFFF4
+PRODUCT_ID = 0x8001  # Test product id
+DEVICE_TYPE_CASTING_VIDEO_PLAYER = 0x23    # Device type library 10.3: Casting Video Player
+
+
+class ProcessManager:
+    """A context manager for managing subprocesses.
+
+    This class provides a context manager for safely starting and stopping a subprocess.
+    """
+
+    def __init__(self, command: List[str], stdout, stderr):
+        self.command = command
+        self.stdout = stdout
+        self.stderr = stderr
+
+    def __enter__(self):
+        self.process = subprocess.Popen(self.command, stdout=self.stdout, stderr=self.stderr, text=True)
+        return self.process
+
+    def __exit__(self, exception_type, exception_value, traceback):
+        self.process.terminate()
+        self.process.wait()
+
+
+def dump_temporary_logs_to_console(log_file_path: str):
+    """Dump log file to the console; log file will be removed once the function exits."""
+    """Write the entire content of `log_file_path` to the console."""
+    print('\nDumping logs from: ', log_file_path)
+
+    with open(log_file_path, 'r') as file:
+        for line in file:
+            print(line.rstrip())
+
+
+def handle_discovery_failure(log_file_paths: List[str]):
+    """Log 'Discovery failed!' as error, dump log files to console, exit on error."""
+    logging.error('Discovery failed!')
+
+    for log_file_path in log_file_paths:
+        try:
+            dump_temporary_logs_to_console(log_file_path)
+        except Exception as e:
+            logging.exception(f"Failed to dump {log_file_path}: {e}")
+
+    sys.exit(1)
+
+
+def extract_value_from_string(line: str) -> int:
+    """Extract and return integer value from given output string.
+
+    The string is expected to be in the following format as it is received
+    from the Linux tv-casting-app output:
+    \x1b[0;34m[1713741926895] [7276:9521344] [DIS] Vendor ID: 65521\x1b[0m
+    The integer value to be extracted here is 65521.
+    """
+    value = line.split(':')[-1].strip().replace('\x1b[0m', '')
+    value = int(value)
+
+    return value
+
+
+def validate_value(expected_value: int, log_paths: List[str], line: str, value_name: str) -> Optional[str]:
+    """Validate a value in a string against an expected value."""
+    value = extract_value_from_string(line)
+
+    if value != expected_value:
+        logging.error(f'{value_name} does not match the expected value!')
+        logging.error(f'Expected {value_name}: {expected_value}')
+        logging.error(line.rstrip('\n'))
+        handle_discovery_failure(log_paths)
+        return None
+
+    # Return the line containing the valid value.
+    return line.rstrip('\n')
+
+
+def start_up_tv_app_success(tv_app_process: subprocess.Popen, linux_tv_app_log_file: TextIO) -> bool:
+    """Check if the Linux tv-app is able to successfully start or until timeout occurs."""
+    start_wait_time = time.time()
+
+    while True:
+        # Check if the time elapsed since the start wait time exceeds the maximum allowed startup time for the TV app.
+        if time.time() - start_wait_time > TV_APP_MAX_START_WAIT_SEC:
+            logging.error("The Linux tv-app process did not start successfully within the timeout.")
+            return False
+
+        tv_app_output_line = tv_app_process.stdout.readline()
+
+        linux_tv_app_log_file.write(tv_app_output_line)
+        linux_tv_app_log_file.flush()
+
+        # Check if the Linux tv-app started successfully.
+        if "Started commissioner" in tv_app_output_line:
+            logging.info('Linux tv-app is up and running!')
+            return True
+
+
+def parse_output_for_valid_commissioner(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]):
+    """Parse the output of the Linux tv-casting-app to find a valid commissioner."""
+    tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info
+
+    valid_discovered_commissioner = None
+    valid_vendor_id = None
+    valid_product_id = None
+    valid_device_type = None
+
+    # Read the output as we receive it from the tv-casting-app subprocess.
+    for line in tv_casting_app_process.stdout:
+        linux_tv_casting_app_log_file.write(line)
+        linux_tv_casting_app_log_file.flush()
+
+        # Fail fast if "No commissioner discovered" string found.
+        if "No commissioner discovered" in line:
+            logging.error(line.rstrip('\n'))
+            handle_discovery_failure(log_paths)
+
+        elif "Discovered Commissioner" in line:
+            valid_discovered_commissioner = line.rstrip('\n')
+
+        elif valid_discovered_commissioner:
+            # Continue parsing the output for the information of interest under 'Discovered Commissioner'
+            if 'Vendor ID:' in line:
+                valid_vendor_id = validate_value(VENDOR_ID, log_paths, line, 'Vendor ID')
+
+            elif 'Product ID:' in line:
+                valid_product_id = validate_value(PRODUCT_ID, log_paths, line, 'Product ID')
+
+            elif 'Device Type:' in line:
+                valid_device_type = validate_value(DEVICE_TYPE_CASTING_VIDEO_PLAYER, log_paths, line, 'Device Type')
+
+        # A valid commissioner has VENDOR_ID, PRODUCT_ID, and DEVICE TYPE in its list of entries.
+        if valid_vendor_id and valid_product_id and valid_device_type:
+            logging.info('Found a valid commissioner in the Linux tv-casting-app logs:')
+            logging.info(valid_discovered_commissioner)
+            logging.info(valid_vendor_id)
+            logging.info(valid_product_id)
+            logging.info(valid_device_type)
+            logging.info('Discovery success!')
+            break
+
+
+@click.command()
+@click.option('--tv-app-rel-path', type=str, default='out/tv-app/chip-tv-app', help='Path to the Linux tv-app executable.')
+@click.option('--tv-casting-app-rel-path', type=str, default='out/tv-casting-app/chip-tv-casting-app', help='Path to the Linux tv-casting-app executable.')
+def test_discovery_fn(tv_app_rel_path, tv_casting_app_rel_path):
+    """Test if the Linux tv-casting-app is able to discover the Linux tv-app.
+
+    Default paths for the executables are provided but can be overridden via command line arguments.
+    For example: python3 run_tv_casting_test.py --tv-app-rel-path=path/to/tv-app
+                 --tv-casting-app-rel-path=path/to/tv-casting-app
+    """
+    # Store the log files to a temporary directory.
+    with tempfile.TemporaryDirectory() as temp_dir:
+        linux_tv_app_log_path = os.path.join(temp_dir, LINUX_TV_APP_LOGS)
+        linux_tv_casting_app_log_path = os.path.join(temp_dir, LINUX_TV_CASTING_APP_LOGS)
+
+        with open(linux_tv_app_log_path, 'w') as linux_tv_app_log_file, open(linux_tv_casting_app_log_path, 'w') as linux_tv_casting_app_log_file:
+
+            # Configure command options to disable stdout buffering during tests.
+            disable_stdout_buffering_cmd = []
+            # On Unix-like systems, use stdbuf to disable stdout buffering.
+            if sys.platform == 'darwin' or sys.platform == 'linux':
+                disable_stdout_buffering_cmd = ['stdbuf', '-o0', '-i0']
+
+            tv_app_abs_path = os.path.abspath(tv_app_rel_path)
+            # Run the Linux tv-app subprocess.
+            with ProcessManager(disable_stdout_buffering_cmd + [tv_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_app_process:
+
+                if not start_up_tv_app_success(tv_app_process, linux_tv_app_log_file):
+                    handle_discovery_failure([linux_tv_app_log_path])
+
+                tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path)
+                # Run the Linux tv-casting-app subprocess.
+                with ProcessManager(disable_stdout_buffering_cmd + [tv_casting_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_casting_app_process:
+                    log_paths = [linux_tv_app_log_path, linux_tv_casting_app_log_path]
+                    tv_casting_app_info = (tv_casting_app_process, linux_tv_casting_app_log_file)
+                    parse_output_for_valid_commissioner(tv_casting_app_info, log_paths)
+
+
+if __name__ == '__main__':
+
+    # Start with a clean slate by removing any previously cached entries.
+    os.system('rm -f /tmp/chip_*')
+
+    # Test discovery between the Linux tv-casting-app and the tv-app.
+    test_discovery_fn()