diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py index 342c4f7f429..8fbe0572379 100644 --- a/homeassistant/components/samsungtv/bridge.py +++ b/homeassistant/components/samsungtv/bridge.py @@ -300,6 +300,7 @@ class SamsungTVWSBridge(SamsungTVBridge): self.token = token self._rest_api: SamsungTVAsyncRest | None = None self._app_list: dict[str, str] | None = None + self._device_info: dict[str, Any] | None = None self._remote: SamsungTVWSAsyncRemote | None = None self._remote_lock = asyncio.Lock() @@ -323,8 +324,20 @@ class SamsungTVWSBridge(SamsungTVBridge): LOGGER.debug("Generated app list: %s", self._app_list) return self._app_list + def _get_device_spec(self, key: str) -> Any | None: + """Check if a flag exists in latest device info.""" + if not ((info := self._device_info) and (device := info.get("device"))): + return None + return device.get(key) + async def async_is_on(self) -> bool: """Tells if the TV is on.""" + if self._get_device_spec("PowerState") is not None: + LOGGER.debug("Checking if TV %s is on using device info", self.host) + # Ensure we get an updated value + info = await self.async_device_info() + return info is not None and info["device"]["PowerState"] == "on" + LOGGER.debug("Checking if TV %s is on using websocket", self.host) if remote := await self._async_get_remote(): return remote.is_alive() return False @@ -383,6 +396,7 @@ class SamsungTVWSBridge(SamsungTVBridge): with contextlib.suppress(HttpApiError, AsyncioTimeoutError): device_info: dict[str, Any] = await self._rest_api.rest_device_info() LOGGER.debug("Device info on %s is: %s", self.host, device_info) + self._device_info = device_info return device_info return None @@ -453,6 +467,9 @@ class SamsungTVWSBridge(SamsungTVBridge): LOGGER.debug( "Created SamsungTVWSBridge for %s (%s)", CONF_NAME, self.host ) + if self._device_info is None: + # Initialise device info on first connect + await self.async_device_info() if self.token != self._remote.token: LOGGER.debug( "SamsungTVWSBridge has provided a new token %s", diff --git a/tests/components/samsungtv/test_media_player.py b/tests/components/samsungtv/test_media_player.py index 4c324abe4d8..af35c40d074 100644 --- a/tests/components/samsungtv/test_media_player.py +++ b/tests/components/samsungtv/test_media_player.py @@ -1,5 +1,6 @@ """Tests for samsungtv component.""" import asyncio +from copy import deepcopy from datetime import datetime, timedelta import logging from unittest.mock import DEFAULT as DEFAULT_MOCK, AsyncMock, Mock, call, patch @@ -7,7 +8,7 @@ from unittest.mock import DEFAULT as DEFAULT_MOCK, AsyncMock, Mock, call, patch import pytest from samsungctl import exceptions from samsungtvws.async_remote import SamsungTVWSAsyncRemote -from samsungtvws.exceptions import ConnectionFailure +from samsungtvws.exceptions import ConnectionFailure, HttpApiError from samsungtvws.remote import ChannelEmitCommand, SendRemoteKey from websockets.exceptions import WebSocketException @@ -267,11 +268,14 @@ async def test_update_off(hass: HomeAssistant, mock_now: datetime) -> None: assert state.state == STATE_OFF -async def test_update_off_ws( - hass: HomeAssistant, remotews: Mock, mock_now: datetime +async def test_update_off_ws_no_power_state( + hass: HomeAssistant, remotews: Mock, rest_api: Mock, mock_now: datetime ) -> None: """Testing update tv off.""" await setup_samsungtv(hass, MOCK_CONFIGWS) + # device_info should only get called once, as part of the setup + rest_api.rest_device_info.assert_called_once() + rest_api.rest_device_info.reset_mock() state = hass.states.get(ENTITY_ID) assert state.state == STATE_ON @@ -286,6 +290,71 @@ async def test_update_off_ws( state = hass.states.get(ENTITY_ID) assert state.state == STATE_OFF + rest_api.rest_device_info.assert_not_called() + + +@pytest.mark.usefixtures("remotews") +async def test_update_off_ws_with_power_state( + hass: HomeAssistant, remotews: Mock, rest_api: Mock, mock_now: datetime +) -> None: + """Testing update tv off.""" + with patch.object( + rest_api, "rest_device_info", side_effect=HttpApiError + ) as mock_device_info, patch.object( + remotews, "start_listening", side_effect=WebSocketException("Boom") + ) as mock_start_listening: + await setup_samsungtv(hass, MOCK_CONFIGWS) + + mock_device_info.assert_called_once() + mock_start_listening.assert_called_once() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_OFF + + # First update uses start_listening once, and initialises device_info + device_info = deepcopy(rest_api.rest_device_info.return_value) + device_info["device"]["PowerState"] = "on" + rest_api.rest_device_info.return_value = device_info + next_update = mock_now + timedelta(minutes=1) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + + remotews.start_listening.assert_called_once() + rest_api.rest_device_info.assert_called_once() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + # After initial update, start_listening shouldn't be called + remotews.start_listening.reset_mock() + + # Second update uses device_info(ON) + rest_api.rest_device_info.reset_mock() + next_update = mock_now + timedelta(minutes=2) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + + rest_api.rest_device_info.assert_called_once() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + # Third update uses device_info (OFF) + rest_api.rest_device_info.reset_mock() + device_info["device"]["PowerState"] = "off" + next_update = mock_now + timedelta(minutes=3) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + + rest_api.rest_device_info.assert_called_once() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_OFF + + remotews.start_listening.assert_not_called() @pytest.mark.usefixtures("remote")