Addressed PR comments from @sharadb-amazon.

Author: shaoltan-amazon

scripts/tests/linux/tv_casting_test_sequences.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env -S python3 -B
+
+# Copyright (c) 2024 Project CHIP Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import sys
+from enum import Enum
+from typing import List, Optional
+
+"""
+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.
+"""
+
+# 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 default time to wait while parsing for output string(s).
+DEFAULT_TIMEOUT_SEC = 10
+
+# 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
+
+
+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)
+
+    Note: The first entry in the test sequence list of steps should always be the output string to verify that the tv-app 
+    is up and running.
+
+    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')`
+    """
+
+    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:
+            logging.error('Step cannot contain both `output_msg` and `input_cmd`. Either `output_msg` or `input_cmd` should be provided.')
+            sys.exit(1)
+        elif output_msg is None and input_cmd is None:
+            logging.error('Step must contain either `output_msg` or `input_cmd`. Both are `None`.')
+            sys.exit(1)
+
+        # 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
+
+
+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
+
+
+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."""
+
+    test_sequences = [
+        Sequence(
+            name='commissionee_generated_passcode_test',
+            steps=[
+                # Validate that the tv-app is up and running.
+                Step(app=App.TV_APP, timeout_sec=TV_APP_MAX_START_WAIT_SEC, output_msg=['Started commissioner']),
+
+                # 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 =', '},'])
+            ]
+        )
+    ]
+
+    return test_sequences