Update Liebherr integration to cloud push (#179009)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

homeassistant/components/liebherr/__init__.py

@@ -78,6 +78,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: LiebherrConfigEntry) ->
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
+    # Start SSE push streams for the initial devices
+    for coordinator in data.coordinators.values():
+        coordinator.async_start_stream()
+
     # Schedule periodic scan for new devices
     async def _async_scan_for_new_devices(_now: datetime) -> None:
         """Scan for new devices added to the account."""
@@ -102,7 +106,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: LiebherrConfigEntry) ->
                 if identifier[0] == DOMAIN
             }
             if device_ids - current_device_ids:
-                # Shut down coordinator if one exists
                 for device_id in device_ids:
                     if coordinator := data.coordinators.pop(device_id, None):
                         await coordinator.async_shutdown()
@@ -124,6 +127,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: LiebherrConfigEntry) ->
                     continue
                 data.coordinators[device.device_id] = coordinator
                 new_coordinators.append(coordinator)
+                coordinator.async_start_stream()
 
         if new_coordinators:
             async_dispatcher_send(

homeassistant/components/liebherr/const.py

@@ -6,6 +6,4 @@ from typing import Final
 DOMAIN: Final = "liebherr"
 MANUFACTURER: Final = "Liebherr"
 
-SCAN_INTERVAL: Final = timedelta(seconds=60)
 DEVICE_SCAN_INTERVAL: Final = timedelta(minutes=5)
-REFRESH_DELAY: Final = timedelta(seconds=5)

homeassistant/components/liebherr/coordinator.py

@@ -1,23 +1,27 @@
 """DataUpdateCoordinator for Liebherr integration."""
 
-from dataclasses import dataclass, field
+import asyncio
+from dataclasses import dataclass, field, replace
 import logging
 from typing import override
 
 from pyliebherrhomeapi import (
+    DeviceControl,
     DeviceState,
     LiebherrAuthenticationError,
     LiebherrClient,
     LiebherrConnectionError,
+    LiebherrNotFoundError,
+    LiebherrPreconditionFailedError,
     LiebherrTimeoutError,
 )
 
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
-from .const import DOMAIN, SCAN_INTERVAL
+from .const import DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -34,7 +38,9 @@ type LiebherrConfigEntry = ConfigEntry[LiebherrData]
 
 
 class LiebherrCoordinator(DataUpdateCoordinator[DeviceState]):
-    """Class to manage fetching Liebherr data from the API for a single device."""
+    """Class to manage Liebherr device state via SSE push updates."""
+
+    config_entry: LiebherrConfigEntry
 
     def __init__(
         self,
@@ -48,11 +54,14 @@ class LiebherrCoordinator(DataUpdateCoordinator[DeviceState]):
             hass,
             logger=_LOGGER,
             name=f"{DOMAIN}_{device_id}",
-            update_interval=SCAN_INTERVAL,
             config_entry=config_entry,
         )
         self.client = client
         self.device_id = device_id
+        self._stream_task: asyncio.Task[None] | None = None
+        # First event after each (re)connect carries the full control set;
+        # subsequent events are deltas that get merged into cached state.
+        self._replace_next_event = True
 
     @override
     async def _async_setup(self) -> None:
@@ -73,7 +82,12 @@ class LiebherrCoordinator(DataUpdateCoordinator[DeviceState]):
 
     @override
     async def _async_update_data(self) -> DeviceState:
-        """Fetch data from API for this device."""
+        """Fetch the initial device state.
+
+        Called once by ``async_config_entry_first_refresh`` to seed
+        ``self.data`` before the SSE stream starts. After startup, all
+        updates arrive via ``_async_run_stream``.
+        """
         try:
             return await self.client.get_device_state(self.device_id)
         except LiebherrAuthenticationError as err:
@@ -93,3 +107,84 @@ class LiebherrCoordinator(DataUpdateCoordinator[DeviceState]):
                 translation_key="device_communication_error",
                 translation_placeholders={"device_id": self.device_id},
             ) from err
