diff --git a/scripts/tests/linux/tv_casting_test_sequence_utils.py b/scripts/tests/linux/tv_casting_test_sequence_utils.py new file mode 100644 index 00000000000000..e95bc434972a2c --- /dev/null +++ b/scripts/tests/linux/tv_casting_test_sequence_utils.py @@ -0,0 +1,107 @@ +#!/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. + +from enum import Enum +from typing import List, Optional + +""" +This file defines the utility classes for creating and managing test sequences to validate the casting experience between +the Linux tv-casting-app and the Linux tv-app. It includes an enumeration for the supported applications (App), a class to +represent individual steps in a test sequence (Step), and a class to represent a complete test sequence (Sequence). +Additionally, it provides helper functions to retrieve specific test sequences or all defined test sequences. +""" + + +class App(Enum): + """An enumeration of the supported applications.""" + + TV_APP = 'tv-app' + TV_CASTING_APP = 'tv-casting-app' + + +class Step: + """A class to represent a step in a test sequence for validation. + + A `Step` object contains attributes relevant to a test step where each object contains: + - `app` subprocess to parse for `output_msg` or send `input_cmd` + - `timeout_sec` specified the timeout duration for parsing the `output_msg` (optional, defaults to DEFAULT_TIMEOUT_SEC) + - `output_msg` or `input_cmd` (mutually exclusive) + + For output message blocks, define the start line, relevant lines, and the last line. If the last line contains trivial closing + characters (e.g., closing brackets, braces, or commas), include the line before it with actual content. For example: + `Step(subprocess_='tv-casting-app', output_msg=['InvokeResponseMessage =', 'exampleData', 'InteractionModelRevision =', '},'])` + + For input commands, define the command string with placeholders for variables that need to be updated. For example: + `Step(subprocess_='tv-casting-app', input_cmd='cast request 0\n')` + """ + + # The maximum default time to wait while parsing for output string(s). + DEFAULT_TIMEOUT_SEC = 10 + + def __init__( + self, + app: App, + timeout_sec: Optional[int] = DEFAULT_TIMEOUT_SEC, + output_msg: Optional[List[str]] = None, + input_cmd: Optional[str] = None, + ): + # Validate that either `output_msg` or `input_cmd` is provided, but not both. + if output_msg is not None and input_cmd is not None: + raise ValueError( + 'Step cannot contain both `output_msg` and `input_cmd`. Either `output_msg` or `input_cmd` should be provided.') + elif output_msg is None and input_cmd is None: + raise ValueError('Step must contain either `output_msg` or `input_cmd`. Both are `None`.') + + # Define either `App.TV_APP` or `App.TV_CASTING_APP` on which we need to parse for `output_msg` or send `input_cmd`. + self.app = app + + # Define the maximum time in seconds for timeout while parsing for the `output_msg`. If not provided, then we use the DEFAULT_TIMEOUT_SEC. + self.timeout_sec = timeout_sec + + # Define the `output_msg` that we need to parse for in a list format. + self.output_msg = output_msg + + # Define the `input_cmd` that we need to send to either the `App.TV_APP` or `App.TV_CASTING_APP`. + self.input_cmd = input_cmd + + +class Sequence: + """A class representing a sequence of steps for testing the casting experience between the Linux tv-casting-app and the tv-app. + + A Sequence object needs to be defined with an appropriate test sequence `name` along with its list of `Step` objects that will + be used for validating the casting experience. + """ + + def __init__(self, name: str, steps: List[Step]): + self.name = name + self.steps = steps + + @staticmethod + def get_test_sequence_by_name(test_sequences: List['Sequence'], test_sequence_name: str) -> Optional['Sequence']: + """Retrieve a test sequence from a list of sequences by its name.""" + + for sequence in test_sequences: + if sequence.name == test_sequence_name: + return sequence + return None + + @staticmethod + def get_test_sequences() -> List['Sequence']: + """Retrieve all the test sequences to validate the casting experience between the Linux tv-casting-app and the Linux tv-app.""" + + from linux.tv_casting_test_sequences import test_sequences + + return test_sequences diff --git a/scripts/tests/linux/tv_casting_test_sequences.py b/scripts/tests/linux/tv_casting_test_sequences.py new file mode 100644 index 00000000000000..9b65e64c9bb575 --- /dev/null +++ b/scripts/tests/linux/tv_casting_test_sequences.py @@ -0,0 +1,152 @@ +#!/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. + +from linux.tv_casting_test_sequence_utils import App, Sequence, Step + +""" +In this file, we define the test sequences with the relevant steps that will be used in the `scripts/tests/run_tv_casting_test.py` +for validating the casting experience between the Linux tv-casting-app and the Linux tv-app. + +At the beginning of each test sequence we need to indicate the start up of the tv-app using the `START_APP` string as the `input_cmd` +followed by the same for the tv-casting-app. On the other hand, at the end of each test sequence we need to ensure that each app will +be stopped by providing the `STOP_APP` string as the `input_cmd`. As noted in the example below of `example_test_sequence`, the first +four steps pertain to starting the apps while the last two are for signaling stopping the apps. + +Note: `START_APP` and `STOP_APP` are reserved for signaling the starting and stopping of apps. + +Example: + test_sequences = [ + Sequence( + name='example_test_sequence', + step=[ + # Signal to start the tv-app. + Step(app=App.TV_APP, input_cmd=START_APP), + + # Validate that the tv-app is up and running. + Step(app=App.TV_APP, timeout_sec=APP_MAX_START_WAIT_SEC, output_msg=['Started commissioner']), + + # Signal to start the tv-casting-app. + Step(app=App.TV_CASTING_APP, input_cmd=START_APP), + + # Validate that the server is properly initialized in the tv-casting-app output. + Step(app=App.TV_CASTING_APP, timeout_sec=APP_MAX_START_WAIT_SEC, output_msg=['Server initialization complete']), + + # Additional steps for testing the casting experience. + + # Signal to stop the tv-casting-app as we finished validation. + Step(app=App.TV_CASTING_APP, input_cmd=STOP_APP), + + # Signal to stop the tv-app as we finished validation. + Step(app=App.TV_APP, input_cmd=STOP_APP) + ] + ) + ] +""" + +# Signal to start the app. +START_APP = 'START' + +# Signal to stop the app. +STOP_APP = 'STOP' + +# The maximum amount of time to wait for the Linux tv-app or Linux tv-casting-app to start before timeout. +APP_MAX_START_WAIT_SEC = 2 + +# 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 + +TEST_TV_CASTING_APP_DEVICE_NAME = 'Test TV casting app' # Test device name for identifying the tv-casting-app + +# Values to verify the subscription state against from the `ReportDataMessage` in the Linux tv-casting-app output. +CLUSTER_MEDIA_PLAYBACK = '0x506' # Application Cluster Spec 6.10.3 Cluster ID: Media Playback +ATTRIBUTE_CURRENT_PLAYBACK_STATE = '0x0000_0000' # Application Cluster Spec 6.10.6 Attribute ID: Current State of Playback + +test_sequences = [ + Sequence( + name='commissionee_generated_passcode_test', + steps=[ + # Signal to start the tv-app. + Step(app=App.TV_APP, input_cmd=START_APP), + + # Validate that the tv-app is up and running. + Step(app=App.TV_APP, timeout_sec=APP_MAX_START_WAIT_SEC, output_msg=['Started commissioner']), + + # Signal to start the tv-casting-app. + Step(app=App.TV_CASTING_APP, input_cmd=START_APP), + + # Validate that the server is properly initialized in the tv-casting-app output. + Step(app=App.TV_CASTING_APP, timeout_sec=APP_MAX_START_WAIT_SEC, output_msg=['Server initialization complete']), + + # Validate that there is a valid discovered commissioner with {VENDOR_ID}, {PRODUCT_ID}, and {DEVICE_TYPE_CASTING_VIDEO_PLAYER} in the tv-casting-app output. + Step(app=App.TV_CASTING_APP, output_msg=['Discovered Commissioner #0', f'Vendor ID: {VENDOR_ID}', f'Product ID: {PRODUCT_ID}', + f'Device Type: {DEVICE_TYPE_CASTING_VIDEO_PLAYER}', 'Supports Commissioner Generated Passcode: true']), + + # Validate that we are ready to send `cast request` command to the tv-casting-app subprocess. + Step(app=App.TV_CASTING_APP, output_msg=['Example: cast request 0']), + + # Send `cast request {valid_discovered_commissioner_number}\n` command to the tv-casting-app subprocess. + Step(app=App.TV_CASTING_APP, input_cmd='cast request 0\n'), + + # Validate that the `Identification Declaration` message block in the tv-casting-app output has the expected values for `device Name`, `vendor id`, and `product id`. + Step(app=App.TV_CASTING_APP, output_msg=['Identification Declaration Start', f'device Name: {TEST_TV_CASTING_APP_DEVICE_NAME}', + f'vendor id: {VENDOR_ID}', f'product id: {PRODUCT_ID}', 'Identification Declaration End']), + + # Validate that the `Identification Declaration` message block in the tv-app output has the expected values for `device Name`, `vendor id`, and `product id`. + Step(app=App.TV_APP, output_msg=['Identification Declaration Start', f'device Name: {TEST_TV_CASTING_APP_DEVICE_NAME}', + f'vendor id: {VENDOR_ID}', f'product id: {PRODUCT_ID}', 'Identification Declaration End']), + + # Validate that we received the cast request from the tv-casting-app on the tv-app output. + Step(app=App.TV_APP, + output_msg=['PROMPT USER: Test TV casting app is requesting permission to cast to this TV, approve?']), + + # Validate that we received the instructions on the tv-app output for sending the `controller ux ok` command. + Step(app=App.TV_APP, output_msg=['Via Shell Enter: controller ux ok|cancel']), + + # Send `controller ux ok` command to the tv-app subprocess. + Step(app=App.TV_APP, input_cmd='controller ux ok\n'), + + # Validate that pairing succeeded between the tv-casting-app and the tv-app. + Step(app=App.TV_APP, output_msg=['Secure Pairing Success']), + + # Validate that commissioning succeeded in the tv-casting-app output. + Step(app=App.TV_CASTING_APP, output_msg=['Commissioning completed successfully']), + + # Validate that commissioning succeeded in the tv-app output. + Step(app=App.TV_APP, output_msg=['------PROMPT USER: commissioning success']), + + # Validate the subscription state by looking at the `Cluster` and `Attribute` values in the `ReportDataMessage` block in the tv-casting-app output. + Step(app=App.TV_CASTING_APP, output_msg=[ + 'ReportDataMessage =', f'Cluster = {CLUSTER_MEDIA_PLAYBACK}', f'Attribute = {ATTRIBUTE_CURRENT_PLAYBACK_STATE}', 'InteractionModelRevision =', '}']), + + # Validate the LaunchURL in the tv-app output. + Step(app=App.TV_APP, + output_msg=['ContentLauncherManager::HandleLaunchUrl TEST CASE ContentURL=https://www.test.com/videoid DisplayString=Test video']), + + # Validate the LaunchURL in the tv-casting-app output. + Step(app=App.TV_CASTING_APP, output_msg=['InvokeResponseMessage =', + 'exampleData', 'InteractionModelRevision =', '},']), + + # Signal to stop the tv-casting-app as we finished validation. + Step(app=App.TV_CASTING_APP, input_cmd=STOP_APP), + + # Signal to stop the tv-app as we finished validation. + Step(app=App.TV_APP, input_cmd=STOP_APP) + ] + ) +] diff --git a/scripts/tests/run_tv_casting_test.py b/scripts/tests/run_tv_casting_test.py index 074d4592d59afe..45cc6171bc3dca 100644 --- a/scripts/tests/run_tv_casting_test.py +++ b/scripts/tests/run_tv_casting_test.py @@ -16,44 +16,31 @@ import logging import os -import re +import signal import subprocess import sys import tempfile import time -from typing import List, Optional, TextIO, Tuple, Union +from typing import List, TextIO, Tuple import click +from linux.tv_casting_test_sequence_utils import App, Sequence, Step +from linux.tv_casting_test_sequences import START_APP, STOP_APP -# 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 +""" +This script can be used to validate the casting experience between the Linux tv-casting-app and the Linux tv-app. -# The maximum amount of time to commission the Linux tv-casting-app and the tv-app before timeout. -COMMISSIONING_STAGE_MAX_WAIT_SEC = 15 +It runs a series of test sequences that check for expected output lines from the tv-casting-app and the tv-app in +a deterministic order. If these lines are not found, it indicates an issue with the casting experience. +""" -# The maximum amount of time to test that the launchURL is sent from the Linux tv-casting-app and received on the tv-app before timeout. -TEST_LAUNCHURL_MAX_WAIT_SEC = 10 - -# The maximum amount of time to verify the subscription state in the Linux tv-casting-app output before timeout. -VERIFY_SUBSCRIPTION_STATE_MAX_WAIT_SEC = 10 +# Configure logging format. +logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s') # 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 - -# Values to verify the subscription state against from the `ReportDataMessage` in the Linux tv-casting-app output. -CLUSTER_MEDIA_PLAYBACK = '0x506' # Application Cluster Spec 6.10.3 Cluster ID: Media Playback -ATTRIBUTE_CURRENT_PLAYBACK_STATE = '0x0000_0000' # Application Cluster Spec 6.10.6 Attribute ID: Current State of Playback - class ProcessManager: """A context manager for managing subprocesses. @@ -76,26 +63,6 @@ def __exit__(self, exception_type, exception_value, traceback): self.process.wait() -class LogValueExtractor: - """A utility class for extracting values from log lines. - - This class provides a centralized way to extract values from log lines and manage the error handling and logging process. - """ - - def __init__(self, casting_state: str, log_paths: List[str]): - self.casting_state = casting_state - self.log_paths = log_paths - - def extract_from(self, line: str, value_name: str): - if value_name in line: - try: - return extract_value_from_string(line, value_name, self.casting_state, self.log_paths) - except ValueError: - logging.error(f'Failed to extract `{value_name}` value from line: {line}') - handle_casting_failure(self.casting_state, self.log_paths) - return None - - 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.""" @@ -106,508 +73,200 @@ def dump_temporary_logs_to_console(log_file_path: str): print(line.rstrip()) -def handle_casting_failure(casting_state: str, log_file_paths: List[str]): - """Log '{casting_state} failed!' as error, dump log files to console, exit on error.""" - logging.error(f'{casting_state} failed!') +def handle_casting_failure(test_sequence_name: str, log_file_paths: List[str]): + """Log failure of validation of test sequence as error, dump log files to console, exit on error.""" + logging.error(f'{test_sequence_name} - Validation of test sequence 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}') + logging.exception(f'{test_sequence_name} - Failed to dump {log_file_path}: {e}') sys.exit(1) -def extract_value_from_string(line: str, value_name: str, casting_state: str, log_paths) -> str: - """Extract and return value from given input string. - - Some string examples as they are received from the Linux tv-casting-app and/or tv-app output: - 1. On 'darwin' machines: - \x1b[0;34m[1715206773402] [20056:2842184] [DMG] Cluster = 0x506,\x1b[0m - The substring to be extracted here is '0x506'. - - Or: - \x1b[0;32m[1714582264602] [77989:2286038] [SVR] Discovered Commissioner #0\x1b[0m - The integer value to be extracted here is '0'. - - Or: - \x1b[0;34m[1713741926895] [7276:9521344] [DIS] Vendor ID: 65521\x1b[0m - The integer value to be extracted here is '65521'. - - Or: - \x1b[0;34m[1714583616179] [7029:2386956] [SVR] device Name: Test TV casting app\x1b[0m - The substring to be extracted here is 'Test TV casting app'. +def stop_app(test_sequence_name: str, app_name: str, app: subprocess.Popen): + """Stop the given `app` subprocess.""" - 2. On 'linux' machines: - [1716224960.316809][6906:6906] CHIP:DMG: \t\t\t\t\tCluster = 0x506,\n - [1716224958.576320][6906:6906] CHIP:SVR: Discovered Commissioner #0 - [1716224958.576407][6906:6906] CHIP:DIS: \tVendor ID: 65521\n - [1716224959.580746][6906:6906] CHIP:SVR: \tdevice Name: Test TV casting app\n - """ - log_line_pattern = '' - if sys.platform == 'darwin': - log_line_pattern = r'\x1b\[0;\d+m\[\d+\] \[\d+:\d+\] \[[A-Z]{1,3}\] (.+)\x1b\[0m' - elif sys.platform == 'linux': - log_line_pattern = r'\[\d+\.\d+\]\[\d+:\d+\] [A-Z]{1,4}:[A-Z]{1,3}: (.+)' - - log_line_match = re.search(log_line_pattern, line) - - if log_line_match: - log_text_of_interest = log_line_match.group(1) - - if '=' in log_text_of_interest: - delimiter = '=' - elif '#' in log_text_of_interest: - delimiter = '#' - else: - delimiter = ':' + app.terminate() + app_exit_code = app.wait() - return log_text_of_interest.split(delimiter)[-1].strip(' ,') + if app.poll() is None: + logging.error(f'{test_sequence_name}: Failed to stop running {app_name}. Process is still running.') else: - raise ValueError(f'Could not extract {value_name} from the following line: {line}') - - -def validate_value(casting_state: str, expected_value: Union[str, int], log_paths: List[str], line: str, value_name: str) -> Optional[str]: - """Validate a value in a string against an expected value during a given casting state.""" - log_value_extractor = LogValueExtractor(casting_state, log_paths) - value = log_value_extractor.extract_from(line, value_name) - if not value: - logging.error(f'Failed to extract {value_name} value from the following line: {line}') - logging.error(f'Failed to validate against the expected {value_name} value: {expected_value}!') - handle_casting_failure(casting_state, log_paths) - - if isinstance(expected_value, int): - value = int(value) - - 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_casting_failure(casting_state, log_paths) - - # 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 initiate_cast_request_success(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], valid_discovered_commissioner_number: str) -> bool: - """Initiate commissioning between Linux tv-casting-app and tv-app by sending `cast request {valid_discovered_commissioner_number}` via Linux tv-casting-app process.""" - tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info - - start_wait_time = time.time() - - while True: - # Check if we exceeded the maximum wait time for initiating 'cast request' from the Linux tv-casting-app to the Linux tv-app. - if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: - logging.error( - f'The command `cast request {valid_discovered_commissioner_number}` was not issued to the Linux tv-casting-app process within the timeout.') - return False - - tv_casting_app_output_line = tv_casting_app_process.stdout.readline() - if tv_casting_app_output_line: - linux_tv_casting_app_log_file.write(tv_casting_app_output_line) - linux_tv_casting_app_log_file.flush() - - if 'cast request 0' in tv_casting_app_output_line: - tv_casting_app_process.stdin.write('cast request ' + valid_discovered_commissioner_number + '\n') - tv_casting_app_process.stdin.flush() - # Move to the next line otherwise we will keep entering this code block - next_line = tv_casting_app_process.stdout.readline() - linux_tv_casting_app_log_file.write(next_line) - linux_tv_casting_app_log_file.flush() - next_line = next_line.rstrip('\n') - logging.info(f'Sent `{next_line}` to the Linux tv-casting-app process.') - - return True - - -def extract_device_info_from_tv_casting_app(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], casting_state: str, log_paths: List[str]) -> Tuple[Optional[str], Optional[int], Optional[int]]: - """Extract device information from the 'Identification Declaration' block in the Linux tv-casting-app output.""" - tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info - log_value_extractor = LogValueExtractor(casting_state, log_paths) - - device_name = None - vendor_id = None - product_id = None - - for line in tv_casting_app_process.stdout: - linux_tv_casting_app_log_file.write(line) - linux_tv_casting_app_log_file.flush() - - if value := log_value_extractor.extract_from(line, 'device Name'): - device_name = value - elif value := log_value_extractor.extract_from(line, 'vendor id'): - vendor_id = int(value) - elif value := log_value_extractor.extract_from(line, 'product id'): - product_id = int(value) - - if device_name and vendor_id and product_id: - break - - return device_name, vendor_id, product_id - - -def validate_identification_declaration_message_on_tv_app(tv_app_info: Tuple[subprocess.Popen, TextIO], expected_device_name: str, expected_vendor_id: int, expected_product_id: int, log_paths: List[str]) -> bool: - """Validate device information from the 'Identification Declaration' block from the Linux tv-app output against the expected values.""" - tv_app_process, linux_tv_app_log_file = tv_app_info - - parsing_identification_block = False - start_wait_time = time.time() - - while True: - # Check if we exceeded the maximum wait time for validating the device information from the Linux tv-app to the corresponding values from the Linux tv-app. - if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: - logging.error( - 'The device information from the Linux tv-app output was not validated against the corresponding values from the Linux tv-casting-app output within the timeout.') - return False - - tv_app_line = tv_app_process.stdout.readline() - - if tv_app_line: - linux_tv_app_log_file.write(tv_app_line) - linux_tv_app_log_file.flush() - - if 'Identification Declaration Start' in tv_app_line: - logging.info('Found the `Identification Declaration` block in the Linux tv-app output:') - logging.info(tv_app_line.rstrip('\n')) - parsing_identification_block = True - elif parsing_identification_block: - logging.info(tv_app_line.rstrip('\n')) - if 'device Name' in tv_app_line: - validate_value('Commissioning', expected_device_name, log_paths, tv_app_line, 'device Name') - elif 'vendor id' in tv_app_line: - validate_value('Commissioning', expected_vendor_id, log_paths, tv_app_line, 'vendor id') - elif 'product id' in tv_app_line: - validate_value('Commissioning', expected_product_id, log_paths, tv_app_line, 'product id') - elif 'Identification Declaration End' in tv_app_line: - parsing_identification_block = False - return True - - -def validate_tv_casting_request_approval(tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]) -> bool: - """Validate that the TV casting request from the Linux tv-casting-app to the Linux tv-app is approved by sending `controller ux ok` via Linux tv-app process.""" - tv_app_process, linux_tv_app_log_file = tv_app_info - - start_wait_time = time.time() - - while True: - # Check if we exceeded the maximum wait time for sending 'controller ux ok' from the Linux tv-app to the Linux tv-casting-app. - if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: - logging.error('The cast request from the Linux tv-casting-app to the Linux tv-app was not approved within the timeout.') - return False - - tv_app_line = tv_app_process.stdout.readline() - - if tv_app_line: - linux_tv_app_log_file.write(tv_app_line) - linux_tv_app_log_file.flush() - - if 'PROMPT USER: Test TV casting app is requesting permission to cast to this TV, approve?' in tv_app_line: - logging.info(tv_app_line.rstrip('\n')) - elif 'Via Shell Enter: controller ux ok|cancel' in tv_app_line: - logging.info(tv_app_line.rstrip('\n')) - - tv_app_process.stdin.write('controller ux ok\n') - tv_app_process.stdin.flush() - - tv_app_line = tv_app_process.stdout.readline() - linux_tv_app_log_file.write(tv_app_line) - linux_tv_app_log_file.flush() - tv_app_line = tv_app_line.rstrip('\n') - logging.info(f'Sent `{tv_app_line}` to the Linux tv-app process.') - return True - - -def validate_commissioning_success(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]) -> bool: - """Parse output of Linux tv-casting-app and Linux tv-app output for strings indicating commissioning status.""" - tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info - tv_app_process, linux_tv_app_log_file = tv_app_info - - start_wait_time = time.time() - - while True: - # Check if we exceeded the maximum wait time for validating commissioning success between the Linux tv-casting-app and the Linux tv-app. - if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: - logging.error( - 'The commissioning between the Linux tv-casting-app process and the Linux tv-app process did not complete successfully within the timeout.') - return False - - tv_casting_line = tv_casting_app_process.stdout.readline() - tv_app_line = tv_app_process.stdout.readline() - - if tv_casting_line: - linux_tv_casting_app_log_file.write(tv_casting_line) - linux_tv_casting_app_log_file.flush() - - if 'Commissioning completed successfully' in tv_casting_line: - logging.info('Commissioning success noted on the Linux tv-casting-app output:') - logging.info(tv_casting_line.rstrip('\n')) - elif 'Commissioning failed' in tv_casting_line: - logging.error('Commissioning failed noted on the Linux tv-casting-app output:') - logging.error(tv_casting_line.rstrip('\n')) - return False - - if tv_app_line: - linux_tv_app_log_file.write(tv_app_line) - linux_tv_app_log_file.flush() - - if 'PROMPT USER: commissioning success' in tv_app_line: - logging.info('Commissioning success noted on the Linux tv-app output:') - logging.info(tv_app_line) + if app_exit_code < 0: + signal_number = -app_exit_code + if signal_number == signal.SIGTERM.value: + logging.info(f'{test_sequence_name}: {app_name} stopped by {signal_number} (SIGTERM) signal.') return True + else: + logging.error( + f'{test_sequence_name}: {app_name} stopped by signal {signal_number} instead of {signal.SIGTERM.value} (SIGTERM).') + else: + logging.error(f'{test_sequence_name}: {app_name} exited with unexpected exit code {app_exit_code}.') + return False -def parse_tv_casting_app_for_report_data_msg(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]): - """Parse the Linux tv-casting-app for `ReportDataMessage` block and return the first message block with valid `Cluster` and `Attribute` values.""" - tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info - log_value_extractor = LogValueExtractor('Testing subscription', log_paths) - - continue_parsing = False - report_data_message = [] - - start_wait_time = time.time() - - while True: - # Check if we exceeded the maximum wait time to parse the Linux tv-casting-app output for `ReportDataMessage` block. - if time.time() - start_wait_time > VERIFY_SUBSCRIPTION_STATE_MAX_WAIT_SEC: - logging.error( - 'The relevant `ReportDataMessage` block for the MediaPlayback:CurrentState subscription was not found in the Linux tv-casting-app process within the timeout.') - report_data_message.clear() - return report_data_message - - tv_casting_line = tv_casting_app_process.stdout.readline() - - if tv_casting_line: - linux_tv_casting_app_log_file.write(tv_casting_line) - linux_tv_casting_app_log_file.flush() - - if 'ReportDataMessage =' in tv_casting_line: - report_data_message.append(tv_casting_line.rstrip('\n')) - continue_parsing = True - elif continue_parsing: - report_data_message.append(tv_casting_line.rstrip('\n')) - - if cluster_value := log_value_extractor.extract_from(tv_casting_line, 'Cluster ='): - if cluster_value != CLUSTER_MEDIA_PLAYBACK: - report_data_message.clear() - continue_parsing = False - - elif attribute_value := log_value_extractor.extract_from(tv_casting_line, 'Attribute ='): - if attribute_value != ATTRIBUTE_CURRENT_PLAYBACK_STATE: - report_data_message.clear() - continue_parsing = False - - elif 'InteractionModelRevision' in tv_casting_line: - # Capture the closing brace `}` of the `ReportDataMessage` block. - tv_casting_line = tv_casting_app_process.stdout.readline() - linux_tv_casting_app_log_file.write(tv_casting_line) - linux_tv_casting_app_log_file.flush() - report_data_message.append(tv_casting_line.rstrip('\n')) - return report_data_message +def parse_output_msg_in_subprocess( + tv_casting_app_info: Tuple[subprocess.Popen, TextIO], + tv_app_info: Tuple[subprocess.Popen, TextIO], + log_paths: List[str], + test_sequence_name: str, + test_sequence_step: Step +): + """Parse the output of a given `app` subprocess and validate its output against the expected `output_msg` in the given `Step`.""" -def parse_tv_app_output_for_launchUrl_msg_success(tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]): - """Parse the Linux tv-app output for the relevant string indicating that the launchUrl was received.""" + if not test_sequence_step.output_msg: + logging.error(f'{test_sequence_name} - No output message provided in the test sequence step.') + return False - tv_app_process, linux_tv_app_log_file = tv_app_info + app_subprocess, app_log_file = (tv_casting_app_info if test_sequence_step.app == App.TV_CASTING_APP else tv_app_info) start_wait_time = time.time() + msg_block = [] - while True: - # Check if we exceeded the maximum wait time to parse the Linux tv-app output for the string related to the launchUrl. - if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: + current_index = 0 + while current_index < len(test_sequence_step.output_msg): + # Check if we exceeded the maximum wait time to parse for the output string(s). + if time.time() - start_wait_time > test_sequence_step.timeout_sec: logging.error( - 'The relevant launchUrl string was not found in the Linux tv-app process within the timeout.') + f'{test_sequence_name} - Did not find the expected output string(s) in the {test_sequence_step.app.value} subprocess within the timeout: {test_sequence_step.output_msg}') return False - tv_app_line = tv_app_process.stdout.readline() + output_line = app_subprocess.stdout.readline() + + if output_line: + app_log_file.write(output_line) + app_log_file.flush() + + if (test_sequence_step.output_msg[current_index] in output_line): + msg_block.append(output_line.rstrip('\n')) + current_index += 1 + elif msg_block: + msg_block.append(output_line.rstrip('\n')) + if (test_sequence_step.output_msg[0] in output_line): + msg_block.clear() + msg_block.append(output_line.rstrip('\n')) + current_index = 1 + # Sanity check that `Discovered Commissioner #0` is the valid commissioner. + elif 'Discovered Commissioner #' in output_line: + logging.error(f'{test_sequence_name} - The valid discovered commissioner should be `Discovered Commissioner #0`.') + handle_casting_failure(test_sequence_name, log_paths) + + if current_index == len(test_sequence_step.output_msg): + logging.info(f'{test_sequence_name} - Found the expected output string(s) in the {test_sequence_step.app.value} subprocess:') + for line in msg_block: + logging.info(f'{test_sequence_name} - {line}') - if tv_app_line: - linux_tv_app_log_file.write(tv_app_line) - linux_tv_app_log_file.flush() - - if 'ContentLauncherManager::HandleLaunchUrl TEST CASE ContentURL=https://www.test.com/videoid DisplayString=Test video' in tv_app_line: - logging.info('Found the launchUrl in the Linux tv-app output:') - logging.info(tv_app_line.rstrip('\n')) return True -def parse_tv_casting_app_output_for_launchUrl_msg_success(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]): - """Parse the Linux tv-casting-app output for relevant strings indicating that the launchUrl was sent.""" - - tv_casting_app_process, linux_tv_casting_app_log_file = tv_casting_app_info - - continue_parsing_invoke_response_msg_block = False - found_example_data_msg = False - start_wait_time = time.time() - - while True: - # Check if we exceeded the maximum wait time to parse the Linux tv-casting-app output for strings related to the launchUrl. - if time.time() - start_wait_time > TEST_LAUNCHURL_MAX_WAIT_SEC: - logging.error( - 'The relevant launchUrl strings were not found in the Linux tv-casting-app process within the timeout.') - return False - - tv_casting_line = tv_casting_app_process.stdout.readline() - - if tv_casting_line: - linux_tv_casting_app_log_file.write(tv_casting_line) - linux_tv_casting_app_log_file.flush() - - if 'InvokeResponseMessage =' in tv_casting_line: - logging.info('Found the `InvokeResponseMessage` block in the Linux tv-casting-app output:') - logging.info(tv_casting_line.rstrip('\n')) - continue_parsing_invoke_response_msg_block = True - - elif continue_parsing_invoke_response_msg_block: - # Sanity check for `exampleData` in the `InvokeResponseMessage` block. - if 'exampleData' in tv_casting_line: - found_example_data_msg = True - - elif 'Received Command Response Data' in tv_casting_line: - if not found_example_data_msg: - logging.error('The `exampleData` string was not found in the `InvokeResponseMessage` block.') - return False - - logging.info('Found the `Received Command Response Data` string in the Linux tv-casting-app output:') - logging.info(tv_casting_line.rstrip('\n')) - return True - - logging.info(tv_casting_line.rstrip('\n')) - - -def test_discovery_fn(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]) -> Optional[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_casting_failure('Discovery', 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('Discovery', VENDOR_ID, log_paths, line, 'Vendor ID') - - elif 'Product ID:' in line: - valid_product_id = validate_value('Discovery', PRODUCT_ID, log_paths, line, 'Product ID') - - elif 'Device Type:' in line: - valid_device_type = validate_value('Discovery', 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 output:') - logging.info(valid_discovered_commissioner) - logging.info(valid_vendor_id) - logging.info(valid_product_id) - logging.info(valid_device_type) - logging.info('Discovery success!\n') - break - - return valid_discovered_commissioner - - -def test_commissioning_fn(valid_discovered_commissioner_number, tv_casting_app_info: Tuple[subprocess.Popen, TextIO], tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]): - """Test commissioning between Linux tv-casting-app and Linux tv-app.""" - - if not initiate_cast_request_success(tv_casting_app_info, valid_discovered_commissioner_number): - handle_casting_failure('Commissioning', log_paths) - - # Extract the values from the 'Identification Declaration' block in the tv-casting-app output that we want to validate against. - expected_device_name, expected_vendor_id, expected_product_id = extract_device_info_from_tv_casting_app( - tv_casting_app_info, 'Commissioning', log_paths) - - if not expected_device_name or not expected_vendor_id or not expected_product_id: - logging.error('There is an error with the expected device info values that were extracted from the `Identification Declaration` block.') - logging.error( - f'expected_device_name: {expected_device_name}, expected_vendor_id: {expected_vendor_id}, expected_product_id: {expected_product_id}') - handle_casting_failure('Commissioning', log_paths) - - if not validate_identification_declaration_message_on_tv_app(tv_app_info, expected_device_name, expected_vendor_id, expected_product_id, log_paths): - handle_casting_failure('Commissioning', log_paths) - - if not validate_tv_casting_request_approval(tv_app_info, log_paths): - handle_casting_failure('Commissioning', log_paths) - - if not validate_commissioning_success(tv_casting_app_info, tv_app_info, log_paths): - handle_casting_failure('Commissioning', log_paths) - - -def test_subscription_fn(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]): - """Test the subscription state of the Linux tv-casting-app by validating the `ReportDataMessage` block.""" - - valid_report_data_msg = parse_tv_casting_app_for_report_data_msg(tv_casting_app_info, log_paths) - - if valid_report_data_msg: - logging.info('Found the `ReportDataMessage` block in the Linux tv-casting-app output:') - - for line in valid_report_data_msg: - logging.info(line) - - logging.info('Testing subscription success!\n') - valid_report_data_msg.clear() +def send_input_cmd_to_subprocess( + tv_casting_app_info: Tuple[subprocess.Popen, TextIO], + tv_app_info: Tuple[subprocess.Popen, TextIO], + test_sequence_name: str, + test_sequence_step: Step +): + """Send a given input command (`input_cmd`) from the `Step` to its given `app` subprocess.""" + + if not test_sequence_step.input_cmd: + logging.error(f'{test_sequence_name} - No input command provided in the test sequence step.') + return False + + app_subprocess, app_log_file = (tv_casting_app_info if test_sequence_step.app == App.TV_CASTING_APP else tv_app_info) + + app_subprocess.stdin.write(test_sequence_step.input_cmd) + app_subprocess.stdin.flush() + + # Read in the next line which should be the `input_cmd` that was issued. + next_line = app_subprocess.stdout.readline() + app_log_file.write(next_line) + app_log_file.flush() + next_line = next_line.rstrip('\n') + + logging.info(f'{test_sequence_name} - Sent `{next_line}` to the {test_sequence_step.app.value} subprocess.') + + return True + + +def handle_output_msg( + tv_casting_app_info: Tuple[subprocess.Popen, TextIO], + tv_app_info: Tuple[subprocess.Popen, TextIO], + log_paths: List[str], + test_sequence_name: str, + test_sequence_step: Step +): + """Handle the output message (`output_msg`) from a test sequence step.""" + + if not parse_output_msg_in_subprocess(tv_casting_app_info, tv_app_info, log_paths, test_sequence_name, test_sequence_step): + handle_casting_failure(test_sequence_name, log_paths) + + +def handle_input_cmd( + tv_casting_app_info: Tuple[subprocess.Popen, TextIO], + tv_app_info: Tuple[subprocess.Popen, TextIO], + log_paths: List[str], + test_sequence_name: str, + test_sequence_step: Step +): + """Handle the input command (`input_cmd`) from a test sequence step.""" + + tv_casting_app_process, tv_casting_app_log_file = tv_casting_app_info + tv_app_process, tv_app_log_file = tv_app_info + + if test_sequence_step.input_cmd == STOP_APP: + if test_sequence_step.app == App.TV_CASTING_APP: + # Stop the tv-casting-app subprocess. + if not stop_app(test_sequence_name, test_sequence_step.app.value, tv_casting_app_process): + handle_casting_failure(test_sequence_name, log_paths) + elif test_sequence_step.app == App.TV_APP: + # Stop the tv-app subprocess. + if not stop_app(test_sequence_name, test_sequence_step.app.value, tv_app_process): + handle_casting_failure(test_sequence_name, log_paths) else: - handle_casting_failure('Testing subscription', log_paths) + if not send_input_cmd_to_subprocess(tv_casting_app_info, tv_app_info, test_sequence_name, test_sequence_step): + handle_casting_failure(test_sequence_name, log_paths) + +def run_test_sequence_steps( + current_index: int, + test_sequence_name: str, + test_sequence_steps: List[Step], + tv_casting_app_info: Tuple[subprocess.Popen, TextIO], + tv_app_info: Tuple[subprocess.Popen, TextIO], + log_paths: List[str] +): + """Run through the test steps from a test sequence starting from the current index and perform actions based on the presence of `output_msg` or `input_cmd`.""" -def test_launchUrl_fn(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], tv_app_info: Tuple[subprocess.Popen, TextIO], log_paths: List[str]): - """Test that the Linux tv-casting-app sent the launchUrl and that the Linux tv-app received the launchUrl.""" + if test_sequence_steps is None: + logging.error('No test sequence steps provided.') - if not parse_tv_app_output_for_launchUrl_msg_success(tv_app_info, log_paths): - handle_casting_failure('Testing launchUrl', log_paths) + while current_index < len(test_sequence_steps): + # Current step in the list of steps. + test_sequence_step = test_sequence_steps[current_index] - if not parse_tv_casting_app_output_for_launchUrl_msg_success(tv_casting_app_info, log_paths): - handle_casting_failure('Testing launchUrl', log_paths) + # A test sequence step contains either an output_msg or input_cmd entry. + if test_sequence_step.output_msg: + handle_output_msg(tv_casting_app_info, tv_app_info, log_paths, test_sequence_name, test_sequence_step) + elif test_sequence_step.input_cmd: + handle_input_cmd(tv_casting_app_info, tv_app_info, log_paths, test_sequence_name, test_sequence_step) - logging.info('Testing launchUrl success!') + current_index += 1 @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_casting_fn(tv_app_rel_path, tv_casting_app_rel_path): - """Test if the Linux tv-casting-app is able to discover and commission the Linux tv-app as part of casting. + """Test if the casting experience between the Linux tv-casting-app and the Linux tv-app continues to work. 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) @@ -615,41 +274,65 @@ def test_casting_fn(tv_app_rel_path, tv_casting_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: + # Get all the test sequences. + test_sequences = Sequence.get_test_sequences() + + # Get the test sequence of interest. + test_sequence = Sequence.get_test_sequence_by_name(test_sequences, 'commissionee_generated_passcode_test') + + if not test_sequence: + logging.error('No test sequence found by the test sequence name provided.') + handle_casting_failure(None, []) + + # At this point, we have retrieved the test sequence of interest. + test_sequence_name = test_sequence.name + test_sequence_steps = test_sequence.steps + # 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'] + current_index = 0 + if test_sequence_steps[current_index].input_cmd != START_APP: + raise ValueError( + f'{test_sequence_name}: The first step in the test sequence must contain `START_APP` as `input_cmd` to indicate starting the tv-app.') + elif test_sequence_steps[current_index].app != App.TV_APP: + raise ValueError(f'{test_sequence_name}: The first step in the test sequence must be to start up the tv-app.') + current_index += 1 + 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], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as tv_app_process: + tv_app_info = (tv_app_process, linux_tv_app_log_file) + + # Verify that the tv-app is up and running. + handle_output_msg(None, tv_app_info, [linux_tv_app_log_path], + test_sequence_name, test_sequence_steps[current_index]) + current_index += 1 - if not start_up_tv_app_success(tv_app_process, linux_tv_app_log_file): - handle_casting_failure('Discovery', [linux_tv_app_log_path]) + if test_sequence_steps[current_index].input_cmd != START_APP: + raise ValueError( + f'{test_sequence_name}: The third step in the test sequence must contain `START_APP` as `input_cmd` to indicate starting the tv-casting-app.') + elif test_sequence_steps[current_index].app != App.TV_CASTING_APP: + raise ValueError( + f'{test_sequence_name}: The third step in the test sequence must be to start up the tv-casting-app.') + current_index += 1 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], stdin=subprocess.PIPE, 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) - tv_app_info = (tv_app_process, linux_tv_app_log_file) - valid_discovered_commissioner = test_discovery_fn(tv_casting_app_info, log_paths) - - if not valid_discovered_commissioner: - handle_casting_failure('Discovery', log_paths) - - # We need the valid discovered commissioner number to continue with commissioning. - log_value_extractor = LogValueExtractor('Commissioning', log_paths) - valid_discovered_commissioner_number = log_value_extractor.extract_from( - valid_discovered_commissioner, 'Discovered Commissioner #') - if not valid_discovered_commissioner_number: - logging.error(f'Failed to find `Discovered Commissioner #` in line: {valid_discovered_commissioner}') - handle_casting_failure('Commissioning', log_paths) - - test_commissioning_fn(valid_discovered_commissioner_number, tv_casting_app_info, tv_app_info, log_paths) - test_subscription_fn(tv_casting_app_info, log_paths) - test_launchUrl_fn(tv_casting_app_info, tv_app_info, log_paths) + + # Verify that the server initialization is completed in the tv-casting-app output. + handle_output_msg(tv_casting_app_info, tv_app_info, log_paths, + test_sequence_name, test_sequence_steps[current_index]) + current_index += 1 + + run_test_sequence_steps(current_index, test_sequence_name, test_sequence_steps, + tv_casting_app_info, tv_app_info, log_paths) if __name__ == '__main__':