Addressed PR comments - Changed workflow step name and command name, updated validate_value function, and added succeeding fast logic.

Author: shaoltan-amazon

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

@@ -17,7 +17,7 @@ name: Test TV Casting Example
 on:
     push:
         branches-ignore:
-            - 'dependabot/**'
+            - "dependabot/**"
     pull_request:
     merge_group:
 
@@ -63,10 +63,11 @@ jobs:
                   ./scripts/run_in_build_env.sh \
                     "scripts/examples/gn_build_example.sh examples/tv-casting-app/linux/ out/tv-casting-app"
 
-            - name: Test Discovery between Linux tv-casting-app and Linux tv-app
+            - name: Test casting from Linux tv-casting-app to Linux tv-app
               run: |
                   ./scripts/run_in_build_env.sh \
-                    "python3 ./scripts/tests/run_tv_casting_test.py test-discovery"
+                    "python3 ./scripts/tests/run_tv_casting_test.py test-casting"
+              timeout-minutes: 1
 
             - name: Uploading Size Reports
               uses: ./.github/actions/upload-size-reports

scripts/tests/run_tv_casting_test.py

@@ -24,8 +24,7 @@
 LINUX_TV_APP_LOGS = './scripts/tests/Linux-tv-app-logs.txt'
 LINUX_TV_CASTING_APP_LOGS = './scripts/tests/Linux-tv-casting-app-logs.txt'
 
-RUN_INTERVAL = 5
-PARSE_INTERVAL = 15
+RUN_INTERVAL = 2
 
 VENDOR_ID = 65521
 PRODUCT_ID = 32769
@@ -34,6 +33,11 @@
 
 # Dump the logs to the console in the case of an error.
 def dump_logs_to_console(log_file):
+    if log_file == LINUX_TV_CASTING_APP_LOGS:
+        print('Dumping Linux TV Casting App Logs to Console.')
+    elif log_file == LINUX_TV_APP_LOGS:
+        print('Dumping Linux TV App Logs to Console.')
+
     with open(log_file, 'r') as file:
         logs = file.read()
         print(logs)
@@ -50,9 +54,10 @@ def remove_log_file(log_file):
 # Whenever a failure is discovered, we should print 'Discovery failed!',
 # dump the logs, clean up the log files, exit on error.
 def handle_discovery_failure():
-    print('Discovery failed!')
+    print('Discovery failed!\n')
 
     dump_logs_to_console(LINUX_TV_CASTING_APP_LOGS)
+    dump_logs_to_console(LINUX_TV_APP_LOGS)
 
     remove_log_file(LINUX_TV_CASTING_APP_LOGS)
     remove_log_file(LINUX_TV_APP_LOGS)
@@ -69,152 +74,150 @@ def extract_value_from_string(line):
 
 
 # Check if the discovered value matches the expected value.
-def check_expected_value(line, expected_value, value_name):
+# Returns False if the value does not match, True otherwise.
+def validate_value(expected_value, line, value_name):
     # Extract the integer value from the string
     value = extract_value_from_string(line)
 
     # If the discovered value does not match the expected value,
-    # print the error and handle the discovery failure.
+    # print the error and return False.
     if value != expected_value:
         print(f'{value_name} does not match the expected value!')
-        print(f'Discovered {value_name}: {value}')
         print(f'Expected {value_name}: {expected_value}')
+        line = line.rstrip('\n')
+        print(line)
 
-        handle_discovery_failure()
-
-
-# Read the logs from the Linux-tv-casting-app-logs.txt file.
-# The discovered commissioner(s) will be stored in a list along with their
-# vendor ID, product ID, and device type.
-def parse_linux_tv_casting_app_logs(log_file):
-
-    with open(log_file, 'r') as file:
-        lines = file.readlines()
-
-    discovered_commissioners = []
-
-    print('Reading from Linux-tv-casting-app-logs.txt')
-
-    # Read through the Linux-tv-casting-app-logs.txt line by line
-    for i, line in enumerate(lines):
-
-        # If commissioner(s) are discovered, then the discovery process was successful.
-        if "commissioner(s) discovered" in line:
-            print(line)
-            print('Discovery success!')
-
-            remove_log_file(LINUX_TV_CASTING_APP_LOGS)
-            remove_log_file(LINUX_TV_APP_LOGS)
+        return False
 
-            break
-
-        # Look for "Discovered Commissioner"
-        if "Discovered Commissioner" in line:
-            print(line)
-
-            # Extract the relevant part of the string
-            commissioner = line.split("Discovered Commissioner")[-1].strip()
-            commissioner = commissioner.replace('\x1b[0m', '')
-
-            # Initialize variables for Vendor ID, Product ID, and Device Type
-            vendor_id = None
-            product_id = None
-            device_type = None
-
-            # Iterate through the subsequent lines to find the strings of interest
-            for next_line in lines[i+1:]:
-
-                if "Vendor ID:" in next_line:
-                    print(next_line)
-                    vendor_id = extract_value_from_string(next_line)
-
-                elif "Product ID:" in next_line:
-                    print(next_line)
-                    product_id = extract_value_from_string(next_line)
-
-                elif "Device Type:" in next_line:
-                    print(next_line)
-                    device_type = extract_value_from_string(next_line)
-
-                elif "commissioner(s) discovered" in next_line:
-                    break
-
-                # If the next line starts with "Discovered Commissioner", break the loop
-                if "Discovered Commissioner" in next_line:
-                    break
-
-            # Append the extracted information to the devices list
-            discovered_commissioners.append({
-                "discovered_commissioner": commissioner,
-                "vendor_id": vendor_id,
-                "product_id": product_id,
-                "device_type": device_type
-            })
-
-    # If the list of discovered commissioners is empty and we didn't find the "No commissioner discovered" string,
-    # then something went wrong. Exit on error.
-    if len(discovered_commissioners) == 0:
-        print('No commissioner(s) discovered! The list of discovered commissioner(s) is empty!')
-        handle_discovery_failure()
+    # Return True if the value matches the expected value
+    return True
 
 
 # Test if the Linux tv-casting-app is able to discover the Linux tv-app.
 # The Linux tv-casting-app and the tv-app will be run in separate processes.
 # Their corresponding output will be written to their respective log files.
-# The log file of the tv-casting-app will be parsed for strings of interest
-# which will be printed to the console.
+# The output of the tv-casting-app will be parsed in realtime for strings of
+# interest which will be printed to the console.
 def test_discovery_fn():
 
     with open(LINUX_TV_APP_LOGS, 'w') as fd1, open(LINUX_TV_CASTING_APP_LOGS, 'w') as fd2:
 
         # Run the Linux tv-app and write the output to file
         tv_app_rel_path = 'out/tv-app/chip-tv-app'
         tv_app_abs_path = os.path.abspath(tv_app_rel_path)
-        subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True)
+        tv_app_process = subprocess.Popen(tv_app_abs_path, stdout=fd1, stderr=subprocess.PIPE, text=True)
 
         time.sleep(RUN_INTERVAL)
 
-        # Run the Linux tv-casting-app and write the output to file
+        # Run the Linux tv-casting-app
         tv_casting_app_rel_path = 'out/tv-casting-app/chip-tv-casting-app'
         tv_casting_app_abs_path = os.path.abspath(tv_casting_app_rel_path)
         tv_casting_app_process = subprocess.Popen(
             tv_casting_app_abs_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
 
+        # Initialize variables
+        continue_parsing = False
+        valid_discovered_commissioner_str = ''
+
+        # Read the output as we get it from the tv-casting-app process
         for line in tv_casting_app_process.stdout:
             # Write the line to the Linux tv-casting-app log file
             fd2.write(line)
 
             # Fail fast if "No commissioner discovered" string found
             if "No commissioner discovered" in line:
+                line = line.rstrip('\n')
                 print(line)
                 handle_discovery_failure()
 
-            # Check if the Vendor ID, Product ID, and Device Type match the expected values
-            if "Vendor ID:" in line:
-                check_expected_value(line, VENDOR_ID, "Vendor ID")
+            # Look for 'Discovered Commissioner'
+            if "Discovered Commissioner" in line:
+                line = line.rstrip('\n')
+                valid_discovered_commissioner_str = line
+
+                # Continue parsing the content that belongs to the "Discovered Commissioner"
+                continue_parsing = True
+
+                # Initialize variables to store the information of interest
+                valid_vendor_id = False
+                valid_product_id = False
+                valid_device_type = False
+
+                valid_vendor_id_str = ''
+                valid_product_id_str = ''
+                valid_device_type_str = ''
+
+            if continue_parsing:
+
+                # Check if the Vendor ID, Product ID, and Device Type match the expected values
+                if "Vendor ID:" in line:
+
+                    # If the value of the Vendor ID does not match the expected value, then
+                    # handle the discovery failure.
+                    valid_vendor_id = validate_value(VENDOR_ID, line, "Vendor ID")
 
-            elif "Product ID:" in line:
-                check_expected_value(line, PRODUCT_ID, "Product ID")
+                    if not valid_vendor_id:
+                        handle_discovery_failure()
+                    else:
+                        line = line.rstrip('\n')
+                        valid_vendor_id_str = line
 
-            elif "Device Type:" in line:
-                check_expected_value(line, DEVICE_TYPE, "Device Type")
+                elif "Product ID:" in line:
 
-            if "commissioner(s) discovered" in line:
-                break
+                    # If the value of Product ID does not match the expected value, then
+                    # handle the discovery failure.
+                    valid_product_id = validate_value(PRODUCT_ID, line, "Product ID")
+
+                    if not valid_product_id:
+                        handle_discovery_failure()
+                    else:
+                        line = line.rstrip('\n')
+                        valid_product_id_str = line
+
+                elif "Device Type:" in line:
+
+                    # If the value of Device Type does not match the expected value, then
+                    # handle the discovery failure.
+                    valid_device_type = validate_value(DEVICE_TYPE, line, "Device Type")
+
+                    if not valid_device_type:
+                        handle_discovery_failure()
+                    else:
+                        line = line.rstrip('\n')
+                        valid_device_type_str = line
+
+                        # At this point, all values of interest are valid, so we stop parsing.
+                        continue_parsing = False
+
+                # We only print the discovered commissioner that has valid vendor id, product id,
+                # and device type. Remove the log files once done.
+                if valid_vendor_id and valid_product_id and valid_device_type:
+                    print(valid_discovered_commissioner_str)
+                    print(valid_vendor_id_str)
+                    print(valid_product_id_str)
+                    print(valid_device_type_str)
+                    print('Discovery success!')
+
+                    remove_log_file(LINUX_TV_CASTING_APP_LOGS)
+                    remove_log_file(LINUX_TV_APP_LOGS)
+
+                    break
 
-    # Wait for the processes to finish writing before attempting to read
-    time.sleep(PARSE_INTERVAL)
+        # Tear down the processes.
+        tv_app_process.terminate()
+        tv_app_process.wait()
 
-    parse_linux_tv_casting_app_logs(LINUX_TV_CASTING_APP_LOGS)
+        tv_casting_app_process.terminate()
+        tv_casting_app_process.wait()
 
 
 @click.group()
 def main():
     pass
 
 
-@main.command('test-discovery', help='Test if the Linux tv-casting-app is able to discover the Linux tv-app.')
-def test_discovery():
+@main.command('test-casting', help='Test casting from Linux tv-casting-app to Linux tv-app.')
+def test_casting():
     test_discovery_fn()