jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 01:38:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add translated action exceptions to LG webOS TV (#136397)

Browse Source 
* Add translated action exceptions to LG webOS TV

* Apply suggestions from code review

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Shay Levy 2025-01-24 02:07:24 +02:00 committed by GitHub
parent 3bbcd37ec8
commit fe67069c91
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 213 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
homeassistant/components/webostv/__init__.py
View File

@ -99,24 +99,6 @@ async def async_update_options(hass: HomeAssistant, entry: WebOsTvConfigEntry) -
await hass.config_entries.async_reload(entry.entry_id)
async def async_control_connect(
hass: HomeAssistant, host: str, key: str | None
) -> WebOsClient:
"""LG Connection."""
client = WebOsClient(
host,
key,
client_session=async_get_clientsession(hass),
)
try:
await client.connect()
except WebOsTvPairError:
_LOGGER.warning("Connected to LG webOS TV %s but not paired", host)
raise
return client
def update_client_key(
hass: HomeAssistant, entry: ConfigEntry, client: WebOsClient
) -> None:

35
homeassistant/components/webostv/config_flow.py
View File

@ -6,22 +6,23 @@ from collections.abc import Mapping
from typing import Any, Self
from urllib.parse import urlparse
from aiowebostv import WebOsTvPairError
from aiowebostv import WebOsClient, WebOsTvPairError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult, OptionsFlow
from homeassistant.const import CONF_CLIENT_SECRET, CONF_HOST
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_UDN,
SsdpServiceInfo,
)
from . import WebOsTvConfigEntry, async_control_connect
from . import WebOsTvConfigEntry
from .const import CONF_SOURCES, DEFAULT_NAME, DOMAIN, WEBOSTV_EXCEPTIONS
from .helpers import async_get_sources
from .helpers import get_sources
DATA_SCHEMA = vol.Schema(
{
@ -31,6 +32,21 @@ DATA_SCHEMA = vol.Schema(
)
async def async_control_connect(
hass: HomeAssistant, host: str, key: str | None
) -> WebOsClient:
"""Create LG WebOS client and connect to the TV."""
client = WebOsClient(
host,
key,
client_session=async_get_clientsession(hass),
)
await client.connect()
return client
class FlowHandler(ConfigFlow, domain=DOMAIN):
"""WebosTV configuration flow."""
@ -195,9 +211,14 @@ class OptionsFlowHandler(OptionsFlow):
options_input = {CONF_SOURCES: user_input[CONF_SOURCES]}
return self.async_create_entry(title="", data=options_input)
# Get sources
sources_list = await async_get_sources(self.hass, self.host, self.key)
if not sources_list:
errors["base"] = "cannot_retrieve"
sources_list = []
try:
client = await async_control_connect(self.hass, self.host, self.key)
sources_list = get_sources(client)
except WebOsTvPairError:
errors["base"] = "error_pairing"
except WEBOSTV_EXCEPTIONS:
errors["base"] = "cannot_connect"
option_sources = self.config_entry.options.get(CONF_SOURCES, [])
sources = [s for s in option_sources if s in sources_list]

8
homeassistant/components/webostv/device_trigger.py
View File

@ -14,7 +14,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
from . import trigger
from . import DOMAIN, trigger
from .helpers import (
async_get_client_by_device_entry,
async_get_device_entry_by_device_id,
@ -75,4 +75,8 @@ async def async_attach_trigger(
hass, trigger_config, action, trigger_info
)
raise HomeAssistantError(f"Unhandled trigger type {trigger_type}")
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unhandled_trigger_type",
translation_placeholders={"trigger_type": trigger_type},
)

11
homeassistant/components/webostv/helpers.py
View File

@ -9,8 +9,8 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.device_registry import DeviceEntry
from . import WebOsTvConfigEntry, async_control_connect
from .const import DOMAIN, LIVE_TV_APP_ID, WEBOSTV_EXCEPTIONS
from . import WebOsTvConfigEntry
from .const import DOMAIN, LIVE_TV_APP_ID
@callback
@ -72,13 +72,8 @@ def async_get_client_by_device_entry(
)
async def async_get_sources(hass: HomeAssistant, host: str, key: str) -> list[str]:
def get_sources(client: WebOsClient) -> list[str]:
"""Construct sources list."""
try:
client = await async_control_connect(hass, host, key)
except WEBOSTV_EXCEPTIONS:
return []
sources = []
found_live_tv = False
for app in client.apps.values():

30
homeassistant/components/webostv/media_player.py
View File

@ -106,21 +106,27 @@ def cmd[_T: LgWebOSMediaPlayerEntity, **_P](
@wraps(func)
async def cmd_wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> None:
"""Wrap all command methods."""
if self.state is MediaPlayerState.OFF:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="device_off",
translation_placeholders={
"name": str(self._entry.title),
"func": func.__name__,
},
)
try:
await func(self, *args, **kwargs)
except WEBOSTV_EXCEPTIONS as exc:
if self.state != MediaPlayerState.OFF:
except WEBOSTV_EXCEPTIONS as error:
raise HomeAssistantError(
f"Error calling {func.__name__} on entity {self.entity_id},"
f" state:{self.state}"
) from exc
_LOGGER.warning(
"Error calling %s on entity %s, state:%s, error: %r",
func.__name__,
self.entity_id,
self.state,
exc,
)
translation_domain=DOMAIN,
translation_key="communication_error",
translation_placeholders={
"name": str(self._entry.title),
"func": func.__name__,
"error": str(error),
},
) from error
return cmd_wrapper

57
homeassistant/components/webostv/notify.py
View File

@ -2,19 +2,18 @@
from __future__ import annotations
import logging
from typing import Any
from aiowebostv import WebOsClient, WebOsTvPairError
from aiowebostv import WebOsClient
from homeassistant.components.notify import ATTR_DATA, BaseNotificationService
from homeassistant.const import ATTR_ICON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import ATTR_CONFIG_ENTRY_ID, WEBOSTV_EXCEPTIONS
_LOGGER = logging.getLogger(__name__)
from . import WebOsTvConfigEntry
from .const import ATTR_CONFIG_ENTRY_ID, DOMAIN, WEBOSTV_EXCEPTIONS
PARALLEL_UPDATES = 0
@ -34,28 +33,48 @@ async def async_get_service(
)
assert config_entry is not None
return LgWebOSNotificationService(config_entry.runtime_data)
return LgWebOSNotificationService(config_entry)
class LgWebOSNotificationService(BaseNotificationService):
"""Implement the notification service for LG WebOS TV."""
def __init__(self, client: WebOsClient) -> None:
def __init__(self, entry: WebOsTvConfigEntry) -> None:
"""Initialize the service."""
self._client = client
self._entry = entry
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a message to the tv."""
try:
if not self._client.is_connected():
await self._client.connect()
client: WebOsClient = self._entry.runtime_data
data = kwargs[ATTR_DATA]
icon_path = data.get(ATTR_ICON) if data else None
await self._client.send_message(message, icon_path=icon_path)
except WebOsTvPairError:
_LOGGER.error("Pairing with TV failed")
except FileNotFoundError:
_LOGGER.error("Icon %s not found", icon_path)
except WEBOSTV_EXCEPTIONS:
_LOGGER.error("TV unreachable")
if not client.is_on:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="notify_device_off",
translation_placeholders={
"name": str(self._entry.title),
"func": __name__,
},
)
try:
await client.send_message(message, icon_path=icon_path)
except FileNotFoundError as error:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="notify_icon_not_found",
translation_placeholders={
"name": str(self._entry.title),
"icon_path": str(icon_path),
},
) from error
except WEBOSTV_EXCEPTIONS as error:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="notify_communication_error",
translation_placeholders={
"name": str(self._entry.title),
"error": str(error),
},
) from error

2
homeassistant/components/webostv/quality_scale.yaml
View File

@ -58,7 +58,7 @@ rules:
entity-translations:
status: exempt
comment: There are no entities to translate.
exception-translations: todo
exception-translations: done
icon-translations:
status: exempt
comment: The only entity can use the device class.

23
homeassistant/components/webostv/strings.json
View File

@ -54,7 +54,8 @@
}
},
"error": {
"cannot_retrieve": "Unable to retrieve the list of sources. Make sure device is switched on"
"cannot_connect": "[%key:component::webostv::config::error::cannot_connect%]",
"error_pairing": "[%key:component::webostv::config::error::error_pairing%]"
}
},
"device_automation": {
@ -109,5 +110,25 @@
}
}
}
},
"exceptions": {
"device_off": {
"message": "Error calling {func} for device {name}: Device is off and cannot be controlled."
},
"communication_error": {
"message": "Communication error while calling {func} for device {name}: {error}"
},
"notify_device_off": {
"message": "Error sending notification to device {name}: Device is off and cannot be controlled."
},
"notify_icon_not_found": {
"message": "Icon {icon_path} not found when sending notification for device {name}"
},
"notify_communication_error": {
"message": "Communication error while sending notification to device {name}: {error}"
},
"unhandled_trigger_type": {
"message": "Unhandled trigger type: {trigger_type}"
}
}
}

10
tests/components/webostv/conftest.py
View File

@ -30,9 +30,15 @@ def mock_setup_entry() -> Generator[AsyncMock]:
@pytest.fixture(name="client")
def client_fixture():
"""Patch of client library for tests."""
with patch(
with (
patch(
"homeassistant.components.webostv.WebOsClient", autospec=True
) as mock_client_class:
) as mock_client_class,
patch(
"homeassistant.components.webostv.config_flow.WebOsClient",
new=mock_client_class,
),
):
client = mock_client_class.return_value
client.hello_info = {"deviceUUID": FAKE_UUID}
client.software_info = {"major_ver": "major", "minor_ver": "minor"}

17
tests/components/webostv/test_config_flow.py
View File

@ -103,16 +103,25 @@ async def test_options_flow_live_tv_in_apps(
assert result["data"][CONF_SOURCES] == ["Live TV", "Input01", "Input02"]
async def test_options_flow_cannot_retrieve(hass: HomeAssistant, client) -> None:
"""Test options config flow cannot retrieve sources."""
@pytest.mark.parametrize(
("side_effect", "error"),
[
(WebOsTvPairError, "error_pairing"),
(ConnectionResetError, "cannot_connect"),
],
)
async def test_options_flow_errors(
hass: HomeAssistant, client, side_effect, error
) -> None:
"""Test options config flow errors."""
entry = await setup_webostv(hass)
client.connect.side_effect = ConnectionResetError
client.connect.side_effect = side_effect
result = await hass.config_entries.options.async_init(entry.entry_id)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_retrieve"}
assert result["errors"] == {"base": error}
# recover
client.connect.side_effect = None

2
tests/components/webostv/test_device_trigger.py
View File

@ -111,7 +111,7 @@ async def test_invalid_trigger_raises(
await setup_webostv(hass)
# Test wrong trigger platform type
with pytest.raises(HomeAssistantError):
with pytest.raises(HomeAssistantError, match="Unhandled trigger type: wrong.type"):
await device_trigger.async_attach_trigger(
hass, {"type": "wrong.type", "device_id": "invalid_device_id"}, None, {}
)

53
tests/components/webostv/test_media_player.py
View File

@ -482,35 +482,44 @@ async def test_client_key_update_on_connect(
assert config_entry.data[CONF_CLIENT_SECRET] == client.client_key
@pytest.mark.parametrize(
("is_on", "exception", "error_message"),
[
(
True,
WebOsTvCommandError("Some error"),
f"Communication error while calling async_media_play for device {TV_NAME}: Some error",
),
(
True,
WebOsTvCommandError("Some other error"),
f"Communication error while calling async_media_play for device {TV_NAME}: Some other error",
),
(
False,
None,
f"Error calling async_media_play for device {TV_NAME}: Device is off and cannot be controlled",
),
],
)
async def test_control_error_handling(
hass: HomeAssistant, client, caplog: pytest.LogCaptureFixture
hass: HomeAssistant,
client,
is_on: bool,
exception: Exception,
error_message: str,
) -> None:
"""Test control errors handling."""
await setup_webostv(hass)
client.play.side_effect = WebOsTvCommandError
data = {ATTR_ENTITY_ID: ENTITY_ID}
# Device on, raise HomeAssistantError
with pytest.raises(HomeAssistantError) as exc:
await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PLAY, data, True)
assert (
str(exc.value)
== f"Error calling async_media_play on entity {ENTITY_ID}, state:on"
)
assert client.play.call_count == 1
# Device off, log a warning
client.is_on = False
client.play.side_effect = TimeoutError
client.play.side_effect = exception
client.is_on = is_on
await client.mock_state_update()
data = {ATTR_ENTITY_ID: ENTITY_ID}
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PLAY, data, True)
assert client.play.call_count == 2
assert (
f"Error calling async_media_play on entity {ENTITY_ID}, state:off, error:"
" TimeoutError()" in caplog.text
)
assert client.play.call_count == int(is_on)
async def test_supported_features(hass: HomeAssistant, client) -> None:

91
tests/components/webostv/test_notify.py
View File

@ -2,7 +2,7 @@
from unittest.mock import call
from aiowebostv import WebOsTvPairError
from aiowebostv import WebOsTvCommandError
import pytest
from homeassistant.components.notify import (
@ -13,6 +13,7 @@ from homeassistant.components.notify import (
from homeassistant.components.webostv import DOMAIN
from homeassistant.const import ATTR_ICON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from homeassistant.util import slugify
@ -74,69 +75,41 @@ async def test_notify(hass: HomeAssistant, client) -> None:
)
async def test_notify_not_connected(hass: HomeAssistant, client) -> None:
"""Test sending a message when client is not connected."""
await setup_webostv(hass)
assert hass.services.has_service(NOTIFY_DOMAIN, SERVICE_NAME)
client.is_connected.return_value = False
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_NAME,
{
ATTR_MESSAGE: MESSAGE,
ATTR_DATA: {
ATTR_ICON: ICON_PATH,
},
},
blocking=True,
)
assert client.mock_calls[0] == call.connect()
assert client.connect.call_count == 2
client.send_message.assert_called_with(MESSAGE, icon_path=ICON_PATH)
async def test_icon_not_found(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, client
) -> None:
"""Test notify icon not found error."""
await setup_webostv(hass)
assert hass.services.has_service(NOTIFY_DOMAIN, SERVICE_NAME)
client.send_message.side_effect = FileNotFoundError
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_NAME,
{
ATTR_MESSAGE: MESSAGE,
ATTR_DATA: {
ATTR_ICON: ICON_PATH,
},
},
blocking=True,
)
assert client.mock_calls[0] == call.connect()
assert client.connect.call_count == 1
client.send_message.assert_called_with(MESSAGE, icon_path=ICON_PATH)
assert f"Icon {ICON_PATH} not found" in caplog.text
@pytest.mark.parametrize(
("side_effect", "error"),
("is_on", "exception", "error_message"),
[
(WebOsTvPairError, "Pairing with TV failed"),
(ConnectionResetError, "TV unreachable"),
(
True,
WebOsTvCommandError("Some error"),
f"Communication error while sending notification to device {TV_NAME}: Some error",
),
(
True,
FileNotFoundError("Some other error"),
f"Icon {ICON_PATH} not found when sending notification for device {TV_NAME}",
),
(
False,
None,
f"Error sending notification to device {TV_NAME}: Device is off and cannot be controlled",
),
],
)
async def test_connection_errors(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, client, side_effect, error
async def test_errors(
hass: HomeAssistant,
client,
is_on: bool,
exception: Exception,
error_message: str,
) -> None:
"""Test connection errors scenarios."""
"""Test error scenarios."""
await setup_webostv(hass)
client.is_on = is_on
assert hass.services.has_service("notify", SERVICE_NAME)
client.is_connected.return_value = False
client.connect.side_effect = side_effect
client.send_message.side_effect = exception
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_NAME,
@ -148,10 +121,8 @@ async def test_connection_errors(
},
blocking=True,
)
assert client.mock_calls[0] == call.connect()
assert client.connect.call_count == 2
client.send_message.assert_not_called()
assert error in caplog.text
assert client.send_message.call_count == int(is_on)
async def test_no_discovery_info(