From 7cdea9b7f6b6e71ec8023e105848c82d3e2ab2e4 Mon Sep 17 00:00:00 2001 From: Shao Ling Tan <161761051+shaoltan-amazon@users.noreply.github.com> Date: Thu, 25 Apr 2024 12:17:01 -0700 Subject: [PATCH] Create test script to verify that the Linux tv-casting-app is able to discover the Linux tv-app. (#32919) --- .../examples-linux-tv-casting-app.yaml | 8 +- scripts/tests/run_tv_casting_test.py | 225 ++++++++++++++++++ 2 files changed, 232 insertions(+), 1 deletion(-) create mode 100644 scripts/tests/run_tv_casting_test.py diff --git a/.github/workflows/examples-linux-tv-casting-app.yaml b/.github/workflows/examples-linux-tv-casting-app.yaml index e7ecb5bdddecc7..679f4be57a35db 100644 --- a/.github/workflows/examples-linux-tv-casting-app.yaml +++ b/.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 }} diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py new file mode 100644 index 00000000000000..7e783102a2fbc5 --- /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()