diff --git a/homeassistant/components/liebherr/__init__.py b/homeassistant/components/liebherr/__init__.py index 577cd1d737ba..2fb8bdf096bb 100644 --- a/homeassistant/components/liebherr/__init__.py +++ b/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( diff --git a/homeassistant/components/liebherr/const.py b/homeassistant/components/liebherr/const.py index ceffd331d66a..a528423ce5ab 100644 --- a/homeassistant/components/liebherr/const.py +++ b/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) diff --git a/homeassistant/components/liebherr/coordinator.py b/homeassistant/components/liebherr/coordinator.py index 32887b08022a..7b0feb9fe7ef 100644 --- a/homeassistant/components/liebherr/coordinator.py +++ b/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() diff --git a/homeassistant/components/liebherr/entity.py b/homeassistant/components/liebherr/entity.py index 5a4fc8fcc0bc..b41ddf7f39c3 100644 --- a/homeassistant/components/liebherr/entity.py +++ b/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. diff --git a/homeassistant/components/liebherr/manifest.json b/homeassistant/components/liebherr/manifest.json index a5217c1030ef..ed92c20049af 100644 --- a/homeassistant/components/liebherr/manifest.json +++ b/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"], diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 9affa7ef0cc5..b98961014781 100644 --- a/homeassistant/generated/integrations.json +++ b/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", diff --git a/tests/components/liebherr/conftest.py b/tests/components/liebherr/conftest.py index 78a219ae9864..a75435430c98 100644 --- a/tests/components/liebherr/conftest.py +++ b/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.""" diff --git a/tests/components/liebherr/snapshots/test_diagnostics.ambr b/tests/components/liebherr/snapshots/test_diagnostics.ambr index 78082f32adce..eb7c6ed4b1c6 100644 --- a/tests/components/liebherr/snapshots/test_diagnostics.ambr +++ b/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([ diff --git a/tests/components/liebherr/test_cover.py b/tests/components/liebherr/test_cover.py index 06c2dc151c86..a3607d0ddb45 100644 --- a/tests/components/liebherr/test_cover.py +++ b/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 diff --git a/tests/components/liebherr/test_init.py b/tests/components/liebherr/test_init.py index 6c9faeb02c20..a78ae05e2478 100644 --- a/tests/components/liebherr/test_init.py +++ b/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 diff --git a/tests/components/liebherr/test_light.py b/tests/components/liebherr/test_light.py index 9b916da96979..b4197a01b4c1 100644 --- a/tests/components/liebherr/test_light.py +++ b/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 diff --git a/tests/components/liebherr/test_number.py b/tests/components/liebherr/test_number.py index 03a88a9f80e5..eb326ec0a170 100644 --- a/tests/components/liebherr/test_number.py +++ b/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) diff --git a/tests/components/liebherr/test_select.py b/tests/components/liebherr/test_select.py index eca5865dc988..7a4d95670011 100644 --- a/tests/components/liebherr/test_select.py +++ b/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 diff --git a/tests/components/liebherr/test_sensor.py b/tests/components/liebherr/test_sensor.py index 39538944b75c..51afcbfacc6d 100644 --- a/tests/components/liebherr/test_sensor.py +++ b/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) diff --git a/tests/components/liebherr/test_switch.py b/tests/components/liebherr/test_switch.py index 51ed0d6948de..e9152540d08f 100644 --- a/tests/components/liebherr/test_switch.py +++ b/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