+
+    @callback
+    def async_start_stream(self) -> None:
+        """Start the SSE stream background task."""
+        if self._stream_task is not None:
+            return
+        self._mark_unavailable()
+        self._stream_task = self.config_entry.async_create_background_task(
+            self.hass,
+            self._async_run_stream(),
+            name=f"{DOMAIN}_stream_{self.device_id}",
+        )
+
+    async def _async_run_stream(self) -> None:
+        """Consume the SSE stream and merge control deltas into state.
+
+        ``stream_controls_forever`` reconnects internally on recoverable
+        errors (connection drops, timeouts, 5xx). Only non-recoverable
+        errors (auth, not-found, precondition) propagate here.
+        """
+        try:
+            async for controls in self.client.stream_controls_forever(
+                self.device_id,
+                on_connect=self._handle_stream_connected,
+                on_disconnect=self._handle_stream_disconnected,
+            ):
+                self._apply_controls(controls)
+        except LiebherrAuthenticationError:
+            self._mark_unavailable()
+            _LOGGER.debug("SSE stream auth failed for %s; starting reauth", self.name)
+            self.config_entry.async_start_reauth(self.hass)
+        except (LiebherrNotFoundError, LiebherrPreconditionFailedError) as err:
+            _LOGGER.warning("SSE stream for device %s stopped: %s", self.device_id, err)
+            self._mark_unavailable()
+
+    @override
+    async def async_shutdown(self) -> None:
+        """Cancel the SSE stream task and shut down the coordinator."""
+        if self._stream_task is not None:
+            self._stream_task.cancel()
+            self._stream_task = None
+        await super().async_shutdown()
+
+    def _apply_controls(self, controls: list[DeviceControl]) -> None:
+        """Apply a control update: replace on (re)connect, merge otherwise."""
+        assert self.data is not None
+        if self._replace_next_event:
+            self._replace_next_event = False
+            new_state = replace(self.data, controls=list(controls))
+        else:
+            merged: dict[tuple[type[DeviceControl], str, int | None], DeviceControl] = {
+                (
+                    type(control),
+                    control.name,
+                    getattr(control, "zone_id", None),
+                ): control
+                for control in self.data.controls
+            }
+            for control in controls:
+                key = (type(control), control.name, getattr(control, "zone_id", None))
+                merged[key] = control
+            new_state = replace(self.data, controls=list(merged.values()))
+        self.async_set_updated_data(new_state)
+
+    @callback
+    def _handle_stream_connected(self) -> None:
+        """Handle SSE (re)connect: next event carries the full state."""
+        self._replace_next_event = True
+
+    @callback
+    def _handle_stream_disconnected(self) -> None:
+        """Handle SSE disconnect: mark entities unavailable."""
+        self._mark_unavailable()
+
+    @callback
+    def _mark_unavailable(self) -> None:
+        """Mark the coordinator as unavailable and notify listeners."""
+        if not self.last_update_success:
+            return
+        self.last_update_success = False
+        self.async_update_listeners()

homeassistant/components/liebherr/entity.py

@@ -1,6 +1,5 @@
 """Base entity for Liebherr integration."""
 
-import asyncio
 from collections.abc import Coroutine
 from typing import Any
 
@@ -15,7 +14,7 @@ from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from .const import DOMAIN, MANUFACTURER, REFRESH_DELAY
+from .const import DOMAIN, MANUFACTURER
 from .coordinator import LiebherrCoordinator
 
 # Zone position to translation key mapping
@@ -56,7 +55,10 @@ class LiebherrEntity(CoordinatorEntity[LiebherrCoordinator]):
         self,
         command: Coroutine[Any, Any, None],
     ) -> None:
-        """Send a command with error handling and delayed refresh."""
+        """Send a command with error handling.
+
+        State updates arrive via the SSE stream — no explicit refresh needed.
+        """
         try:
             await command
         except (LiebherrConnectionError, LiebherrTimeoutError) as err:
@@ -65,9 +67,6 @@ class LiebherrEntity(CoordinatorEntity[LiebherrCoordinator]):
                 translation_key="communication_error",
             ) from err
 
