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

Refactor run app in SamsungTV (#67616)

Browse Source 
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2022-03-05 01:37:27 +01:00 committed by GitHub
parent acd906dfab
commit c5f7e7d1b0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 53 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -11,6 +11,7 @@ from samsungctl import Remote
from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledResponse
from samsungtvws.async_remote import SamsungTVWSAsyncRemote
from samsungtvws.async_rest import SamsungTVAsyncRest
from samsungtvws.command import SamsungTVCommand
from samsungtvws.exceptions import ConnectionFailure, HttpApiError
from samsungtvws.remote import ChannelEmitCommand, SendRemoteKey
from websockets.exceptions import WebSocketException
@ -128,7 +129,7 @@ class SamsungTVBridge(ABC):
"""Tells if the TV is on."""
@abstractmethod
async def async_send_key(self, key: str, key_type: str | None = None) -> None:
async def async_send_key(self, key: str) -> None:
"""Send a key to the tv and handles exceptions."""
@abstractmethod
@ -237,7 +238,7 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
pass
return self._remote
async def async_send_key(self, key: str, key_type: str | None = None) -> None:
async def async_send_key(self, key: str) -> None:
"""Send the key using legacy protocol."""
await self.hass.async_add_executor_job(self._send_key, key)
@ -388,22 +389,25 @@ class SamsungTVWSBridge(SamsungTVBridge):
return None
async def async_send_key(self, key: str, key_type: str | None = None) -> None:
async def async_launch_app(self, app_id: str) -> None:
"""Send the launch_app command using websocket protocol."""
await self._async_send_command(ChannelEmitCommand.launch_app(app_id))
async def async_send_key(self, key: str) -> None:
"""Send the key using websocket protocol."""
if key == "KEY_POWEROFF":
key = "KEY_POWER"
await self._async_send_command(SendRemoteKey.click(key))
async def _async_send_command(self, command: SamsungTVCommand) -> None:
"""Send the commands using websocket protocol."""
try:
# recreate connection if connection was dead
retry_count = 1
for _ in range(retry_count + 1):
try:
if remote := await self._async_get_remote():
if key_type == "run_app":
await remote.send_command(
ChannelEmitCommand.launch_app(key)
)
else:
await remote.send_command(SendRemoteKey.click(key))
await remote.send_command(command)
break
except (
BrokenPipeError,

18
homeassistant/components/samsungtv/media_player.py
View File

@ -173,12 +173,20 @@ class SamsungTVDevice(MediaPlayerEntity):
if self._app_list is not None:
self._attr_source_list.extend(self._app_list)
async def _async_send_key(self, key: str, key_type: str | None = None) -> None:
async def _async_launch_app(self, app_id: str) -> None:
"""Send launch_app to the tv."""
if self._power_off_in_progress():
LOGGER.info("TV is powering off, not sending launch_app command")
return
assert isinstance(self._bridge, SamsungTVWSBridge)
await self._bridge.async_launch_app(app_id)
async def _async_send_key(self, key: str) -> None:
"""Send a key to the tv and handles exceptions."""
if self._power_off_in_progress() and key != "KEY_POWEROFF":
LOGGER.info("TV is powering off, not sending command: %s", key)
LOGGER.info("TV is powering off, not sending key: %s", key)
return
await self._bridge.async_send_key(key, key_type)
await self._bridge.async_send_key(key)
def _power_off_in_progress(self) -> bool:
return (
@ -248,7 +256,7 @@ class SamsungTVDevice(MediaPlayerEntity):
) -> None:
"""Support changing a channel."""
if media_type == MEDIA_TYPE_APP:
await self._async_send_key(media_id, "run_app")
await self._async_launch_app(media_id)
return
if media_type != MEDIA_TYPE_CHANNEL:
@ -284,7 +292,7 @@ class SamsungTVDevice(MediaPlayerEntity):
async def async_select_source(self, source: str) -> None:
"""Select input source."""
if self._app_list and source in self._app_list:
await self._async_send_key(self._app_list[source], "run_app")
await self._async_launch_app(self._app_list[source])
return
if source in SOURCES:

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

@ -624,29 +624,41 @@ async def test_device_class(hass: HomeAssistant) -> None:
assert state.attributes[ATTR_DEVICE_CLASS] is MediaPlayerDeviceClass.TV.value
async def test_turn_off_websocket(hass: HomeAssistant, remotews: Mock) -> None:
async def test_turn_off_websocket(
hass: HomeAssistant, remotews: Mock, caplog: pytest.LogCaptureFixture
) -> None:
"""Test for turn_off."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[OSError("Boom"), DEFAULT_MOCK],
):
await setup_samsungtv(hass, MOCK_CONFIGWS)
remotews.send_command.reset_mock()
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key called
assert remotews.send_command.call_count == 1
command = remotews.send_command.call_args_list[0].args[0]
assert isinstance(command, SendRemoteKey)
assert command.params["DataOfCmd"] == "KEY_POWER"
remotews.send_command.reset_mock()
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key not called
assert remotews.send_command.call_count == 1
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key called
assert remotews.send_command.call_count == 1
command = remotews.send_command.call_args_list[0].args[0]
assert isinstance(command, SendRemoteKey)
assert command.params["DataOfCmd"] == "KEY_POWER"
# commands not sent : power off in progress
remotews.send_command.reset_mock()
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
assert "TV is powering off, not sending key: KEY_VOLUP" in caplog.text
assert await hass.services.async_call(
DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "Deezer"},
True,
)
assert "TV is powering off, not sending launch_app command" in caplog.text
remotews.send_command.assert_not_called()
async def test_turn_off_legacy(hass: HomeAssistant, remote: Mock) -> None: