github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-08-06 09:34:49 +02:00
Code Releases Activity

Fix Samsung TV state when the device is turned off (#67541)

Browse Source 
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2022-03-03 19:06:33 +01:00 committed by GitHub
parent 1a78e18eeb
commit 74483d2669
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 89 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
homeassistant/components/samsungtv/bridge.py
View File

@ -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",

75
tests/components/samsungtv/test_media_player.py
View File

@ -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")