-        await asyncio.sleep(REFRESH_DELAY.total_seconds())
-        await self.coordinator.async_request_refresh()
-
 
 class LiebherrZoneEntity(LiebherrEntity):
     """Base entity for zone-based Liebherr entities.

homeassistant/components/liebherr/manifest.json

@@ -5,7 +5,7 @@
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/liebherr",
   "integration_type": "hub",
-  "iot_class": "cloud_polling",
+  "iot_class": "cloud_push",
   "loggers": ["pyliebherrhomeapi"],
   "quality_scale": "platinum",
   "requirements": ["pyliebherrhomeapi==0.5.1"],

homeassistant/generated/integrations.json

@@ -3950,7 +3950,7 @@
       "name": "Liebherr",
       "integration_type": "hub",
       "config_flow": true,
-      "iot_class": "cloud_polling"
+      "iot_class": "cloud_push"
     },
     "lifx": {
       "name": "LIFX",

tests/components/liebherr/conftest.py

@@ -1,8 +1,8 @@
 """Common fixtures for the liebherr tests."""
 
-from collections.abc import Generator
+import asyncio
+from collections.abc import AsyncIterator, Callable, Generator
 import copy
-from datetime import timedelta
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from pyliebherrhomeapi import (
@@ -10,6 +10,7 @@ from pyliebherrhomeapi import (
     BioFreshPlusControl,
     BioFreshPlusMode,
     Device,
+    DeviceControl,
     DeviceState,
     DeviceType,
     DoorState,
@@ -23,6 +24,11 @@ from pyliebherrhomeapi import (
     ToggleControl,
     ZonePosition,
 )
+from pyliebherrhomeapi.exceptions import (
+    LiebherrAuthenticationError,
+    LiebherrConnectionError,
+    LiebherrTimeoutError,
+)
 import pytest
 
 from homeassistant.components.liebherr.const import DOMAIN
@@ -139,16 +145,6 @@ MOCK_DEVICE_STATE = DeviceState(
 )
 
 
-@pytest.fixture(autouse=True)
-def patch_refresh_delay() -> Generator[None]:
-    """Patch REFRESH_DELAY to 0 to avoid delays in tests."""
-    with patch(
-        "homeassistant.components.liebherr.entity.REFRESH_DELAY",
-        timedelta(seconds=0),
-    ):
-        yield
-
-
 @pytest.fixture
 def mock_setup_entry() -> Generator[AsyncMock]:
     """Override async_setup_entry."""
@@ -158,6 +154,71 @@ def mock_setup_entry() -> Generator[AsyncMock]:
         yield mock_setup_entry
 
 
+class SSEStreamHelper:
+    """Test helper simulating ``LiebherrClient.stream_controls_forever``.
+
+    Yields the current ``client.get_device_state`` result to the coordinator
+    when :meth:`async_push` is called. Auth errors from the mocked
+    ``get_device_state`` propagate to the coordinator; connection/timeout
+    errors trigger the ``on_disconnect`` callback and drop the pending
+    update without terminating the stream.
+    """
+
+    def __init__(self, hass: HomeAssistant, client: MagicMock) -> None:
+        """Initialize the helper."""
+        self._hass = hass
+        self._client = client
+        self._events: dict[str, asyncio.Event] = {}
+        self._on_disconnect: dict[str, Callable[[], None] | None] = {}
+        self._on_connect: dict[str, Callable[[], None] | None] = {}
+        self._reconnect_next: dict[str, bool] = {}
+
+    def _stream(
+        self,
+        device_id: str,
+        *,
+        on_connect: Callable[[], None] | None = None,
+        on_disconnect: Callable[[], None] | None = None,
+        **_: object,
+    ) -> AsyncIterator[list[DeviceControl]]:
+        self._on_connect[device_id] = on_connect
+        self._on_disconnect[device_id] = on_disconnect
+        return self._iter(device_id)
+
+    async def _iter(self, device_id: str) -> AsyncIterator[list[DeviceControl]]:
+        event = self._events.setdefault(device_id, asyncio.Event())
+        state = await self._client.get_device_state(device_id)
+        if (cb := self._on_connect.get(device_id)) is not None:
+            cb()
+        yield state.controls
+        while True:
+            await event.wait()
+            event.clear()
+            reconnect = self._reconnect_next.pop(device_id, False)
+            try:
+                state = await self._client.get_device_state(device_id)
+            except LiebherrAuthenticationError:
+                raise
+            except LiebherrConnectionError, LiebherrTimeoutError:
+                if (cb := self._on_disconnect.get(device_id)) is not None:
+                    cb()
+                continue
+            if reconnect and (cb := self._on_connect.get(device_id)) is not None:
+                cb()
+            yield state.controls
+
+    async def async_push(self, device_id: str = "test_device_id") -> None:
+        """Trigger a stream event: coordinator re-reads ``get_device_state``."""
+        event = self._events.setdefault(device_id, asyncio.Event())
+        event.set()
+        await self._hass.async_block_till_done()
+
+    async def async_reconnect(self, device_id: str = "test_device_id") -> None:
+        """Trigger a stream event that simulates a reconnect (full state replace)."""
+        self._reconnect_next[device_id] = True
+        await self.async_push(device_id)
+
+
 @pytest.fixture
 def mock_config_entry() -> MockConfigEntry:
     """Return a mock config entry."""
@@ -169,7 +230,9 @@ def mock_config_entry() -> MockConfigEntry:
 
 
 @pytest.fixture
-def mock_liebherr_client() -> Generator[MagicMock]:
+def mock_liebherr_client(
+    hass: HomeAssistant,
+) -> Generator[MagicMock]:
     """Return a mocked Liebherr client."""
     with (
         patch(
@@ -197,9 +260,18 @@ def mock_liebherr_client() -> Generator[MagicMock]:
         client.set_bio_fresh_plus = AsyncMock()
         client.set_presentation_light = AsyncMock()
         client.trigger_auto_door = AsyncMock()
+        helper = SSEStreamHelper(hass, client)
+        client.stream_controls_forever.side_effect = helper._stream
+        client._sse_helper = helper
         yield client
 
 
+@pytest.fixture
+def sse_helper(mock_liebherr_client: MagicMock) -> SSEStreamHelper:
+    """Return the SSE stream helper for the mocked client."""
+    return mock_liebherr_client._sse_helper
+
+
 @pytest.fixture
 def platforms() -> list[Platform]:
     """Fixture to specify platforms to test."""

tests/components/liebherr/snapshots/test_diagnostics.ambr

@@ -6,7 +6,7 @@
         'coordinator': dict({
           'last_exception': None,
           'last_update_success': True,
-          'update_interval': '0:01:00',
+          'update_interval': 'None',
         }),
         'data': dict({
           'controls': list([

tests/components/liebherr/test_cover.py

@@ -35,7 +35,7 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
 
-from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE
+from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE, SSEStreamHelper
 
 from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
 
@@ -92,18 +92,16 @@ async def test_covers(
 async def test_cover_state_after_poll(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
     door_state: DoorState | None,
     expected_state: str,
 ) -> None:
-    """Test cover state after polling different door states."""
+    """Test cover state after an SSE push with different door states."""
     mock_liebherr_client.get_device_state.side_effect = lambda *a, **kw: (
         _mock_door_state(door_state)
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_push()
 
     state = hass.states.get(ENTITY_ID)
     assert state is not None
@@ -121,6 +119,7 @@ async def test_cover_state_after_poll(
 async def test_cover_service_calls(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
+    sse_helper: SSEStreamHelper,
     service: str,
     door_state: DoorState,
     expected_state: str,
@@ -144,6 +143,9 @@ async def test_cover_service_calls(
         value=expected_value,
     )
 
+    # Push the confirmed state via SSE to clear the optimistic state
+    await sse_helper.async_push()
+
     state = hass.states.get(ENTITY_ID)
     assert state is not None
     assert state.state == expected_state
@@ -153,9 +155,9 @@ async def test_cover_service_calls(
 async def test_cover_state_settles_after_poll(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
-    """Test door state settles correctly across command and subsequent poll."""
+    """Test door state settles correctly across command and subsequent push."""
     mock_liebherr_client.get_device_state.side_effect = lambda *a, **kw: (
         _mock_door_state(DoorState.OPEN)
     )
@@ -167,18 +169,18 @@ async def test_cover_state_settles_after_poll(
         blocking=True,
     )
 
+    await sse_helper.async_push()
+
     state = hass.states.get(ENTITY_ID)
     assert state is not None
     assert state.state == STATE_OPEN
 
-    # Door closes on next scheduled poll
+    # Door closes on next stream event
     mock_liebherr_client.get_device_state.side_effect = lambda *a, **kw: (
         _mock_door_state(DoorState.CLOSED)
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_push()
 
     state = hass.states.get(ENTITY_ID)
     assert state is not None
@@ -223,7 +225,7 @@ async def test_cover_failure(
 async def test_cover_when_control_missing(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test cover entity behavior when auto door control is removed."""
     state = hass.states.get(ENTITY_ID)
