Create test script to verify that the Linux tv-casting-app is able to discover the Linux tv-app. (#32919)
diff --git a/.github/workflows/examples-linux-tv-casting-app.yaml b/.github/workflows/examples-linux-tv-casting-app.yaml
index e7ecb5b..679f4be 100644
--- a/.github/workflows/examples-linux-tv-casting-app.yaml
+++ b/.github/workflows/examples-linux-tv-casting-app.yaml
@@ -17,7 +17,7 @@
on:
push:
branches-ignore:
- - 'dependabot/**'
+ - "dependabot/**"
pull_request:
merge_group:
@@ -63,6 +63,12 @@
./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 }}
diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py
new file mode 100644
index 0000000..7e78310
--- /dev/null
+++ b/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()
