Fix and re-enable Linux tv CI test for commissioner-generated-passcode flow

Author: shaoltan-amazon

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

@@ -72,16 +72,13 @@ jobs:
                     "python3 ./scripts/tests/run_tv_casting_test.py"
               timeout-minutes: 2 # Comment this out to debug if GitHub Action times out.
 
-            # TODO: this test is flaky and was disabled
-            #       https://github.com/project-chip/connectedhomeip/issues/34598
-            #
-            #  - name:
-            #        Test casting from Linux tv-casting-app to Linux tv-app -
-            #        Commissioner Generated Passcode
-            #    run: |
-            #        ./scripts/run_in_build_env.sh \
-            #          "python3 ./scripts/tests/run_tv_casting_test.py --commissioner-generated-passcode=True"
-            #    timeout-minutes: 2 # Comment this out to debug if GitHub Action times out.
+            - name:
+                  Test casting from Linux tv-casting-app to Linux tv-app -
+                  Commissioner Generated Passcode
+              run: |
+                  ./scripts/run_in_build_env.sh \
+                    "python3 ./scripts/tests/run_tv_casting_test.py --commissioner-generated-passcode=True"
+              timeout-minutes: 2 # Comment this out to debug if GitHub Action times out.
 
             - name: Uploading Size Reports
               uses: ./.github/actions/upload-size-reports

scripts/tests/linux/log_line_processing.py

@@ -63,26 +63,30 @@ def _io_thread(self):
         err_wait = select.poll()
         err_wait.register(self.process.stderr, select.POLLIN | select.POLLHUP)
 
-        with open(self.output_path, "wt") as f:
+        with open(self.output_path, "wt", buffering=1) as f:
             f.write("PROCESS START: %s\n" % time.ctime())
+            f.flush()
             while not self.done:
                 changes = out_wait.poll(0.1)
                 if changes:
                     out_line = self.process.stdout.readline()
-                    if not out_line:
+                    if not out_line or out_line == "\n":
                         # stdout closed (otherwise readline should have at least \n)
                         continue
                     f.write(out_line)
+                    f.flush()
                     self.output_lines.put(out_line)
 
                 changes = err_wait.poll(0)
                 if changes:
                     err_line = self.process.stderr.readline()
-                    if not err_line:
+                    if not err_line or err_line == "\n":
                         # stderr closed (otherwise readline should have at least \n)
                         continue
                     f.write(f"!!STDERR!! : {err_line}")
+                    f.flush()
             f.write("PROCESS END: %s\n" % time.ctime())
+            f.flush()
 
     def __enter__(self):
         self.done = False
@@ -92,6 +96,7 @@ def __enter__(self):
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
+            bufsize=1,
         )
         self.io_thread = threading.Thread(target=self._io_thread)
         self.io_thread.start()
@@ -116,10 +121,21 @@ def __exit__(self, exception_type, exception_value, traceback):
 
     def next_output_line(self, timeout_sec=None):
         """Fetch an item from the output queue, potentially with a timeout."""
-        try:
-            return self.output_lines.get(timeout=timeout_sec)
-        except queue.Empty:
-            return None
+        end_time = time.time() + (timeout_sec if timeout_sec is not None else 0)
+        sleep_duration = 0.1
+        while True:
+            remaining_time = end_time - time.time()
+            if remaining_time <= 0:
+                return None
+                
+            if not self.output_lines.empty():
+                return self.output_lines.get_nowait()
+            else:
+                out_line = self.process.stdout.readline()
+                if out_line:
+                    self.output_lines.put(out_line)
+                time.sleep(sleep_duration)
+                sleep_duration = min(10.0, sleep_duration * 2)
 
     def send_to_program(self, input_cmd):
         """Sends the given input command string to the program.

scripts/tests/run_tv_casting_test.py

@@ -87,28 +87,28 @@ def stop_app(test_sequence_name: str, app_name: str, app: ProcessOutputCapture):
 
     if app.process.poll() is None:
         raise TestStepException(
-            f"{test_sequence_name}: Failed to stop running {app_name}. Process is still running.",
+            f"{test_sequence_name} - Failed to stop running {app_name}. Process is still running.",
             test_sequence_name,
             None,
         )
 
     if app_exit_code >= 0:
         raise TestStepException(
-            f"{test_sequence_name}: {app_name} exited with unexpected exit code {app_exit_code}.",
+            f"{test_sequence_name} - {app_name} exited with unexpected exit code {app_exit_code}.",
             test_sequence_name,
             None,
         )
 
     signal_number = -app_exit_code
     if signal_number != signal.SIGTERM.value:
         raise TestStepException(
-            f"{test_sequence_name}: {app_name} stopped by signal {signal_number} instead of {signal.SIGTERM.value} (SIGTERM).",
+            f"{test_sequence_name} - {app_name} stopped by signal {signal_number} instead of {signal.SIGTERM.value} (SIGTERM).",
             test_sequence_name,
             None,
         )
 
     logging.info(
-        f"{test_sequence_name}: {app_name} stopped by {signal_number} (SIGTERM) signal."
+        f"{test_sequence_name} - {app_name} stopped by {signal_number} (SIGTERM) signal."
     )
 
 
@@ -205,6 +205,8 @@ def send_input_cmd_to_subprocess(
         f"{test_sequence_name} - Sent `{input_cmd}` to the {app_name} subprocess."
     )
 
+    time.sleep(0.5)
+
 
 def handle_input_cmd(
     processes: RunningProcesses, test_sequence_name: str, test_sequence_step: Step