Simplify the implementation of `run_tv_casting_test.py` for easier addition of future test sequences that verifies that Linux tv-casting-app continues to work with Linux tv-app. (#33855)
* Simplify the implementation of `run_tv_casting_test.py` for easier addition of future test sequences that verifies that Linux tv-casting-app continues to work with Linux tv-app.
* Addressed PR comments from @sharadb-amazon.
* Addressed @sharadb-amazon PR comments. Also restructured the code to prevent circular import issues.
* Run restyle.
* Fixed typo.
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 0000000..e95bc43
--- /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 0000000..9b65e64
--- /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 074d459..45cc617 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
+
+"""
+This script can be used to validate the casting experience between the Linux tv-casting-app and the Linux tv-app.
+
+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.
+"""
# 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
-
-# The maximum amount of time to commission the Linux tv-casting-app and the tv-app before timeout.
-COMMISSIONING_STAGE_MAX_WAIT_SEC = 15
-
-# 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
-
# 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 @@
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 @@
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.
+def stop_app(test_sequence_name: str, app_name: str, app: subprocess.Popen):
+ """Stop the given `app` subprocess."""
- 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'.
+ app.terminate()
+ app_exit_code = app.wait()
- 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'.
-
- 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 = '#'
+ if app.poll() is None:
+ logging.error(f'{test_sequence_name}: Failed to stop running {app_name}. Process is still running.')
+ else:
+ 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:
- delimiter = ':'
+ logging.error(f'{test_sequence_name}: {app_name} exited with unexpected exit code {app_exit_code}.')
- return log_text_of_interest.split(delimiter)[-1].strip(' ,')
+ return False
+
+
+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`."""
+
+ if not test_sequence_step.output_msg:
+ logging.error(f'{test_sequence_name} - No output message 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)
+
+ start_wait_time = time.time()
+ msg_block = []
+
+ 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(
+ 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
+
+ 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}')
+
+ return True
+
+
+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:
- raise ValueError(f'Could not extract {value_name} from the following line: {line}')
+ 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 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)
+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`."""
- if isinstance(expected_value, int):
- value = int(value)
+ if test_sequence_steps is None:
+ logging.error('No test sequence steps provided.')
- 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)
+ while current_index < len(test_sequence_steps):
+ # Current step in the list of steps.
+ test_sequence_step = test_sequence_steps[current_index]
- # Return the line containing the valid value.
- return line.rstrip('\n')
+ # 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)
-
-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)
- return True
-
-
-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_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."""
-
- 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 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:
- logging.error(
- 'The relevant launchUrl string was not found in the Linux tv-app process 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 '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()
- else:
- handle_casting_failure('Testing subscription', log_paths)
-
-
-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 not parse_tv_app_output_for_launchUrl_msg_success(tv_app_info, log_paths):
- handle_casting_failure('Testing launchUrl', log_paths)
-
- if not parse_tv_casting_app_output_for_launchUrl_msg_success(tv_casting_app_info, log_paths):
- handle_casting_failure('Testing launchUrl', log_paths)
-
- 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 @@
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)
- if not start_up_tv_app_success(tv_app_process, linux_tv_app_log_file):
- handle_casting_failure('Discovery', [linux_tv_app_log_path])
+ # 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 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)
+ # 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
- # 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)
+ run_test_sequence_steps(current_index, test_sequence_name, test_sequence_steps,
+ tv_casting_app_info, tv_app_info, log_paths)
if __name__ == '__main__':
