From fd949ea94f8da66f56b7ee4cff96a893de057c9c Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Tue, 9 Apr 2024 13:06:48 -0700 Subject: [PATCH 01/12] Create test script to verify that the Linux tv-casting-app is able to discover the Linux tv-app. (#32918) --- .../examples-linux-tv-casting-app.yaml | 5 + scripts/tests/run_tv_casting_test.py | 178 ++++++++++++++++++ 2 files changed, 183 insertions(+) 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..cb6b726c23db96 100644 --- a/.github/workflows/examples-linux-tv-casting-app.yaml +++ b/.github/workflows/examples-linux-tv-casting-app.yaml @@ -63,6 +63,11 @@ jobs: ./scripts/run_in_build_env.sh \ "scripts/examples/gn_build_example.sh examples/tv-casting-app/linux/ out/tv-casting-app" + - name: Test Discovery between Linux tv-casting-app and Linux tv-app + run: | + ./scripts/run_in_build_env.sh \ + "python3 ./scripts/tests/run_tv_casting_test.py test-discovery" + - 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..3912a4a08e5b5c --- /dev/null +++ b/scripts/tests/run_tv_casting_test.py @@ -0,0 +1,178 @@ +#!/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 os +import subprocess +import sys +import time + +import click + + +# Dump the logs to the console in the case of an error. +def dump_logs_to_console(log_file): + with open(log_file, 'r') as file: + logs = file.read() + print(logs) + + +# Remove the log files once the script is done running. +def remove_log_file(log_file): + if os.path.exists(log_file): + os.remove(log_file) + else: + print("The file does not exist.") + + +# Read the logs from the Linux-tv-casting-app-logs.txt file. +# The discovered commissioner(s) will be stored in a list along with their +# vendor ID, product ID, and device type. +def read_linux_tv_casting_app_logs(log_file): + + with open(log_file, 'r') as file: + lines = file.readlines() + + discovered_commissioners = [] + + print('Reading from Linux-tv-casting-app-logs.txt') + + # Read through the Linux-tv-casting-app-logs.txt line by line + for i, line in enumerate(lines): + + # If commissioner(s) are discovered, then the discovery process was successful. + if "commissioner(s) discovered" in line: + print(line) + print('Discovery success!') + + remove_log_file('./scripts/tests/Linux-tv-casting-app-logs.txt') + remove_log_file('./scripts/tests/Linux-tv-app-logs.txt') + + break + + # If no commissioner was discovered, then something went wrong. + # Exit on error. + if "No commissioner discovered" in line: + print(line) + print('Discovery failed!') + + dump_logs_to_console('./scripts/tests/Linux-tv-casting-app-logs.txt') + + remove_log_file('./scripts/tests/Linux-tv-casting-app-logs.txt') + remove_log_file('./scripts/tests/Linux-tv-app-logs.txt') + + sys.exit(1) + + # Look for "Discovered Commissioner" + if "Discovered Commissioner" in line: + print(line) + + # Extract the relevant part of the string + commissioner = line.split("Discovered Commissioner")[-1].strip() + commissioner = commissioner.replace('\x1b[0m', '') + + # Initialize variables for Vendor ID, Product ID, and Device Type + vendor_id = None + product_id = None + device_type = None + + # Iterate through the subsequent lines to find the strings of interest + for next_line in lines[i+1:]: + + if "Vendor ID:" in next_line: + print(next_line) + + vendor_id = next_line.split(":")[-1].strip() + vendor_id = vendor_id.replace('\x1b[0m', '') + + elif "Product ID:" in next_line: + print(next_line) + + product_id = next_line.split(":")[-1].strip() + product_id = product_id.replace('\x1b[0m', '') + + elif "Device Type:" in next_line: + print(next_line) + + device_type = next_line.split(":")[-1].strip() + device_type = device_type.replace('\x1b[0m', '') + + elif "commissioner(s) discovered" in next_line: + break + + # If the next line starts with "Discovered Commissioner", break the loop + if "Discovered Commissioner" in next_line: + break + + # Append the extracted information to the devices list + discovered_commissioners.append({ + "discovered_commissioner": commissioner, + "vendor_id": vendor_id, + "product_id": product_id, + "device_type": device_type + }) + + # If the list of discovered commissioners is empty and we didn't find the "No commissioner discovered" string, + # then something went wrong. Exit on error. + if len(discovered_commissioners) == 0: + print('Discovery failed! No commissioner(s) discovered! The list of discovered commissioner(s) is empty!') + + dump_logs_to_console('./scripts/tests/Linux-tv-casting-app-logs.txt') + + remove_log_file('./scripts/tests/Linux-tv-casting-app-logs.txt') + remove_log_file('./scripts/tests/Linux-tv-app-logs.txt') + + sys.exit(1) + + +# Test if the Linux tv-casting-app is able to discover the Linux tv-app. +# The Linux tv-casting-app and the tv-app will be run in separate processes. +# Their corresponding output will be written to their respective log files. +# The log file of the tv-casting-app will be parsed for strings of interest +# which will be printed to the console. +def test_discovery_fn(): + with open('./scripts/tests/Linux-tv-app-logs.txt', 'w') as fd1, open('./scripts/tests/Linux-tv-casting-app-logs.txt', 'w') as fd2: + + # Run the Linux tv-app and write the output to file + tv_app_rel_path = 'out/tv-app/chip-tv-app' + tv_app_abs_path = os.path.abspath(tv_app_rel_path) + p1 = subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) + + time.sleep(5) + + # Run the Linux tv-casting-app and write the output to file + tv_casting_app_rel_path = 'out/tv-casting-app/chip-tv-casting-app' + tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) + p2 = subprocess.Popen(tv_casting_app_abs_path, stdout=fd2, stderr=subprocess.PIPE, text=True) + + # Wait for the processes to finish writing before attempting to read + time.sleep(15) + + read_linux_tv_casting_app_logs('./scripts/tests/Linux-tv-casting-app-logs.txt') + + +@click.group() +def main(): + pass + + +@main.command('test-discovery', help='Test if the Linux tv-casting-app is able to discover the Linux tv-app.') +def test_discovery(): + test_discovery_fn() + + +if __name__ == '__main__': + + main() From d78be632e8d7ba86caa345902c5ead2349100017 Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Tue, 9 Apr 2024 13:48:37 -0700 Subject: [PATCH 02/12] Remove unused variables to address code-lints error. --- scripts/tests/run_tv_casting_test.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 3912a4a08e5b5c..fea55dd4719c03 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -148,14 +148,14 @@ def test_discovery_fn(): # Run the Linux tv-app and write the output to file tv_app_rel_path = 'out/tv-app/chip-tv-app' tv_app_abs_path = os.path.abspath(tv_app_rel_path) - p1 = subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) + subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) time.sleep(5) # Run the Linux tv-casting-app and write the output to file tv_casting_app_rel_path = 'out/tv-casting-app/chip-tv-casting-app' tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) - p2 = subprocess.Popen(tv_casting_app_abs_path, stdout=fd2, stderr=subprocess.PIPE, text=True) + subprocess.Popen(tv_casting_app_abs_path, stdout=fd2, stderr=subprocess.PIPE, text=True) # Wait for the processes to finish writing before attempting to read time.sleep(15) From d81306fcfa284a9d6201df85ea7a205340c3e3f4 Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Wed, 10 Apr 2024 14:46:53 -0700 Subject: [PATCH 03/12] Addressed PR comments and re-organized code structure. --- scripts/tests/run_tv_casting_test.py | 123 ++++++++++++++++++--------- 1 file changed, 85 insertions(+), 38 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index fea55dd4719c03..58358250af3083 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -21,6 +21,16 @@ import click +LINUX_TV_APP_LOGS = './scripts/tests/Linux-tv-app-logs.txt' +LINUX_TV_CASTING_APP_LOGS = './scripts/tests/Linux-tv-casting-app-logs.txt' + +RUN_INTERVAL = 5 +PARSE_INTERVAL = 15 + +VENDOR_ID = 65521 +PRODUCT_ID = 32769 +DEVICE_TYPE = 35 + # Dump the logs to the console in the case of an error. def dump_logs_to_console(log_file): @@ -37,10 +47,46 @@ def remove_log_file(log_file): print("The file does not exist.") +# Whenever a failure is discovered, we should print 'Discovery failed!', +# dump the logs, clean up the log files, exit on error. +def handle_discovery_failure(): + print('Discovery failed!') + + dump_logs_to_console(LINUX_TV_CASTING_APP_LOGS) + + remove_log_file(LINUX_TV_CASTING_APP_LOGS) + remove_log_file(LINUX_TV_APP_LOGS) + + sys.exit(1) + + +# Helper function to extract the integer value from a string. +def extract_value_from_string(line): + value = line.split(":")[-1].strip().replace('\x1b[0m', '') + value = int(value) + + return value + + +# Check if the discovered value matches the expected value. +def check_expected_value(line, expected_value, value_name): + # Extract the integer value from the string + value = extract_value_from_string(line) + + # If the discovered value does not match the expected value, + # print the error and handle the discovery failure. + if value != expected_value: + print(f'{value_name} does not match the expected value!') + print(f'Discovered {value_name}: {value}') + print(f'Expected {value_name}: {expected_value}') + + handle_discovery_failure() + + # Read the logs from the Linux-tv-casting-app-logs.txt file. # The discovered commissioner(s) will be stored in a list along with their # vendor ID, product ID, and device type. -def read_linux_tv_casting_app_logs(log_file): +def parse_linux_tv_casting_app_logs(log_file): with open(log_file, 'r') as file: lines = file.readlines() @@ -57,24 +103,11 @@ def read_linux_tv_casting_app_logs(log_file): print(line) print('Discovery success!') - remove_log_file('./scripts/tests/Linux-tv-casting-app-logs.txt') - remove_log_file('./scripts/tests/Linux-tv-app-logs.txt') + remove_log_file(LINUX_TV_CASTING_APP_LOGS) + remove_log_file(LINUX_TV_APP_LOGS) break - # If no commissioner was discovered, then something went wrong. - # Exit on error. - if "No commissioner discovered" in line: - print(line) - print('Discovery failed!') - - dump_logs_to_console('./scripts/tests/Linux-tv-casting-app-logs.txt') - - remove_log_file('./scripts/tests/Linux-tv-casting-app-logs.txt') - remove_log_file('./scripts/tests/Linux-tv-app-logs.txt') - - sys.exit(1) - # Look for "Discovered Commissioner" if "Discovered Commissioner" in line: print(line) @@ -93,21 +126,15 @@ def read_linux_tv_casting_app_logs(log_file): if "Vendor ID:" in next_line: print(next_line) - - vendor_id = next_line.split(":")[-1].strip() - vendor_id = vendor_id.replace('\x1b[0m', '') + vendor_id = extract_value_from_string(next_line) elif "Product ID:" in next_line: print(next_line) - - product_id = next_line.split(":")[-1].strip() - product_id = product_id.replace('\x1b[0m', '') + product_id = extract_value_from_string(next_line) elif "Device Type:" in next_line: print(next_line) - - device_type = next_line.split(":")[-1].strip() - device_type = device_type.replace('\x1b[0m', '') + device_type = extract_value_from_string(next_line) elif "commissioner(s) discovered" in next_line: break @@ -127,14 +154,8 @@ def read_linux_tv_casting_app_logs(log_file): # If the list of discovered commissioners is empty and we didn't find the "No commissioner discovered" string, # then something went wrong. Exit on error. if len(discovered_commissioners) == 0: - print('Discovery failed! No commissioner(s) discovered! The list of discovered commissioner(s) is empty!') - - dump_logs_to_console('./scripts/tests/Linux-tv-casting-app-logs.txt') - - remove_log_file('./scripts/tests/Linux-tv-casting-app-logs.txt') - remove_log_file('./scripts/tests/Linux-tv-app-logs.txt') - - sys.exit(1) + print('No commissioner(s) discovered! The list of discovered commissioner(s) is empty!') + handle_discovery_failure() # Test if the Linux tv-casting-app is able to discover the Linux tv-app. @@ -143,24 +164,48 @@ def read_linux_tv_casting_app_logs(log_file): # The log file of the tv-casting-app will be parsed for strings of interest # which will be printed to the console. def test_discovery_fn(): - with open('./scripts/tests/Linux-tv-app-logs.txt', 'w') as fd1, open('./scripts/tests/Linux-tv-casting-app-logs.txt', 'w') as fd2: + + with open(LINUX_TV_APP_LOGS, 'w') as fd1, open(LINUX_TV_CASTING_APP_LOGS, 'w') as fd2: # Run the Linux tv-app and write the output to file tv_app_rel_path = 'out/tv-app/chip-tv-app' tv_app_abs_path = os.path.abspath(tv_app_rel_path) subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) - time.sleep(5) + time.sleep(RUN_INTERVAL) # Run the Linux tv-casting-app and write the output to file tv_casting_app_rel_path = 'out/tv-casting-app/chip-tv-casting-app' tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) - subprocess.Popen(tv_casting_app_abs_path, stdout=fd2, stderr=subprocess.PIPE, text=True) + tv_casting_app_process = subprocess.Popen( + tv_casting_app_abs_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + + for line in tv_casting_app_process.stdout: + # Write the line to the Linux tv-casting-app log file + fd2.write(line) + + # Fail fast if "No commissioner discovered" string found + if "No commissioner discovered" in line: + print(line) + handle_discovery_failure() + + # Check if the Vendor ID, Product ID, and Device Type match the expected values + if "Vendor ID:" in line: + check_expected_value(line, VENDOR_ID, "Vendor ID") + + elif "Product ID:" in line: + check_expected_value(line, PRODUCT_ID, "Product ID") + + elif "Device Type:" in line: + check_expected_value(line, DEVICE_TYPE, "Device Type") + + if "commissioner(s) discovered" in line: + break # Wait for the processes to finish writing before attempting to read - time.sleep(15) + time.sleep(PARSE_INTERVAL) - read_linux_tv_casting_app_logs('./scripts/tests/Linux-tv-casting-app-logs.txt') + parse_linux_tv_casting_app_logs(LINUX_TV_CASTING_APP_LOGS) @click.group() @@ -174,5 +219,7 @@ def test_discovery(): if __name__ == '__main__': + # Start with a clean slate by removing any previously cached entries. + os.system('rm -f /tmp/chip_*') main() From 5a7907d3c65d79f566007daef5e366654c799913 Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Tue, 16 Apr 2024 15:07:44 -0700 Subject: [PATCH 04/12] Addressed PR comments - Changed workflow step name and command name, updated validate_value function, and added succeeding fast logic. --- .../examples-linux-tv-casting-app.yaml | 7 +- scripts/tests/run_tv_casting_test.py | 201 +++++++++--------- 2 files changed, 106 insertions(+), 102 deletions(-) diff --git a/.github/workflows/examples-linux-tv-casting-app.yaml b/.github/workflows/examples-linux-tv-casting-app.yaml index cb6b726c23db96..05ac421e623259 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,10 +63,11 @@ jobs: ./scripts/run_in_build_env.sh \ "scripts/examples/gn_build_example.sh examples/tv-casting-app/linux/ out/tv-casting-app" - - name: Test Discovery between Linux tv-casting-app and Linux tv-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 test-discovery" + "python3 ./scripts/tests/run_tv_casting_test.py test-casting" + timeout-minutes: 1 - name: Uploading Size Reports uses: ./.github/actions/upload-size-reports diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 58358250af3083..8a765b2db29504 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -24,8 +24,7 @@ LINUX_TV_APP_LOGS = './scripts/tests/Linux-tv-app-logs.txt' LINUX_TV_CASTING_APP_LOGS = './scripts/tests/Linux-tv-casting-app-logs.txt' -RUN_INTERVAL = 5 -PARSE_INTERVAL = 15 +RUN_INTERVAL = 2 VENDOR_ID = 65521 PRODUCT_ID = 32769 @@ -34,6 +33,11 @@ # Dump the logs to the console in the case of an error. def dump_logs_to_console(log_file): + if log_file == LINUX_TV_CASTING_APP_LOGS: + print('Dumping Linux TV Casting App Logs to Console.') + elif log_file == LINUX_TV_APP_LOGS: + print('Dumping Linux TV App Logs to Console.') + with open(log_file, 'r') as file: logs = file.read() print(logs) @@ -50,9 +54,10 @@ def remove_log_file(log_file): # Whenever a failure is discovered, we should print 'Discovery failed!', # dump the logs, clean up the log files, exit on error. def handle_discovery_failure(): - print('Discovery failed!') + print('Discovery failed!\n') dump_logs_to_console(LINUX_TV_CASTING_APP_LOGS) + dump_logs_to_console(LINUX_TV_APP_LOGS) remove_log_file(LINUX_TV_CASTING_APP_LOGS) remove_log_file(LINUX_TV_APP_LOGS) @@ -69,100 +74,30 @@ def extract_value_from_string(line): # Check if the discovered value matches the expected value. -def check_expected_value(line, expected_value, value_name): +# Returns False if the value does not match, True otherwise. +def validate_value(expected_value, line, value_name): # Extract the integer value from the string value = extract_value_from_string(line) # If the discovered value does not match the expected value, - # print the error and handle the discovery failure. + # print the error and return False. if value != expected_value: print(f'{value_name} does not match the expected value!') - print(f'Discovered {value_name}: {value}') print(f'Expected {value_name}: {expected_value}') + line = line.rstrip('\n') + print(line) - handle_discovery_failure() - - -# Read the logs from the Linux-tv-casting-app-logs.txt file. -# The discovered commissioner(s) will be stored in a list along with their -# vendor ID, product ID, and device type. -def parse_linux_tv_casting_app_logs(log_file): - - with open(log_file, 'r') as file: - lines = file.readlines() - - discovered_commissioners = [] - - print('Reading from Linux-tv-casting-app-logs.txt') - - # Read through the Linux-tv-casting-app-logs.txt line by line - for i, line in enumerate(lines): - - # If commissioner(s) are discovered, then the discovery process was successful. - if "commissioner(s) discovered" in line: - print(line) - print('Discovery success!') - - remove_log_file(LINUX_TV_CASTING_APP_LOGS) - remove_log_file(LINUX_TV_APP_LOGS) + return False - break - - # Look for "Discovered Commissioner" - if "Discovered Commissioner" in line: - print(line) - - # Extract the relevant part of the string - commissioner = line.split("Discovered Commissioner")[-1].strip() - commissioner = commissioner.replace('\x1b[0m', '') - - # Initialize variables for Vendor ID, Product ID, and Device Type - vendor_id = None - product_id = None - device_type = None - - # Iterate through the subsequent lines to find the strings of interest - for next_line in lines[i+1:]: - - if "Vendor ID:" in next_line: - print(next_line) - vendor_id = extract_value_from_string(next_line) - - elif "Product ID:" in next_line: - print(next_line) - product_id = extract_value_from_string(next_line) - - elif "Device Type:" in next_line: - print(next_line) - device_type = extract_value_from_string(next_line) - - elif "commissioner(s) discovered" in next_line: - break - - # If the next line starts with "Discovered Commissioner", break the loop - if "Discovered Commissioner" in next_line: - break - - # Append the extracted information to the devices list - discovered_commissioners.append({ - "discovered_commissioner": commissioner, - "vendor_id": vendor_id, - "product_id": product_id, - "device_type": device_type - }) - - # If the list of discovered commissioners is empty and we didn't find the "No commissioner discovered" string, - # then something went wrong. Exit on error. - if len(discovered_commissioners) == 0: - print('No commissioner(s) discovered! The list of discovered commissioner(s) is empty!') - handle_discovery_failure() + # Return True if the value matches the expected value + return True # Test if the Linux tv-casting-app is able to discover the Linux tv-app. # The Linux tv-casting-app and the tv-app will be run in separate processes. # Their corresponding output will be written to their respective log files. -# The log file of the tv-casting-app will be parsed for strings of interest -# which will be printed to the console. +# The output of the tv-casting-app will be parsed in realtime for strings of +# interest which will be printed to the console. def test_discovery_fn(): with open(LINUX_TV_APP_LOGS, 'w') as fd1, open(LINUX_TV_CASTING_APP_LOGS, 'w') as fd2: @@ -170,42 +105,110 @@ def test_discovery_fn(): # Run the Linux tv-app and write the output to file tv_app_rel_path = 'out/tv-app/chip-tv-app' tv_app_abs_path = os.path.abspath(tv_app_rel_path) - subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) + tv_app_process = subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) time.sleep(RUN_INTERVAL) - # Run the Linux tv-casting-app and write the output to file + # Run the Linux tv-casting-app tv_casting_app_rel_path = 'out/tv-casting-app/chip-tv-casting-app' tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) tv_casting_app_process = subprocess.Popen( tv_casting_app_abs_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + # Initialize variables + continue_parsing = False + valid_discovered_commissioner_str = '' + + # Read the output as we get it from the tv-casting-app process for line in tv_casting_app_process.stdout: # Write the line to the Linux tv-casting-app log file fd2.write(line) # Fail fast if "No commissioner discovered" string found if "No commissioner discovered" in line: + line = line.rstrip('\n') print(line) handle_discovery_failure() - # Check if the Vendor ID, Product ID, and Device Type match the expected values - if "Vendor ID:" in line: - check_expected_value(line, VENDOR_ID, "Vendor ID") + # Look for 'Discovered Commissioner' + if "Discovered Commissioner" in line: + line = line.rstrip('\n') + valid_discovered_commissioner_str = line + + # Continue parsing the content that belongs to the "Discovered Commissioner" + continue_parsing = True + + # Initialize variables to store the information of interest + valid_vendor_id = False + valid_product_id = False + valid_device_type = False + + valid_vendor_id_str = '' + valid_product_id_str = '' + valid_device_type_str = '' + + if continue_parsing: + + # Check if the Vendor ID, Product ID, and Device Type match the expected values + if "Vendor ID:" in line: + + # If the value of the Vendor ID does not match the expected value, then + # handle the discovery failure. + valid_vendor_id = validate_value(VENDOR_ID, line, "Vendor ID") - elif "Product ID:" in line: - check_expected_value(line, PRODUCT_ID, "Product ID") + if not valid_vendor_id: + handle_discovery_failure() + else: + line = line.rstrip('\n') + valid_vendor_id_str = line - elif "Device Type:" in line: - check_expected_value(line, DEVICE_TYPE, "Device Type") + elif "Product ID:" in line: - if "commissioner(s) discovered" in line: - break + # If the value of Product ID does not match the expected value, then + # handle the discovery failure. + valid_product_id = validate_value(PRODUCT_ID, line, "Product ID") + + if not valid_product_id: + handle_discovery_failure() + else: + line = line.rstrip('\n') + valid_product_id_str = line + + elif "Device Type:" in line: + + # If the value of Device Type does not match the expected value, then + # handle the discovery failure. + valid_device_type = validate_value(DEVICE_TYPE, line, "Device Type") + + if not valid_device_type: + handle_discovery_failure() + else: + line = line.rstrip('\n') + valid_device_type_str = line + + # At this point, all values of interest are valid, so we stop parsing. + continue_parsing = False + + # We only print the discovered commissioner that has valid vendor id, product id, + # and device type. Remove the log files once done. + if valid_vendor_id and valid_product_id and valid_device_type: + print(valid_discovered_commissioner_str) + print(valid_vendor_id_str) + print(valid_product_id_str) + print(valid_device_type_str) + print('Discovery success!') + + remove_log_file(LINUX_TV_CASTING_APP_LOGS) + remove_log_file(LINUX_TV_APP_LOGS) + + break - # Wait for the processes to finish writing before attempting to read - time.sleep(PARSE_INTERVAL) + # Tear down the processes. + tv_app_process.terminate() + tv_app_process.wait() - parse_linux_tv_casting_app_logs(LINUX_TV_CASTING_APP_LOGS) + tv_casting_app_process.terminate() + tv_casting_app_process.wait() @click.group() @@ -213,8 +216,8 @@ def main(): pass -@main.command('test-discovery', help='Test if the Linux tv-casting-app is able to discover the Linux tv-app.') -def test_discovery(): +@main.command('test-casting', help='Test casting from Linux tv-casting-app to Linux tv-app.') +def test_casting(): test_discovery_fn() From b574a55fd559965dfa18627811150f8bf5c7cdba Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Wed, 17 Apr 2024 09:45:37 -0700 Subject: [PATCH 05/12] Addressed PR comments - Created def tear_down and utilized logging module instead of print statements. --- scripts/tests/run_tv_casting_test.py | 57 ++++++++++++++++++---------- 1 file changed, 37 insertions(+), 20 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 8a765b2db29504..4216905d7649c2 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -14,6 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import logging import os import subprocess import sys @@ -21,6 +22,9 @@ import click +# Configure logging format. +logging.basicConfig(level=logging.DEBUG, format='%(levelname)s - %(message)s') + LINUX_TV_APP_LOGS = './scripts/tests/Linux-tv-app-logs.txt' LINUX_TV_CASTING_APP_LOGS = './scripts/tests/Linux-tv-casting-app-logs.txt' @@ -33,28 +37,31 @@ # Dump the logs to the console in the case of an error. def dump_logs_to_console(log_file): + if log_file == LINUX_TV_CASTING_APP_LOGS: - print('Dumping Linux TV Casting App Logs to Console.') + logging.debug('Dumping Linux TV Casting App Logs to Console.') elif log_file == LINUX_TV_APP_LOGS: - print('Dumping Linux TV App Logs to Console.') + logging.debug('Dumping Linux TV App Logs to Console.') with open(log_file, 'r') as file: logs = file.read() - print(logs) + logging.debug(logs) # Remove the log files once the script is done running. def remove_log_file(log_file): + if os.path.exists(log_file): os.remove(log_file) else: - print("The file does not exist.") + logging.error('The file does not exist.') # Whenever a failure is discovered, we should print 'Discovery failed!', # dump the logs, clean up the log files, exit on error. def handle_discovery_failure(): - print('Discovery failed!\n') + + logging.error('Discovery failed!\n') dump_logs_to_console(LINUX_TV_CASTING_APP_LOGS) dump_logs_to_console(LINUX_TV_APP_LOGS) @@ -67,7 +74,8 @@ def handle_discovery_failure(): # Helper function to extract the integer value from a string. def extract_value_from_string(line): - value = line.split(":")[-1].strip().replace('\x1b[0m', '') + + value = line.split(':')[-1].strip().replace('\x1b[0m', '') value = int(value) return value @@ -76,16 +84,17 @@ def extract_value_from_string(line): # Check if the discovered value matches the expected value. # Returns False if the value does not match, True otherwise. def validate_value(expected_value, line, value_name): + # Extract the integer value from the string value = extract_value_from_string(line) # If the discovered value does not match the expected value, # print the error and return False. if value != expected_value: - print(f'{value_name} does not match the expected value!') - print(f'Expected {value_name}: {expected_value}') + logging.error(f'{value_name} does not match the expected value!') + logging.error(f'Expected {value_name}: {expected_value}') line = line.rstrip('\n') - print(line) + logging.error(line) return False @@ -93,6 +102,14 @@ def validate_value(expected_value, line, value_name): return True +# Tear down the processes once done running. +def tear_down(processes): + + for process in processes: + process.terminate() + process.wait() + + # Test if the Linux tv-casting-app is able to discover the Linux tv-app. # The Linux tv-casting-app and the tv-app will be run in separate processes. # Their corresponding output will be written to their respective log files. @@ -121,17 +138,20 @@ def test_discovery_fn(): # Read the output as we get it from the tv-casting-app process for line in tv_casting_app_process.stdout: + # Write the line to the Linux tv-casting-app log file fd2.write(line) # Fail fast if "No commissioner discovered" string found if "No commissioner discovered" in line: + line = line.rstrip('\n') - print(line) + logging.error(line) handle_discovery_failure() # Look for 'Discovered Commissioner' if "Discovered Commissioner" in line: + line = line.rstrip('\n') valid_discovered_commissioner_str = line @@ -192,11 +212,12 @@ def test_discovery_fn(): # We only print the discovered commissioner that has valid vendor id, product id, # and device type. Remove the log files once done. if valid_vendor_id and valid_product_id and valid_device_type: - print(valid_discovered_commissioner_str) - print(valid_vendor_id_str) - print(valid_product_id_str) - print(valid_device_type_str) - print('Discovery success!') + + logging.info(valid_discovered_commissioner_str) + logging.info(valid_vendor_id_str) + logging.info(valid_product_id_str) + logging.info(valid_device_type_str) + logging.info('Discovery success!') remove_log_file(LINUX_TV_CASTING_APP_LOGS) remove_log_file(LINUX_TV_APP_LOGS) @@ -204,11 +225,7 @@ def test_discovery_fn(): break # Tear down the processes. - tv_app_process.terminate() - tv_app_process.wait() - - tv_casting_app_process.terminate() - tv_casting_app_process.wait() + tear_down([tv_app_process, tv_casting_app_process]) @click.group() From c6b5082361b09f98d650dc51cf2d58757a4c060f Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Wed, 17 Apr 2024 11:05:48 -0700 Subject: [PATCH 06/12] Commit to trigger CI checks to re-run since one workflow failed. Added '.' at the end of code comments. --- scripts/tests/run_tv_casting_test.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 4216905d7649c2..2889587b8336e8 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -85,7 +85,7 @@ def extract_value_from_string(line): # Returns False if the value does not match, True otherwise. def validate_value(expected_value, line, value_name): - # Extract the integer value from the string + # Extract the integer value from the string. value = extract_value_from_string(line) # If the discovered value does not match the expected value, @@ -98,7 +98,7 @@ def validate_value(expected_value, line, value_name): return False - # Return True if the value matches the expected value + # Return True if the value matches the expected value. return True @@ -119,46 +119,46 @@ def test_discovery_fn(): with open(LINUX_TV_APP_LOGS, 'w') as fd1, open(LINUX_TV_CASTING_APP_LOGS, 'w') as fd2: - # Run the Linux tv-app and write the output to file + # Run the Linux tv-app and write the output to file. tv_app_rel_path = 'out/tv-app/chip-tv-app' tv_app_abs_path = os.path.abspath(tv_app_rel_path) tv_app_process = subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) time.sleep(RUN_INTERVAL) - # Run the Linux tv-casting-app + # Run the Linux tv-casting-app. tv_casting_app_rel_path = 'out/tv-casting-app/chip-tv-casting-app' tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) tv_casting_app_process = subprocess.Popen( tv_casting_app_abs_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - # Initialize variables + # Initialize variables. continue_parsing = False valid_discovered_commissioner_str = '' - # Read the output as we get it from the tv-casting-app process + # Read the output as we get it from the tv-casting-app process. for line in tv_casting_app_process.stdout: - # Write the line to the Linux tv-casting-app log file + # Write the line to the Linux tv-casting-app log file. fd2.write(line) - # Fail fast if "No commissioner discovered" string found + # Fail fast if "No commissioner discovered" string found. if "No commissioner discovered" in line: line = line.rstrip('\n') logging.error(line) handle_discovery_failure() - # Look for 'Discovered Commissioner' + # Look for 'Discovered Commissioner'. if "Discovered Commissioner" in line: line = line.rstrip('\n') valid_discovered_commissioner_str = line - # Continue parsing the content that belongs to the "Discovered Commissioner" + # Continue parsing the content that belongs to the "Discovered Commissioner". continue_parsing = True - # Initialize variables to store the information of interest + # Initialize variables to store the information of interest. valid_vendor_id = False valid_product_id = False valid_device_type = False @@ -169,7 +169,7 @@ def test_discovery_fn(): if continue_parsing: - # Check if the Vendor ID, Product ID, and Device Type match the expected values + # Check if the Vendor ID, Product ID, and Device Type match the expected values. if "Vendor ID:" in line: # If the value of the Vendor ID does not match the expected value, then From ea0c55b55c79885f89d12e0030304690bbd9dbfd Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Tue, 23 Apr 2024 09:45:16 -0700 Subject: [PATCH 07/12] Addressed PR comments from @andy31415: Used context managers for log file handling, and for managing subprocesses. Added clarifying code comments, utilized temporary directory to store log files, added path arguments to the test program, added command to not allow buffering for output streams, and changed from sleep() to tv-app start up time-out condition. --- scripts/tests/run_tv_casting_test.py | 422 +++++++++++++++------------ 1 file changed, 240 insertions(+), 182 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 2889587b8336e8..2ca19cf851323f 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -18,228 +18,286 @@ import os import subprocess import sys +import tempfile import time +from typing import List, Optional import click # Configure logging format. -logging.basicConfig(level=logging.DEBUG, format='%(levelname)s - %(message)s') +logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s') -LINUX_TV_APP_LOGS = './scripts/tests/Linux-tv-app-logs.txt' -LINUX_TV_CASTING_APP_LOGS = './scripts/tests/Linux-tv-casting-app-logs.txt' +# The maximum amount of time to wait for the Linux tv-app to start before timeout. +TV_APP_MAX_START_WAIT_SEC = 2 -RUN_INTERVAL = 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 = 65521 PRODUCT_ID = 32769 DEVICE_TYPE = 35 -# Dump the logs to the console in the case of an error. -def dump_logs_to_console(log_file): - - if log_file == LINUX_TV_CASTING_APP_LOGS: - logging.debug('Dumping Linux TV Casting App Logs to Console.') - elif log_file == LINUX_TV_APP_LOGS: - logging.debug('Dumping Linux TV App Logs to Console.') - - with open(log_file, 'r') as file: - logs = file.read() - logging.debug(logs) - - -# Remove the log files once the script is done running. -def remove_log_file(log_file): - - if os.path.exists(log_file): - os.remove(log_file) - else: - logging.error('The file does not exist.') - - -# Whenever a failure is discovered, we should print 'Discovery failed!', -# dump the logs, clean up the log files, exit on error. -def handle_discovery_failure(): - - logging.error('Discovery failed!\n') - - dump_logs_to_console(LINUX_TV_CASTING_APP_LOGS) - dump_logs_to_console(LINUX_TV_APP_LOGS) - - remove_log_file(LINUX_TV_CASTING_APP_LOGS) - remove_log_file(LINUX_TV_APP_LOGS) +class LogFileManager: + """ + A context manager for managing log files. + + This class provides a context manager for safely opening and closing log files. + It ensures that log files are properly managed, allowing reading, writing, or + both, depending on the specified mode. + """ + + # Initialize LogFileManager. + # log_file_path (str): The path to the log file. + # mode (str): The file mode for opening the log file (default is 'w+' for read/write mode). + def __init__(self, log_file_path: str, mode: str = 'w+'): + self.log_file_path = log_file_path + self.mode = mode + + # Enter the context manager to open and return the log file. + def __enter__(self): + try: + self.file = open(self.log_file_path, self.mode) + except FileNotFoundError: + # Handle file not found error + raise FileNotFoundError(f"Log file '{self.log_file_path}' not found.") + except IOError: + # Handle IO error + raise IOError(f"Error opening log file '{self.log_file_path}'.") + return self.file + + # Exit the context manager, closing and removing the log file created. + # exception_type: The type of exception that occurred, if any. + # exception_value: The value of the exception, if any. + # traceback: The traceback of the exception. + def __exit__(self, exception_type, exception_value, traceback): + self.file.close() + + if os.path.exists(self.log_file_path): + os.remove(self.log_file_path) + + +class ProcessManager: + """ + A context manager for managing subprocesses. + + This class provides a context manager for safely starting and stopping a subprocess. + """ + + # Initialize ProcessManager. + # command (list): The command to execute as a subprocess. + # stdout (file): File-like object to which the subprocess's standard output will be redirected. + # stderr (file): File-like object to which the subprocess's standard error will be redirected. + def __init__(self, command: List[str], stdout, stderr): + self.command = command + self.stdout = stdout + self.stderr = stderr + + # Enter the context manager to start the subprocess and return it. + def __enter__(self): + self.process = subprocess.Popen(self.command, stdout=self.stdout, stderr=self.stderr, text=True) + return self.process + + # Exit the context manager, terminating the subprocess. + # exception_type: The type of exception that occurred, if any. + # exception_value: The value of the exception, if any. + # traceback: The traceback of the exception. + def __exit__(self, exception_type, exception_value, traceback): + self.process.terminate() + self.process.wait() + + +# Dump the contents of a log file to the console. +# log_file_path: The path to the log file. +def dump_logs_to_console(log_file_path: str): + print('\nDumping logs from: ', log_file_path) + + with LogFileManager(log_file_path, 'r') as file: + for line in file: + print(line.rstrip()) + + +# Log 'Discovery failed!' as an error, dump the contents of the log files +# to the console, exit on error. +# log_file_paths: A list of paths to the log files, i.e. the path to the +# Linux tv-casting-app logs and the tv-app logs. +def handle_discovery_failure(log_file_paths: List[str]): + logging.error('Discovery failed!') + + for log_file_path in log_file_paths: + dump_logs_to_console(log_file_path) sys.exit(1) -# Helper function to extract the integer value from a string. -def extract_value_from_string(line): - +# Extract and return an integer value from a given output string. +# line: The string containing the integer value. +# +# 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. +def extract_value_from_string(line: str) -> int: value = line.split(':')[-1].strip().replace('\x1b[0m', '') value = int(value) return value -# Check if the discovered value matches the expected value. -# Returns False if the value does not match, True otherwise. -def validate_value(expected_value, line, value_name): - +# Validate if the discovered value matches the expected value. +# expected_value: The expected integer value, i.e. any of the VENDOR_ID, +# PRODUCT_ID, or DEVICE_TYPE constants. +# line: The string containing the value of interest that will be compared +# to the expected value. +# value_name: The name of the discovered value, i.e. 'Vendor ID', 'Product ID', +# or 'Device Type'. +# Return False if the discovered value does not match, True otherwise. +def validate_value(expected_value: int, line: str, value_name: str) -> bool: # Extract the integer value from the string. value = extract_value_from_string(line) # If the discovered value does not match the expected value, - # print the error and return False. + # log the error and return False. if value != expected_value: logging.error(f'{value_name} does not match the expected value!') logging.error(f'Expected {value_name}: {expected_value}') - line = line.rstrip('\n') - logging.error(line) - + logging.error(line.rstrip('\n')) return False # Return True if the value matches the expected value. return True -# Tear down the processes once done running. -def tear_down(processes): - - for process in processes: - process.terminate() - process.wait() - - -# Test if the Linux tv-casting-app is able to discover the Linux tv-app. -# The Linux tv-casting-app and the tv-app will be run in separate processes. -# Their corresponding output will be written to their respective log files. -# The output of the tv-casting-app will be parsed in realtime for strings of -# interest which will be printed to the console. -def test_discovery_fn(): - - with open(LINUX_TV_APP_LOGS, 'w') as fd1, open(LINUX_TV_CASTING_APP_LOGS, 'w') as fd2: - - # Run the Linux tv-app and write the output to file. - tv_app_rel_path = 'out/tv-app/chip-tv-app' - tv_app_abs_path = os.path.abspath(tv_app_rel_path) - tv_app_process = subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True) - - time.sleep(RUN_INTERVAL) - - # Run the Linux tv-casting-app. - tv_casting_app_rel_path = 'out/tv-casting-app/chip-tv-casting-app' - tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) - tv_casting_app_process = subprocess.Popen( - tv_casting_app_abs_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - # Initialize variables. - continue_parsing = False - valid_discovered_commissioner_str = '' - - # Read the output as we get it from the tv-casting-app process. - for line in tv_casting_app_process.stdout: - - # Write the line to the Linux tv-casting-app log file. - fd2.write(line) - - # Fail fast if "No commissioner discovered" string found. - if "No commissioner discovered" in line: - - line = line.rstrip('\n') - logging.error(line) - handle_discovery_failure() - - # Look for 'Discovered Commissioner'. - if "Discovered Commissioner" in line: - - line = line.rstrip('\n') - valid_discovered_commissioner_str = line - - # Continue parsing the content that belongs to the "Discovered Commissioner". - continue_parsing = True - - # Initialize variables to store the information of interest. - valid_vendor_id = False - valid_product_id = False - valid_device_type = False - - valid_vendor_id_str = '' - valid_product_id_str = '' - valid_device_type_str = '' - - if continue_parsing: - - # Check if the Vendor ID, Product ID, and Device Type match the expected values. - if "Vendor ID:" in line: - - # If the value of the Vendor ID does not match the expected value, then - # handle the discovery failure. - valid_vendor_id = validate_value(VENDOR_ID, line, "Vendor ID") - - if not valid_vendor_id: - handle_discovery_failure() - else: - line = line.rstrip('\n') - valid_vendor_id_str = line - - elif "Product ID:" in line: - - # If the value of Product ID does not match the expected value, then - # handle the discovery failure. - valid_product_id = validate_value(PRODUCT_ID, line, "Product ID") - - if not valid_product_id: - handle_discovery_failure() - else: - line = line.rstrip('\n') - valid_product_id_str = line - - elif "Device Type:" in line: - - # If the value of Device Type does not match the expected value, then - # handle the discovery failure. - valid_device_type = validate_value(DEVICE_TYPE, line, "Device Type") - - if not valid_device_type: - handle_discovery_failure() - else: - line = line.rstrip('\n') - valid_device_type_str = line - - # At this point, all values of interest are valid, so we stop parsing. +# Test if the Linux tv-casting-app is able to discover the Linux tv-app. Both will +# run separately as subprocesses, with their outputs written to respective log files. +# 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 +@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): + + # 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) + + # Open and write to the log file for the Linux tv-app. + with LogFileManager(linux_tv_app_log_path, 'w') as linux_tv_app_log_file: + tv_app_abs_path = os.path.abspath(tv_app_rel_path) + + if sys.platform == 'darwin': + # Try to avoid any stdout buffering in our tests. + cmd = ['stdbuf', '-o0', '-i0'] + + # Run the Linux tv-app subprocess. + with ProcessManager(cmd + [tv_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_app_process: + start_wait_time = time.time() + + # Loop until either the subprocess starts successfully or timeout occurs. + 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.") + handle_discovery_failure([linux_tv_app_log_path]) + + # Read one line of output at a time. + tv_app_output_line = tv_app_process.stdout.readline() + + # Write the output to the file. + 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!') + + # If the string is found, then break out of the loop and go ahead with running the Linux tv-casting-app. + break + + # Open and write to the log file for the Linux tv-casting-app. + with LogFileManager(linux_tv_casting_app_log_path, 'w') as linux_tv_casting_app_log_file: + tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) + + # Run the Linux tv-casting-app subprocess. + with ProcessManager(cmd + [tv_casting_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_casting_app_process: + # Initialize variables. continue_parsing = False - - # We only print the discovered commissioner that has valid vendor id, product id, - # and device type. Remove the log files once done. - if valid_vendor_id and valid_product_id and valid_device_type: - - logging.info(valid_discovered_commissioner_str) - logging.info(valid_vendor_id_str) - logging.info(valid_product_id_str) - logging.info(valid_device_type_str) - logging.info('Discovery success!') - - remove_log_file(LINUX_TV_CASTING_APP_LOGS) - remove_log_file(LINUX_TV_APP_LOGS) - - break - - # Tear down the processes. - tear_down([tv_app_process, tv_casting_app_process]) - - -@click.group() -def main(): - pass - - -@main.command('test-casting', help='Test casting from Linux tv-casting-app to Linux tv-app.') -def test_casting(): - test_discovery_fn() + valid_discovered_commissioner = '' + + # Read the output as we receive it from the tv-casting-app subprocess. + for line in tv_casting_app_process.stdout: + # Write to the Linux tv-casting-app log file. + 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([linux_tv_app_log_path, linux_tv_casting_app_log_path]) + + # Look for 'Discovered Commissioner'. + if "Discovered Commissioner" in line: + valid_discovered_commissioner = line.rstrip('\n') + + # Continue parsing the content that belongs to the "Discovered Commissioner". + continue_parsing = True + + # Initialize variables to store the information of interest. + valid_vendor_id: Optional[str] = None + valid_product_id: Optional[str] = None + valid_device_type: Optional[str] = None + + if continue_parsing: + # Check if the Vendor ID, Product ID, and Device Type match the expected constant values. + # If they do not match, then handle the discovery failure. + if 'Vendor ID:' in line: + valid_vendor_id = validate_value(VENDOR_ID, line, 'Vendor ID') + + if not valid_vendor_id: + handle_discovery_failure([linux_tv_app_log_path, linux_tv_casting_app_log_path]) + else: + valid_vendor_id = line.rstrip('\n') + + elif 'Product ID:' in line: + valid_product_id = validate_value(PRODUCT_ID, line, 'Product ID') + + if not valid_product_id: + handle_discovery_failure([linux_tv_app_log_path, linux_tv_casting_app_log_path]) + else: + valid_product_id = line.rstrip('\n') + + elif 'Device Type:' in line: + valid_device_type = validate_value(DEVICE_TYPE, line, 'Device Type') + + if not valid_device_type: + handle_discovery_failure([linux_tv_app_log_path, linux_tv_casting_app_log_path]) + else: + valid_device_type = line.rstrip('\n') + + # At this point, all values of interest are valid, so we stop parsing. + continue_parsing = False + + # Only a discovered commissioner that has valid vendor id, product id, + # and device type will allow for 'Discovery success!'. + 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!') + return if __name__ == '__main__': + # Start with a clean slate by removing any previously cached entries. os.system('rm -f /tmp/chip_*') - main() + # Test discovery between the Linux tv-casting-app and the tv-app. + test_discovery_fn() From ef4eead2d2ed7e71e1458c2a6880a5241eecf203 Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Tue, 23 Apr 2024 10:26:05 -0700 Subject: [PATCH 08/12] Added additional comments to clarify the constants VENDOR_ID, PRODUCT_ID, and DEVICE_TYPE. Updated the run command in the workflow yaml file. --- .github/workflows/examples-linux-tv-casting-app.yaml | 2 +- scripts/tests/run_tv_casting_test.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/examples-linux-tv-casting-app.yaml b/.github/workflows/examples-linux-tv-casting-app.yaml index 05ac421e623259..679f4be57a35db 100644 --- a/.github/workflows/examples-linux-tv-casting-app.yaml +++ b/.github/workflows/examples-linux-tv-casting-app.yaml @@ -66,7 +66,7 @@ jobs: - 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 test-casting" + "python3 ./scripts/tests/run_tv_casting_test.py" timeout-minutes: 1 - name: Uploading Size Reports diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 2ca19cf851323f..3729d2f4230b1f 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -36,9 +36,9 @@ # 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 = 65521 -PRODUCT_ID = 32769 -DEVICE_TYPE = 35 +VENDOR_ID = 65521 # Test vendor id +PRODUCT_ID = 32769 # Test product id +DEVICE_TYPE = 35 # Casting video player class LogFileManager: From 05e5c616ef2b466669bc4b1965e2c8020560e04b Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Tue, 23 Apr 2024 11:57:51 -0700 Subject: [PATCH 09/12] Added fix to allow the relevant CI check to pass. --- scripts/tests/run_tv_casting_test.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 3729d2f4230b1f..0397096f5ab35e 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -190,12 +190,14 @@ def test_discovery_fn(tv_app_rel_path, tv_casting_app_rel_path): with LogFileManager(linux_tv_app_log_path, 'w') as linux_tv_app_log_file: tv_app_abs_path = os.path.abspath(tv_app_rel_path) - if sys.platform == 'darwin': - # Try to avoid any stdout buffering in our tests. - cmd = ['stdbuf', '-o0', '-i0'] + # 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'] # Run the Linux tv-app subprocess. - with ProcessManager(cmd + [tv_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_app_process: + with ProcessManager(disable_stdout_buffering_cmd + [tv_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_app_process: start_wait_time = time.time() # Loop until either the subprocess starts successfully or timeout occurs. @@ -224,7 +226,7 @@ def test_discovery_fn(tv_app_rel_path, tv_casting_app_rel_path): tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path) # Run the Linux tv-casting-app subprocess. - with ProcessManager(cmd + [tv_casting_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_casting_app_process: + with ProcessManager(disable_stdout_buffering_cmd + [tv_casting_app_abs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_casting_app_process: # Initialize variables. continue_parsing = False valid_discovered_commissioner = '' From 496d71e5093dd2e8b058f63d6a99122c2fc4f4d9 Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Tue, 23 Apr 2024 13:22:13 -0700 Subject: [PATCH 10/12] Added '.' to the end of some comments to kick-off a re-run for CI check that failed. --- scripts/tests/run_tv_casting_test.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 0397096f5ab35e..6127adffae341a 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -190,9 +190,9 @@ def test_discovery_fn(tv_app_rel_path, tv_casting_app_rel_path): with LogFileManager(linux_tv_app_log_path, 'w') as linux_tv_app_log_file: tv_app_abs_path = os.path.abspath(tv_app_rel_path) - # Configure command options to disable stdout buffering during tests + # Configure command options to disable stdout buffering during tests. disable_stdout_buffering_cmd = [] - # On Unix-like systems, use stdbuf to disable stdout buffering + # 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'] From 1ac2d356caf3822b5a960ea52c1a28b3d4a20baa Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Wed, 24 Apr 2024 16:41:26 -0700 Subject: [PATCH 11/12] Addressed @andy31415 PR comments - Updated constants to hex format, updated to follow doc-style comment format, removed redundant comments, removed LogFileManager class as temporary directory will handle file clean up, split test_discovery_fn to improve code readability. --- scripts/tests/run_tv_casting_test.py | 297 ++++++++++----------------- 1 file changed, 108 insertions(+), 189 deletions(-) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 6127adffae341a..1837c1989627a3 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -20,7 +20,7 @@ import sys import tempfile import time -from typing import List, Optional +from typing import List, Optional, TextIO, Tuple import click @@ -36,159 +36,163 @@ # 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 = 65521 # Test vendor id -PRODUCT_ID = 32769 # Test product id -DEVICE_TYPE = 35 # Casting video player - - -class LogFileManager: - """ - A context manager for managing log files. - - This class provides a context manager for safely opening and closing log files. - It ensures that log files are properly managed, allowing reading, writing, or - both, depending on the specified mode. - """ - - # Initialize LogFileManager. - # log_file_path (str): The path to the log file. - # mode (str): The file mode for opening the log file (default is 'w+' for read/write mode). - def __init__(self, log_file_path: str, mode: str = 'w+'): - self.log_file_path = log_file_path - self.mode = mode - - # Enter the context manager to open and return the log file. - def __enter__(self): - try: - self.file = open(self.log_file_path, self.mode) - except FileNotFoundError: - # Handle file not found error - raise FileNotFoundError(f"Log file '{self.log_file_path}' not found.") - except IOError: - # Handle IO error - raise IOError(f"Error opening log file '{self.log_file_path}'.") - return self.file - - # Exit the context manager, closing and removing the log file created. - # exception_type: The type of exception that occurred, if any. - # exception_value: The value of the exception, if any. - # traceback: The traceback of the exception. - def __exit__(self, exception_type, exception_value, traceback): - self.file.close() - - if os.path.exists(self.log_file_path): - os.remove(self.log_file_path) +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. + """A context manager for managing subprocesses. This class provides a context manager for safely starting and stopping a subprocess. """ - # Initialize ProcessManager. - # command (list): The command to execute as a subprocess. - # stdout (file): File-like object to which the subprocess's standard output will be redirected. - # stderr (file): File-like object to which the subprocess's standard error will be redirected. def __init__(self, command: List[str], stdout, stderr): self.command = command self.stdout = stdout self.stderr = stderr - # Enter the context manager to start the subprocess and return it. def __enter__(self): self.process = subprocess.Popen(self.command, stdout=self.stdout, stderr=self.stderr, text=True) return self.process - # Exit the context manager, terminating the subprocess. - # exception_type: The type of exception that occurred, if any. - # exception_value: The value of the exception, if any. - # traceback: The traceback of the exception. def __exit__(self, exception_type, exception_value, traceback): self.process.terminate() self.process.wait() -# Dump the contents of a log file to the console. -# log_file_path: The path to the log file. -def dump_logs_to_console(log_file_path: str): +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.""" print('\nDumping logs from: ', log_file_path) - with LogFileManager(log_file_path, 'r') as file: + with open(log_file_path, 'r') as file: for line in file: print(line.rstrip()) -# Log 'Discovery failed!' as an error, dump the contents of the log files -# to the console, exit on error. -# log_file_paths: A list of paths to the log files, i.e. the path to the -# Linux tv-casting-app logs and the tv-app logs. 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: - dump_logs_to_console(log_file_path) + 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) -# Extract and return an integer value from a given output string. -# line: The string containing the integer value. -# -# 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. 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 -# Validate if the discovered value matches the expected value. -# expected_value: The expected integer value, i.e. any of the VENDOR_ID, -# PRODUCT_ID, or DEVICE_TYPE constants. -# line: The string containing the value of interest that will be compared -# to the expected value. -# value_name: The name of the discovered value, i.e. 'Vendor ID', 'Product ID', -# or 'Device Type'. -# Return False if the discovered value does not match, True otherwise. -def validate_value(expected_value: int, line: str, value_name: str) -> bool: - # Extract the integer value from the string. +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 the discovered value does not match the expected value, - # log the error and return False. 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')) - return False + 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() - # Return True if the value matches the expected value. - return True + 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 -# Test if the Linux tv-casting-app is able to discover the Linux tv-app. Both will -# run separately as subprocesses, with their outputs written to respective log files. -# 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 @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) - # Open and write to the log file for the Linux tv-app. - with LogFileManager(linux_tv_app_log_path, 'w') as linux_tv_app_log_file: - tv_app_abs_path = os.path.abspath(tv_app_rel_path) + 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 = [] @@ -196,104 +200,19 @@ def test_discovery_fn(tv_app_rel_path, tv_casting_app_rel_path): 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: - start_wait_time = time.time() - - # Loop until either the subprocess starts successfully or timeout occurs. - 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.") - handle_discovery_failure([linux_tv_app_log_path]) - - # Read one line of output at a time. - tv_app_output_line = tv_app_process.stdout.readline() - - # Write the output to the file. - 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!') - - # If the string is found, then break out of the loop and go ahead with running the Linux tv-casting-app. - break - - # Open and write to the log file for the Linux tv-casting-app. - with LogFileManager(linux_tv_casting_app_log_path, 'w') as linux_tv_casting_app_log_file: - 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: - # Initialize variables. - continue_parsing = False - valid_discovered_commissioner = '' - - # Read the output as we receive it from the tv-casting-app subprocess. - for line in tv_casting_app_process.stdout: - # Write to the Linux tv-casting-app log file. - 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([linux_tv_app_log_path, linux_tv_casting_app_log_path]) - - # Look for 'Discovered Commissioner'. - if "Discovered Commissioner" in line: - valid_discovered_commissioner = line.rstrip('\n') - - # Continue parsing the content that belongs to the "Discovered Commissioner". - continue_parsing = True - - # Initialize variables to store the information of interest. - valid_vendor_id: Optional[str] = None - valid_product_id: Optional[str] = None - valid_device_type: Optional[str] = None - - if continue_parsing: - # Check if the Vendor ID, Product ID, and Device Type match the expected constant values. - # If they do not match, then handle the discovery failure. - if 'Vendor ID:' in line: - valid_vendor_id = validate_value(VENDOR_ID, line, 'Vendor ID') - - if not valid_vendor_id: - handle_discovery_failure([linux_tv_app_log_path, linux_tv_casting_app_log_path]) - else: - valid_vendor_id = line.rstrip('\n') - - elif 'Product ID:' in line: - valid_product_id = validate_value(PRODUCT_ID, line, 'Product ID') - - if not valid_product_id: - handle_discovery_failure([linux_tv_app_log_path, linux_tv_casting_app_log_path]) - else: - valid_product_id = line.rstrip('\n') - - elif 'Device Type:' in line: - valid_device_type = validate_value(DEVICE_TYPE, line, 'Device Type') - - if not valid_device_type: - handle_discovery_failure([linux_tv_app_log_path, linux_tv_casting_app_log_path]) - else: - valid_device_type = line.rstrip('\n') - - # At this point, all values of interest are valid, so we stop parsing. - continue_parsing = False - - # Only a discovered commissioner that has valid vendor id, product id, - # and device type will allow for 'Discovery success!'. - 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!') - return + + 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__': From e082247b72d81479f41c5ae6ce37fdedc0b7dcb8 Mon Sep 17 00:00:00 2001 From: Shao Ling Tan Date: Thu, 25 Apr 2024 10:36:51 -0700 Subject: [PATCH 12/12] Added additional code comment suggested by @andy31415 to the `dump_temporary_logs_to_console` function. --- scripts/tests/run_tv_casting_test.py | 1 + 1 file changed, 1 insertion(+) diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 1837c1989627a3..7e783102a2fbc5 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -63,6 +63,7 @@ def __exit__(self, exception_type, exception_value, traceback): 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: