Testing with the timeout error logic.

Author: shaoltan-amazon

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 pdb # SHAO remove when done testing!
 import logging
 import os
 import subprocess
@@ -31,14 +32,18 @@
 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 = 10
+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
 
+# Start the clock for this function
+# START_TIME = 0 # SHAO remove when done testing!
+START_TIME = time.time()
+
 # 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'
@@ -88,6 +93,8 @@ def dump_temporary_logs_to_console(log_file_path: str):
 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(casting_state + ' failed!')
+    pdb.set_trace()
+    print('SHAO remove when done testing!!!')
 
     for log_file_path in log_file_paths:
         try:
@@ -99,45 +106,53 @@ def handle_casting_failure(casting_state: str, log_file_paths: List[str]):
 
 
 # def extract_value_from_string(line: str) -> str: # SHAO OG
-def extract_value_from_string(line: str, casting_state: str, log_paths) -> str:
+def extract_value_from_string(line: str, value_name: str, casting_state: str, log_paths) -> str:
     """Extract and return value from given input string.
 
     The string is expected to be in the following format as it is received
     from the Linux tv-casting-app output:
     \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 value '0' will be extracted from the string.
+    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'.
     """
-    # if '=' in line:
-    #     value = line.split('=')[-1].strip().replace(',\x1b[0m', '')
-    # else:
-    #     value = line.split(':')[-1].strip().replace('\x1b[0m', '')
+    if '=' in line:
+        value = line.split('=')[-1].strip().replace(',\x1b[0m', '')
+    else:
+        value = line.split(':')[-1].strip().replace('\x1b[0m', '')
 
-    # return value
+    return value
 
-    if ':' in line:
-        if '=' in line:
-            delimiter = '='
-        else:
-            delimiter = ':'
+    # SHAO working
+    # if ':' in line:
+    #     if '=' in line:
+    #         delimiter = '='
+    #     elif '#' in line:
+    #         delimiter = '#'
+    #     else:
+    #         delimiter = ':'
 
-        value = line.split(delimiter)[-1].strip().replace('\x1b[0m', '').rstrip(',')
-    else:
-        logging.error('Could not extract the value from the following line: %s', line)
-        handle_casting_failure(casting_state, log_paths)
+    #     value = line.split(delimiter)[-1].strip().replace('\x1b[0m', '').rstrip(',') # SHAO OG working
+    #     # value = line.split(delimiter)[-1].strip().replace('\x1b[0m', '') # SHAO forcing timeout!
 