@@ -235,9 +237,7 @@ async def test_cover_when_control_missing(
         device=MOCK_DEVICE, controls=[]
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_reconnect()
 
     state = hass.states.get(ENTITY_ID)
     assert state is not None

tests/components/liebherr/test_init.py

@@ -7,9 +7,11 @@ from unittest.mock import MagicMock, patch
 
 from freezegun.api import FrozenDateTimeFactory
 from pyliebherrhomeapi import (
+    AutoDoorControl,
     Device,
     DeviceState,
     DeviceType,
+    DoorState,
     IceMakerControl,
     IceMakerMode,
     TemperatureControl,
@@ -20,16 +22,19 @@ from pyliebherrhomeapi import (
 from pyliebherrhomeapi.exceptions import (
     LiebherrAuthenticationError,
     LiebherrConnectionError,
+    LiebherrNotFoundError,
+    LiebherrPreconditionFailedError,
+    LiebherrTimeoutError,
 )
 import pytest
 
 from homeassistant.components.liebherr.const import DOMAIN
 from homeassistant.config_entries import ConfigEntryState
-from homeassistant.const import Platform
+from homeassistant.const import STATE_UNAVAILABLE, Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import device_registry as dr
 
-from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE
+from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE, SSEStreamHelper
 
 from tests.common import MockConfigEntry, async_fire_time_changed
 
@@ -88,6 +93,134 @@ async def test_coordinator_setup_errors(
     assert mock_config_entry.state is expected_state
 
 
+@pytest.mark.parametrize(
+    ("side_effect", "expected_state"),
+    [
+        (LiebherrAuthenticationError("Invalid API key"), ConfigEntryState.SETUP_ERROR),
+        (LiebherrTimeoutError("Request timed out"), ConfigEntryState.SETUP_RETRY),
+    ],
+    ids=["auth_failed", "timeout"],
+)
+async def test_coordinator_initial_refresh_errors(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_liebherr_client: MagicMock,
+    side_effect: Exception,
+    expected_state: ConfigEntryState,
+) -> None:
+    """Test coordinator initial refresh errors."""
+    mock_config_entry.add_to_hass(hass)
+    mock_liebherr_client.get_device_state.side_effect = side_effect
+
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert mock_config_entry.state is expected_state
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_start_stream_is_idempotent(
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test starting an active stream does not create another task."""
+    coordinator = mock_config_entry.runtime_data.coordinators[MOCK_DEVICE.device_id]
+    stream_task = coordinator._stream_task
+
+    coordinator.async_start_stream()
+
+    assert coordinator._stream_task is stream_task
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_stream_delta_preserves_same_control_in_other_zone(
+    mock_liebherr_client: MagicMock,
+    mock_config_entry: MockConfigEntry,
+    sse_helper: SSEStreamHelper,
+) -> None:
+    """Test a zoned control delta preserves same-name controls in other zones."""
+    coordinator = mock_config_entry.runtime_data.coordinators[MOCK_DEVICE.device_id]
+    zone_1 = AutoDoorControl(
+        name="autodoor",
+        type="AutoDoorControl",
+        zone_id=1,
+        zone_position=ZonePosition.TOP,
+        value=DoorState.CLOSED,
+    )
+    zone_2 = AutoDoorControl(
+        name="autodoor",
+        type="AutoDoorControl",
+        zone_id=2,
+        zone_position=ZonePosition.BOTTOM,
+        value=DoorState.CLOSED,
+    )
+    coordinator.async_set_updated_data(
+        DeviceState(device=MOCK_DEVICE, controls=[zone_1, zone_2])
+    )
+    updated_zone_1 = AutoDoorControl(
+        name="autodoor",
+        type="AutoDoorControl",
+        zone_id=1,
+        zone_position=ZonePosition.TOP,
+        value=DoorState.OPEN,
+    )
+    mock_liebherr_client.get_device_state.side_effect = lambda *args, **kwargs: (
+        DeviceState(device=MOCK_DEVICE, controls=[updated_zone_1])
+    )
+
+    await sse_helper.async_push()
+
+    assert coordinator.data is not None
+    assert [
+        control
+        for control in coordinator.data.controls
+        if isinstance(control, AutoDoorControl)
+    ] == [updated_zone_1, zone_2]
+
+
+@pytest.mark.usefixtures("init_integration")
+@pytest.mark.parametrize(
+    "exception",
+    [
+        LiebherrNotFoundError("Device not found"),
+        LiebherrPreconditionFailedError("Device not onboarded"),
+    ],
+    ids=["not_found", "precondition_failed"],
+)
+async def test_terminal_stream_error_marks_entities_unavailable(
+    hass: HomeAssistant,
+    mock_liebherr_client: MagicMock,
+    sse_helper: SSEStreamHelper,
+    exception: Exception,
+) -> None:
+    """Test terminal stream errors mark entities unavailable."""
+    mock_liebherr_client.get_device_state.side_effect = exception
+
+    await sse_helper.async_push()
+
+    state = hass.states.get("sensor.test_fridge_top_zone")
+    assert state is not None
+    assert state.state == STATE_UNAVAILABLE
+
+
+@pytest.mark.usefixtures("init_integration")
+async def test_repeated_stream_disconnect(
+    hass: HomeAssistant,
+    mock_liebherr_client: MagicMock,
+    sse_helper: SSEStreamHelper,
+) -> None:
+    """Test repeated stream disconnects leave entities unavailable."""
+    mock_liebherr_client.get_device_state.side_effect = LiebherrConnectionError(
+        "Connection failed"
+    )
+
+    await sse_helper.async_push()
+    await sse_helper.async_push()
+
+    state = hass.states.get("sensor.test_fridge_top_zone")
+    assert state is not None
+    assert state.state == STATE_UNAVAILABLE
+
+
 async def test_unload_entry(
     hass: HomeAssistant,
     mock_config_entry: MockConfigEntry,
@@ -354,7 +487,7 @@ async def test_stale_device_removal(
 
     # Simulate the new device being removed from the account.
     # Make get_device_state raise for new_device_id so we can detect
-    # if the stale coordinator is still polling after shutdown.
+    # if the stale coordinator is still consuming its stream after shutdown.
     mock_liebherr_client.get_devices.return_value = [MOCK_DEVICE]
 
     def _get_device_state_after_removal(device_id: str, **kw: Any) -> DeviceState:
@@ -378,11 +511,9 @@ async def test_stale_device_removal(
         (DOMAIN, "new_device_id"), mock_config_entry.entry_id
     )
 
-    # Advance past the coordinator update interval to confirm the stale
-    # coordinator is no longer polling (would raise AssertionError above)
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    # Trigger a stream event for the removed device to confirm the stale
+    # coordinator's stream task was cancelled (would raise AssertionError above)
+    await mock_liebherr_client._sse_helper.async_push("new_device_id")
 
     # Original device should still work
     assert hass.states.get("sensor.test_fridge_top_zone") is not None

tests/components/liebherr/test_light.py

@@ -27,7 +27,7 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
 
-from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE
+from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE, SSEStreamHelper
 
 from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
 
@@ -86,7 +86,6 @@ async def test_light_service_calls(
 ) -> None:
     """Test light turn on/off service calls."""
     entity_id = "light.test_fridge_presentation_light"
-    initial_call_count = mock_liebherr_client.get_device_state.call_count
 
     await hass.services.async_call(
         LIGHT_DOMAIN,
@@ -100,9 +99,6 @@ async def test_light_service_calls(
         target=expected_target,
     )
 
-    # Verify coordinator refresh was triggered
-    assert mock_liebherr_client.get_device_state.call_count > initial_call_count
-
 
 @pytest.mark.usefixtures("init_integration")
 async def test_light_failure(
@@ -131,7 +127,7 @@ async def test_light_failure(
 async def test_light_when_control_missing(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test light entity behavior when control is removed."""
     entity_id = "light.test_fridge_presentation_light"
@@ -145,9 +141,7 @@ async def test_light_when_control_missing(
         device=MOCK_DEVICE, controls=[]
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_reconnect()
 
     state = hass.states.get(entity_id)
     assert state is not None
@@ -167,13 +161,13 @@ async def test_light_when_control_missing(
 async def test_light_state_updates(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
     value: int | None,
     max_value: int,
     expected_state: str,
     expected_brightness: int | None,
 ) -> None:
-    """Test light entity state after coordinator update."""
+    """Test light entity state after a stream update."""
     entity_id = "light.test_fridge_presentation_light"
 
     mock_liebherr_client.get_device_state.side_effect = lambda *a, **kw: DeviceState(
@@ -188,9 +182,7 @@ async def test_light_state_updates(
         ],
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_push()
 
     state = hass.states.get(entity_id)
     assert state is not None

tests/components/liebherr/test_number.py

@@ -1,10 +1,8 @@
 """Test the Liebherr number platform."""
 
 import copy
-from datetime import timedelta
 from unittest.mock import MagicMock, patch
 
-from freezegun.api import FrozenDateTimeFactory
 from pyliebherrhomeapi import (
     Device,
     DeviceState,
@@ -30,9 +28,9 @@ from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
 from homeassistant.helpers import entity_registry as er
 from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
 
-from .conftest import MOCK_DEVICE
+from .conftest import MOCK_DEVICE, SSEStreamHelper
 
-from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
+from tests.common import MockConfigEntry, snapshot_platform
 
 
 @pytest.fixture
@@ -121,8 +119,6 @@ async def test_set_temperature(
     """Test setting the temperature."""
     entity_id = "number.test_fridge_top_zone_setpoint"
 
-    initial_call_count = mock_liebherr_client.get_device_state.call_count
-
     await hass.services.async_call(
         NUMBER_DOMAIN,
         SERVICE_SET_VALUE,
@@ -137,9 +133,6 @@ async def test_set_temperature(
         unit=TemperatureUnit.CELSIUS,
     )
 
-    # Verify coordinator refresh was triggered
-    assert mock_liebherr_client.get_device_state.call_count > initial_call_count
-
 
 @pytest.mark.usefixtures("init_integration")
 async def test_set_temperature_after_unit_conversion(
@@ -220,7 +213,7 @@ async def test_set_temperature_failure(
 async def test_number_when_control_missing(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test number entity behavior when temperature control is removed."""
     entity_id = "number.test_fridge_top_zone_setpoint"
@@ -238,10 +231,7 @@ async def test_number_when_control_missing(
         device=MOCK_DEVICE, controls=[]
     )
 
-    # Advance time to trigger coordinator refresh
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_reconnect()
 
     # State should be unavailable
     state = hass.states.get(entity_id)

tests/components/liebherr/test_select.py

@@ -2,11 +2,9 @@
 
 import copy
 import dataclasses
-from datetime import timedelta
 from typing import Any
 from unittest.mock import MagicMock, patch
 
-from freezegun.api import FrozenDateTimeFactory
 from pyliebherrhomeapi import (
     BioFreshPlusMode,
     Device,
@@ -39,9 +37,9 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
 
-from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE
+from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE, SSEStreamHelper
 
-from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
+from tests.common import MockConfigEntry, snapshot_platform
 
 
 @pytest.fixture
@@ -131,8 +129,6 @@ async def test_select_service_calls(
     kwargs: dict[str, Any],
 ) -> None:
     """Test select option service calls."""
-    initial_call_count = mock_liebherr_client.get_device_state.call_count
-
     await hass.services.async_call(
         SELECT_DOMAIN,
         SERVICE_SELECT_OPTION,
@@ -142,9 +138,6 @@ async def test_select_service_calls(
 
     getattr(mock_liebherr_client, method).assert_called_once_with(**kwargs)
 
-    # Verify coordinator refresh was triggered
-    assert mock_liebherr_client.get_device_state.call_count > initial_call_count
-
 
 @pytest.mark.parametrize(
     ("entity_id", "method", "option"),
@@ -187,7 +180,7 @@ async def test_select_failure(
 async def test_select_when_control_missing(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test select entity behavior when control is removed."""
     entity_id = "select.test_fridge_bottom_zone_icemaker"
@@ -201,9 +194,7 @@ async def test_select_when_control_missing(
         device=MOCK_DEVICE, controls=[]
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_reconnect()
 
     state = hass.states.get(entity_id)
     assert state is not None
@@ -272,7 +263,7 @@ async def test_single_zone_select(
 async def test_select_current_option_none_mode(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test select entity state when control mode returns None."""
     entity_id = "select.test_fridge_top_zone_hydrobreeze"
@@ -296,9 +287,7 @@ async def test_select_current_option_none_mode(
         state_with_none_mode
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_push()
 
     state = hass.states.get(entity_id)
     assert state is not None

tests/components/liebherr/test_sensor.py

@@ -1,10 +1,9 @@
 """Test the Liebherr sensor platform."""
 
 import copy
-from datetime import timedelta
+from dataclasses import replace
 from unittest.mock import MagicMock, patch
 
-from freezegun.api import FrozenDateTimeFactory
 from pyliebherrhomeapi import (
     Device,
     DeviceState,
@@ -26,10 +25,11 @@ from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import STATE_UNAVAILABLE, Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.event import async_track_state_change_event
 
-from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE
+from .conftest import MOCK_DEVICE, MOCK_DEVICE_STATE, SSEStreamHelper
 
-from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
+from tests.common import MockConfigEntry, snapshot_platform
 
 
 @pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
@@ -153,10 +153,10 @@ async def test_multi_zone_with_none_position(
 async def test_sensor_update_failure(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
     exception: Exception,
 ) -> None:
-    """Test sensor becomes unavailable when coordinator update fails."""
+    """Test sensor becomes unavailable when the stream disconnects."""
     entity_id = "sensor.test_fridge_top_zone"
 
     # Initial state should be available with value
@@ -164,32 +164,42 @@ async def test_sensor_update_failure(
     assert state is not None
     assert state.state == "5"
 
-    # Simulate update error
+    # Simulate a stream disconnect via the mocked ``get_device_state``.
     mock_liebherr_client.get_device_state.side_effect = exception
 
-    # Advance time to trigger coordinator refresh (60 second interval)
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_push()
 
     # Sensor should now be unavailable
     state = hass.states.get(entity_id)
     assert state is not None
     assert state.state == STATE_UNAVAILABLE
 
-    # Simulate recovery
-    mock_liebherr_client.get_device_state.side_effect = lambda *a, **kw: copy.deepcopy(
-        MOCK_DEVICE_STATE
+    # Simulate reconnect with a changed top-zone temperature.
+    fresh_state = replace(
+        MOCK_DEVICE_STATE,
+        controls=[
+            replace(control, value=6)
+            if isinstance(control, TemperatureControl) and control.zone_id == 1
+            else control
+            for control in MOCK_DEVICE_STATE.controls
+        ],
+    )
+    mock_liebherr_client.get_device_state.side_effect = lambda *a, **kw: fresh_state
+    reconnect_states: list[str] = []
+    unsubscribe = async_track_state_change_event(
+        hass,
+        entity_id,
+        lambda event: reconnect_states.append(event.data["new_state"].state),
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_reconnect()
+    unsubscribe()
 
-    # Sensor should recover
+    # Sensor should recover directly with fresh data, without exposing stale data.
     state = hass.states.get(entity_id)
     assert state is not None
-    assert state.state == "5"
+    assert state.state == "6"
+    assert reconnect_states == ["6"]
 
 
 @pytest.mark.usefixtures("entity_registry_enabled_by_default", "init_integration")
@@ -197,7 +207,7 @@ async def test_sensor_update_auth_failure_triggers_reauth(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
     mock_config_entry: MockConfigEntry,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test authentication error triggers reauth flow."""
     entity_id = "sensor.test_fridge_top_zone"
@@ -207,17 +217,13 @@ async def test_sensor_update_auth_failure_triggers_reauth(
     assert state is not None
     assert state.state == "5"
 
-    # Simulate auth error
+    # Simulate auth error from the SSE stream
     mock_liebherr_client.get_device_state.side_effect = LiebherrAuthenticationError(
         "API key revoked"
     )
 
-    # Advance time to trigger coordinator refresh (60 second interval)
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_push()
 
-    # Sensor should now be unavailable
     state = hass.states.get(entity_id)
     assert state is not None
     assert state.state == STATE_UNAVAILABLE
@@ -236,7 +242,7 @@ async def test_sensor_unavailable_when_control_missing(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
     mock_config_entry: MockConfigEntry,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test sensor becomes unavailable when control is removed."""
     entity_id = "sensor.test_fridge_top_zone"
@@ -246,15 +252,13 @@ async def test_sensor_unavailable_when_control_missing(
     assert state is not None
     assert state.state == "5"
 
-    # Device stops reporting controls (e.g., zone removed or API issue)
+    # Device stops reporting controls (e.g., zone removed or API issue).
+    # Only observable via a full-state event on stream reconnect.
     mock_liebherr_client.get_device_state.side_effect = lambda *a, **kw: DeviceState(
         device=MOCK_DEVICE, controls=[]
     )
 
-    # Advance time to trigger coordinator refresh
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_reconnect()
 
     # Sensor should now be unavailable
     state = hass.states.get(entity_id)

tests/components/liebherr/test_switch.py

@@ -1,11 +1,9 @@
 """Test the Liebherr switch platform."""
 
 import copy
-from datetime import timedelta
 from typing import Any
 from unittest.mock import MagicMock, patch
 
-from freezegun.api import FrozenDateTimeFactory
 from pyliebherrhomeapi import (
     Device,
     DeviceState,
@@ -32,9 +30,9 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
 
-from .conftest import MOCK_DEVICE
+from .conftest import MOCK_DEVICE, SSEStreamHelper
 
-from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
+from tests.common import MockConfigEntry, snapshot_platform
 
 
 @pytest.fixture
@@ -104,8 +102,6 @@ async def test_switch_service_calls(
     kwargs: dict[str, Any],
 ) -> None:
     """Test switch turn on/off service calls."""
-    initial_call_count = mock_liebherr_client.get_device_state.call_count
-
     await hass.services.async_call(
         SWITCH_DOMAIN,
         service,
@@ -115,9 +111,6 @@ async def test_switch_service_calls(
 
     getattr(mock_liebherr_client, method).assert_called_once_with(**kwargs)
 
-    # Verify coordinator refresh was triggered
-    assert mock_liebherr_client.get_device_state.call_count > initial_call_count
-
 
 @pytest.mark.parametrize(
     ("entity_id", "method"),
@@ -154,7 +147,7 @@ async def test_switch_failure(
 async def test_switch_when_control_missing(
     hass: HomeAssistant,
     mock_liebherr_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
+    sse_helper: SSEStreamHelper,
 ) -> None:
     """Test switch entity behavior when toggle control is removed."""
     entity_id = "switch.test_fridge_top_zone_supercool"
@@ -168,9 +161,7 @@ async def test_switch_when_control_missing(
         device=MOCK_DEVICE, controls=[]
     )
 
-    freezer.tick(timedelta(seconds=61))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
+    await sse_helper.async_reconnect()
 
     state = hass.states.get(entity_id)
     assert state is not None