Fix flakey stream hls test (#114046)

homeassistant/components/stream/__init__.py

@@ -409,6 +409,13 @@ class Stream:
         self._fast_restart_once = True
         self._thread_quit.set()
 
+    def _set_state(self, available: bool) -> None:
+        """Set the stream state by updating the callback."""
+        # Call with call_soon_threadsafe since we know _async_update_state is always
+        # all callback function instead of using add_job which would have to work
+        # it out each time
+        self.hass.loop.call_soon_threadsafe(self._async_update_state, available)
+
     def _run_worker(self) -> None:
         """Handle consuming streams and restart keepalive streams."""
         # Keep import here so that we can import stream integration without installing reqs
@@ -419,7 +426,7 @@ class Stream:
         wait_timeout = 0
         while not self._thread_quit.wait(timeout=wait_timeout):
             start_time = time.time()
-            self.hass.add_job(self._async_update_state, True)
+            self._set_state(True)
             self._diagnostics.set_value(
                 "keepalive", self.dynamic_stream_settings.preload_stream
             )
@@ -451,7 +458,7 @@ class Stream:
                     continue
                 break
 
-            self.hass.add_job(self._async_update_state, False)
+            self._set_state(False)
             # To avoid excessive restarts, wait before restarting
             # As the required recovery time may be different for different setups, start
             # with trying a short wait_timeout and increase it on each reconnection attempt.

tests/components/stream/test_hls.py

@@ -2,14 +2,13 @@
 
 from datetime import timedelta
 from http import HTTPStatus
-import logging
 from unittest.mock import patch
 from urllib.parse import urlparse
 
 import av
 import pytest
 
-from homeassistant.components.stream import create_stream
+from homeassistant.components.stream import Stream, create_stream
 from homeassistant.components.stream.const import (
     EXT_X_START_LL_HLS,
     EXT_X_START_NON_LL_HLS,
@@ -300,28 +299,22 @@ async def test_stream_retries(
     open_future1 = hass.loop.create_future()
     open_future2 = hass.loop.create_future()
     futures = [open_future2, open_future1]
-    cur_time = 0
 
-    def time_side_effect():
-        logging.info("time side effect")
-        nonlocal cur_time
-        if cur_time >= 80:
-            logging.info("changing return value")
+    original_set_state = Stream._set_state
+
+    def set_state_wrapper(self, state: bool) -> None:
+        if state is False:
             should_retry.return_value = False  # Thread should exit and be joinable.
-        cur_time += 40
-        return cur_time
+        original_set_state(self, state)
 
     def av_open_side_effect(*args, **kwargs):
         hass.loop.call_soon_threadsafe(futures.pop().set_result, None)
         raise av.error.InvalidDataError(-2, "error")
 
     with patch("av.open") as av_open, patch(
-        "homeassistant.components.stream.time"
-    ) as mock_time, patch(
-        "homeassistant.components.stream.STREAM_RESTART_INCREMENT", 0
-    ):
+        "homeassistant.components.stream.Stream._set_state", set_state_wrapper
+    ), patch("homeassistant.components.stream.STREAM_RESTART_INCREMENT", 0):
         av_open.side_effect = av_open_side_effect
-        mock_time.time.side_effect = time_side_effect
         # Request stream. Enable retries which are disabled by default in tests.
         should_retry.return_value = True
         await stream.start()