-    return value
+    # else:
+    #     logging.error(f'Could not extract {value_name} from the following line: {line}')
+    #     handle_casting_failure(casting_state, log_paths)
+
+    # return value
 
 
 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."""
     # value = extract_value_from_string(line) # SHAO OG
-    value = extract_value_from_string(line, casting_state, log_paths)
+    value = extract_value_from_string(line, value_name, casting_state, log_paths)
 
     if isinstance(expected_value, int):
         value = int(value)
@@ -157,8 +172,10 @@ def start_up_tv_app_success(tv_app_process: subprocess.Popen, linux_tv_app_log_f
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time()
         # 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:
+        # if time.time() - start_wait_time > TV_APP_MAX_START_WAIT_SEC: # SHAO OG
+        if current_time - start_wait_time > TV_APP_MAX_START_WAIT_SEC or current_time - START_TIME > 50: # SHAO remove when done
             logging.error('The Linux tv-app process did not start successfully within the timeout.')
             return False
 
@@ -180,8 +197,10 @@ def initiate_cast_request_success(tv_casting_app_info: Tuple[subprocess.Popen, T
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time()
         # 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:
+        # if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: # SHAO OG
+        if current_time - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC or current_time - START_TIME > 50: # SHAO remove when done
             logging.error('The command `cast request ' + valid_discovered_commissioner_number +
                           '` was not issued to the Linux tv-casting-app process within the timeout.')
             return False
@@ -217,14 +236,14 @@ def extract_device_info_from_tv_casting_app(tv_casting_app_info: Tuple[subproces
 
         if 'device Name' in line:
             # device_name = extract_value_from_string(line) # SHAO OG
-            device_name = extract_value_from_string(line, casting_state, log_paths)
+            device_name = extract_value_from_string(line, '`device Name`', casting_state, log_paths)
         elif 'vendor id' in line:
             # vendor_id = extract_value_from_string(line) # SHAO OG
-            vendor_id = extract_value_from_string(line, casting_state, log_paths)
+            vendor_id = extract_value_from_string(line, '`vendor id`', casting_state, log_paths)
             vendor_id = int(vendor_id)
         elif 'product id' in line:
             # product_id = extract_value_from_string(line) # SHAO OG
-            product_id = extract_value_from_string(line, casting_state, log_paths)
+            product_id = extract_value_from_string(line, '`product id`', casting_state, log_paths)
             product_id = int(product_id)
 
         if device_name and vendor_id and product_id:
@@ -241,8 +260,11 @@ def validate_identification_declaration_message_on_tv_app(tv_app_info: Tuple[sub
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time() # SHAO remove when done
         # 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:
+        # if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: # SHAO OG 
+        if current_time - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC or current_time - START_TIME > 50: # SHAO remove when done
+            # print(f'current_elapsed_time: {current_elapsed_time}, elapsed_time: {elapsed_time}, current_time: {current_time}, START_TIME: {START_TIME}') # SHAO remove when done testing
             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
@@ -277,8 +299,10 @@ def validate_tv_casting_request_approval(tv_app_info: Tuple[subprocess.Popen, Te
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time() # SHAO remove when done
         # 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:
+        # if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: # SHAO OG
+        if current_time - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC or current_time - START_TIME > 50: # SHAO remove when done
             logging.error('The cast request from the Linux tv-casting-app to the Linux tv-app was not approved within the timeout.')
             return False
 
@@ -312,8 +336,10 @@ def validate_commissioning_success(tv_casting_app_info: Tuple[subprocess.Popen,
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time() # SHAO remove when done
         # 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:
+        if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: # SHAO OG
+        
             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
@@ -353,8 +379,10 @@ def parse_tv_casting_app_for_report_data_msg(tv_casting_app_info: Tuple[subproce
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time()
         # 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:
+        # if time.time() - start_wait_time > VERIFY_SUBSCRIPTION_STATE_MAX_WAIT_SEC: # SHAO OG
+        if current_time - start_wait_time > VERIFY_SUBSCRIPTION_STATE_MAX_WAIT_SEC or current_time - START_TIME > 50: # SHAO remove when done
             logging.error(
                 'The relevant `ReportDataMessage` block was not found in the Linux tv-casting-app process within the timeout.')
             report_data_message.clear()
@@ -374,14 +402,14 @@ def parse_tv_casting_app_for_report_data_msg(tv_casting_app_info: Tuple[subproce
 
                 if 'Cluster =' in tv_casting_line:
                     # cluster_value = extract_value_from_string(tv_casting_line) # SHAO OG
-                    cluster_value = extract_value_from_string(tv_casting_line, 'Testing subscription', log_paths)
+                    cluster_value = extract_value_from_string(tv_casting_line, '`Cluster ID`', 'Testing subscription', log_paths)
                     if cluster_value != CLUSTER_MEDIA_PLAYBACK:
                         report_data_message.clear()
                         continue_parsing = False
 
                 elif 'Attribute =' in tv_casting_line:
                     # attribute_value = extract_value_from_string(tv_casting_line) # SHAO OG
-                    attribute_value = extract_value_from_string(tv_casting_line, 'Testing subscription', log_paths)
+                    attribute_value = extract_value_from_string(tv_casting_line, '`Attribute ID`', 'Testing subscription', log_paths)
                     if attribute_value != ATTRIBUTE_CURRENT_PLAYBACK_STATE:
                         report_data_message.clear()
                         continue_parsing = False
@@ -403,8 +431,10 @@ def parse_tv_app_output_for_launchUrl_msg_success(tv_app_info: Tuple[subprocess.
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time() # SHAO remove when done
         # 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:
+        # if time.time() - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC: # SHAO OG
+        if current_time - start_wait_time > COMMISSIONING_STAGE_MAX_WAIT_SEC or current_time - START_TIME > 50: # SHAO remove when done
             logging.error(
                 'The relevant launchUrl string was not found in the Linux tv-app process within the timeout.')
             return False
@@ -431,8 +461,10 @@ def parse_tv_casting_app_output_for_launchUrl_msg_success(tv_casting_app_info: T
     start_wait_time = time.time()
 
     while True:
+        current_time = time.time()
         # 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:
+        # if time.time() - start_wait_time > TEST_LAUNCHURL_MAX_WAIT_SEC: # SHAO OG
+        if current_time - start_wait_time > TEST_LAUNCHURL_MAX_WAIT_SEC or current_time - START_TIME > 50: # SHAO remove when done
             logging.error(
                 'The relevant launchUrl strings were not found in the Linux tv-casting-app process within the timeout.')
             return False
@@ -479,6 +511,11 @@ def test_discovery_fn(tv_casting_app_info: Tuple[subprocess.Popen, TextIO], log_
         linux_tv_casting_app_log_file.write(line)
         linux_tv_casting_app_log_file.flush()
 
+        # SHAO remove when done
+        if time.time() - START_TIME > 50:
+            logging.error('Could not find a valid discovered commissioner within timeout.')
+            handle_casting_failure('Discovery', log_paths)
+
         # Fail fast if "No commissioner discovered" string found.
         if "No commissioner discovered" in line:
             logging.error(line.rstrip('\n'))
@@ -570,6 +607,9 @@ def test_casting_fn(tv_app_rel_path, tv_casting_app_rel_path):
     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
     """
+    # START_TIME = time.time() # SHAO added
+    # print(f'START_TIME set at beginning: {START_TIME}')
+
     # 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)
@@ -604,7 +644,8 @@ def test_casting_fn(tv_app_rel_path, tv_casting_app_rel_path):
                     # We need the valid discovered commissioner number to continue with commissioning.
                     # Example string: \x1b[0;32m[1714582264602] [77989:2286038] [SVR] Discovered Commissioner #0\x1b[0m
                     #                 The value '0' will be extracted from the string.
-                    valid_discovered_commissioner_number = valid_discovered_commissioner.split('#')[-1].replace('\x1b[0m', '')
+                    valid_discovered_commissioner_number = valid_discovered_commissioner.split('#')[-1].replace('\x1b[0m', '') # SHAO OG
+                    # valid_discovered_commissioner_number = extract_value_from_string(valid_discovered_commissioner, '`Discovered Commissioner Number`', 'Discovery', 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)