Update Matter Casting automation test script to use the simplified Linux tv-casting-app build command in the workflow yaml file.

Author: shaoltan-amazon

.github/workflows/examples-linux-tv-casting-app.yaml

@@ -61,7 +61,7 @@ jobs:
             - name: Build Linux tv-casting-app
               run: |
                   ./scripts/run_in_build_env.sh \
-                    "scripts/examples/gn_build_example.sh examples/tv-casting-app/linux/ out/tv-casting-app"
+                    "scripts/examples/gn_build_example.sh examples/tv-casting-app/linux/ out/tv-casting-app chip_casting_simplified=true"
 
             - name: Test casting from Linux tv-casting-app to Linux tv-app
               run: |

scripts/tests/linux/tv_casting_test_sequences.py

@@ -18,7 +18,7 @@
 
 """
 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. 
+test script 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 
@@ -71,11 +71,8 @@
 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
+# Value to verify the subscription state against in the Linux tv-casting-app output.
+ATTRIBUTE_CURRENT_PLAYBACK_STATE = 0x0000_0000  # Application Cluster Spec 6.10.6 Attribute ID: Current State of Playback
 
 test_sequences = [
     Sequence(
@@ -93,27 +90,24 @@
             # 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}',
+            # Validate that there is a valid discovered casting player with {PRODUCT_ID}, {VENDOR_ID}, and {DEVICE_TYPE_CASTING_VIDEO_PLAYER} in the tv-casting-app output.
+            Step(app=App.TV_CASTING_APP, output_msg=['Discovered CastingPlayer #0', f'Product ID: {PRODUCT_ID}', f'Vendor ID: {VENDOR_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 tv-casting-app begins the commissioning process.
+            Step(app=App.TV_CASTING_APP, output_msg=[
+                 'CastingPlayer::VerifyOrEstablishConnection() calling OpenBasicCommissioningWindow()']),
 
-            # 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 the `IdentificationDeclaration` message sent from the tv-casting-app to the tv-app will contain the {VENDOR_ID} of the target content app.
+            Step(app=App.TV_CASTING_APP, output_msg=['IdentificationDeclarationOptions::TargetAppInfos list:']),
+            Step(app=App.TV_CASTING_APP, output_msg=[f'TargetAppInfo 1, Vendor ID: {VENDOR_ID}']),
 
             # 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?']),
+                 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']),
@@ -124,23 +118,24 @@
             # 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 the connection succeeded in the tv-casting-app output.
+            Step(app=App.TV_CASTING_APP, output_msg=['Successfully connected to CastingPlayer']),
 
             # 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 that we are able to read the application VendorID value and that it matches {VENDOR_ID}.
+            Step(app=App.TV_CASTING_APP, output_msg=[f'Read VendorID value: {VENDOR_ID}']),
+
+            # Validate that we are able to subscribe to the media playback cluster by reading the CurrentState value and that it matches {ATTRIBUTE_CURRENT_PLAYBACK_STATE}.
+            Step(app=App.TV_CASTING_APP, output_msg=[f'Read CurrentState value: {ATTRIBUTE_CURRENT_PLAYBACK_STATE}']),
 
             # 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 =', '},']),
+            Step(app=App.TV_CASTING_APP, output_msg=['LaunchURL Success with response.data: exampleData']),
 
             # Signal to stop the tv-casting-app as we finished validation.
             Step(app=App.TV_CASTING_APP, input_cmd=STOP_APP),

scripts/tests/run_tv_casting_test.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import glob
 import logging
 import os
 import signal
@@ -63,6 +64,21 @@ def __exit__(self, exception_type, exception_value, traceback):
         self.process.wait()
 
 
+def remove_cached_files(cached_file_pattern: str) -> bool:
+    """Remove any cached files that match the provided pattern."""
+
+    cached_files = glob.glob(cached_file_pattern)  # Returns a list of paths that match the pattern.
+
+    for cached_file in cached_files:
+        try:
+            os.remove(cached_file)
+        except OSError as e:
+            logging.error(f'Failed to remove cached file `{cached_file}` with error: `{e.strerror}`.')
+            return False
+
+    return True
+
+
 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."""
@@ -86,7 +102,7 @@ def handle_casting_failure(test_sequence_name: str, log_file_paths: List[str]):
     sys.exit(1)
 
 
-def stop_app(test_sequence_name: str, app_name: str, app: subprocess.Popen):
+def stop_app(test_sequence_name: str, app_name: str, app: subprocess.Popen) -> bool:
     """Stop the given `app` subprocess."""
 
     app.terminate()
@@ -115,7 +131,7 @@ def parse_output_msg_in_subprocess(
     log_paths: List[str],
     test_sequence_name: str,
     test_sequence_step: Step
-):
+) -> bool:
     """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:
@@ -168,25 +184,22 @@ def send_input_cmd_to_subprocess(
     tv_app_info: Tuple[subprocess.Popen, TextIO],
     test_sequence_name: str,
     test_sequence_step: Step
-):
+) -> bool:
     """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_name = test_sequence_step.app.value
 
-    app_subprocess.stdin.write(test_sequence_step.input_cmd)
+    input_cmd = test_sequence_step.input_cmd
+    app_subprocess.stdin.write(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.')
+    input_cmd = input_cmd.rstrip('\n')
+    logging.info(f'{test_sequence_name} - Sent `{input_cmd}` to the {app_name} subprocess.')
 
     return True
 
@@ -338,7 +351,9 @@ def test_casting_fn(tv_app_rel_path, tv_casting_app_rel_path):
 if __name__ == '__main__':
 
     # Start with a clean slate by removing any previously cached entries.
-    os.system('rm -f /tmp/chip_*')
+    cached_file_pattern = '/tmp/chip_*'
+    if not remove_cached_files(cached_file_pattern):
+        sys.exit(1)
 
     # Test casting (discovery and commissioning) between the Linux tv-casting-app and the tv-app.
     test_casting_fn()