Fix LG webOS TV actions not returning responses (#136743)

This commit is contained in:
Shay Levy 2025-01-28 20:55:06 +02:00 committed by GitHub
parent bae9516fc2
commit 55fc01be8e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 87 additions and 22 deletions

View File

@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Awaitable, Callable, Coroutine from collections.abc import Callable, Coroutine
from contextlib import suppress from contextlib import suppress
from datetime import timedelta from datetime import timedelta
from functools import wraps from functools import wraps
@ -23,7 +23,7 @@ from homeassistant.components.media_player import (
MediaType, MediaType,
) )
from homeassistant.const import ATTR_COMMAND, ATTR_SUPPORTED_FEATURES from homeassistant.const import ATTR_COMMAND, ATTR_SUPPORTED_FEATURES
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant, ServiceResponse, SupportsResponse
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, entity_platform from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -78,9 +78,24 @@ COMMAND_SCHEMA: VolDictType = {
SOUND_OUTPUT_SCHEMA: VolDictType = {vol.Required(ATTR_SOUND_OUTPUT): cv.string} SOUND_OUTPUT_SCHEMA: VolDictType = {vol.Required(ATTR_SOUND_OUTPUT): cv.string}
SERVICES = ( SERVICES = (
(SERVICE_BUTTON, BUTTON_SCHEMA, "async_button"), (
(SERVICE_COMMAND, COMMAND_SCHEMA, "async_command"), SERVICE_BUTTON,
(SERVICE_SELECT_SOUND_OUTPUT, SOUND_OUTPUT_SCHEMA, "async_select_sound_output"), BUTTON_SCHEMA,
"async_button",
SupportsResponse.NONE,
),
(
SERVICE_COMMAND,
COMMAND_SCHEMA,
"async_command",
SupportsResponse.OPTIONAL,
),
(
SERVICE_SELECT_SOUND_OUTPUT,
SOUND_OUTPUT_SCHEMA,
"async_select_sound_output",
SupportsResponse.OPTIONAL,
),
) )
@ -92,19 +107,23 @@ async def async_setup_entry(
"""Set up the LG webOS TV platform.""" """Set up the LG webOS TV platform."""
platform = entity_platform.async_get_current_platform() platform = entity_platform.async_get_current_platform()
for service_name, schema, method in SERVICES: for service_name, schema, method, supports_response in SERVICES:
platform.async_register_entity_service(service_name, schema, method) platform.async_register_entity_service(
service_name, schema, method, supports_response=supports_response
)
async_add_entities([LgWebOSMediaPlayerEntity(entry)]) async_add_entities([LgWebOSMediaPlayerEntity(entry)])
def cmd[_T: LgWebOSMediaPlayerEntity, **_P]( def cmd[_R, **_P](
func: Callable[Concatenate[_T, _P], Awaitable[None]], func: Callable[Concatenate[LgWebOSMediaPlayerEntity, _P], Coroutine[Any, Any, _R]],
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: ) -> Callable[Concatenate[LgWebOSMediaPlayerEntity, _P], Coroutine[Any, Any, _R]]:
"""Catch command exceptions.""" """Catch command exceptions."""
@wraps(func) @wraps(func)
async def cmd_wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> None: async def cmd_wrapper(
self: LgWebOSMediaPlayerEntity, *args: _P.args, **kwargs: _P.kwargs
) -> _R:
"""Wrap all command methods.""" """Wrap all command methods."""
if self.state is MediaPlayerState.OFF: if self.state is MediaPlayerState.OFF:
raise HomeAssistantError( raise HomeAssistantError(
@ -116,7 +135,7 @@ def cmd[_T: LgWebOSMediaPlayerEntity, **_P](
}, },
) )
try: try:
await func(self, *args, **kwargs) return await func(self, *args, **kwargs)
except WEBOSTV_EXCEPTIONS as error: except WEBOSTV_EXCEPTIONS as error:
raise HomeAssistantError( raise HomeAssistantError(
translation_domain=DOMAIN, translation_domain=DOMAIN,
@ -376,9 +395,9 @@ class LgWebOSMediaPlayerEntity(RestoreEntity, MediaPlayerEntity):
await self._client.set_mute(mute) await self._client.set_mute(mute)
@cmd @cmd
async def async_select_sound_output(self, sound_output: str) -> None: async def async_select_sound_output(self, sound_output: str) -> ServiceResponse:
"""Select the sound output.""" """Select the sound output."""
await self._client.change_sound_output(sound_output) return await self._client.change_sound_output(sound_output)
@cmd @cmd
async def async_media_play_pause(self) -> None: async def async_media_play_pause(self) -> None:
@ -481,9 +500,9 @@ class LgWebOSMediaPlayerEntity(RestoreEntity, MediaPlayerEntity):
await self._client.button(button) await self._client.button(button)
@cmd @cmd
async def async_command(self, command: str, **kwargs: Any) -> None: async def async_command(self, command: str, **kwargs: Any) -> ServiceResponse:
"""Send a command.""" """Send a command."""
await self._client.request(command, payload=kwargs.get(ATTR_PAYLOAD)) return await self._client.request(command, payload=kwargs.get(ATTR_PAYLOAD))
async def _async_fetch_image(self, url: str) -> tuple[bytes | None, str | None]: async def _async_fetch_image(self, url: str) -> tuple[bytes | None, str | None]:
"""Retrieve an image. """Retrieve an image.

View File

@ -1,4 +1,14 @@
# serializer version: 1 # serializer version: 1
# name: test_command
dict({
'media_player.lg_webos_tv_model': dict({
'muted': False,
'returnValue': True,
'scenario': 'mastervolume_tv_speaker_ext',
'volume': 1,
}),
})
# ---
# name: test_entity_attributes # name: test_entity_attributes
StateSnapshot({ StateSnapshot({
'attributes': ReadOnlyDict({ 'attributes': ReadOnlyDict({
@ -57,3 +67,11 @@
'via_device_id': None, 'via_device_id': None,
}) })
# --- # ---
# name: test_select_sound_output
dict({
'media_player.lg_webos_tv_model': dict({
'method': 'setSystemSettings',
'returnValue': True,
}),
})
# ---

View File

@ -229,17 +229,30 @@ async def test_button(hass: HomeAssistant, client) -> None:
client.button.assert_called_with("test") client.button.assert_called_with("test")
async def test_command(hass: HomeAssistant, client) -> None: async def test_command(
hass: HomeAssistant,
client,
snapshot: SnapshotAssertion,
) -> None:
"""Test generic command functionality.""" """Test generic command functionality."""
await setup_webostv(hass) await setup_webostv(hass)
client.request.return_value = {
"returnValue": True,
"scenario": "mastervolume_tv_speaker_ext",
"volume": 1,
"muted": False,
}
data = { data = {
ATTR_ENTITY_ID: ENTITY_ID, ATTR_ENTITY_ID: ENTITY_ID,
ATTR_COMMAND: "test", ATTR_COMMAND: "audio/getVolume",
} }
await hass.services.async_call(DOMAIN, SERVICE_COMMAND, data, True) response = await hass.services.async_call(
DOMAIN, SERVICE_COMMAND, data, True, return_response=True
)
await hass.async_block_till_done() await hass.async_block_till_done()
client.request.assert_called_with("test", payload=None) client.request.assert_called_with("audio/getVolume", payload=None)
assert response == snapshot
async def test_command_with_optional_arg(hass: HomeAssistant, client) -> None: async def test_command_with_optional_arg(hass: HomeAssistant, client) -> None:
@ -258,17 +271,32 @@ async def test_command_with_optional_arg(hass: HomeAssistant, client) -> None:
) )
async def test_select_sound_output(hass: HomeAssistant, client) -> None: async def test_select_sound_output(
hass: HomeAssistant,
client,
snapshot: SnapshotAssertion,
) -> None:
"""Test select sound output service.""" """Test select sound output service."""
await setup_webostv(hass) await setup_webostv(hass)
client.change_sound_output.return_value = {
"returnValue": True,
"method": "setSystemSettings",
}
data = { data = {
ATTR_ENTITY_ID: ENTITY_ID, ATTR_ENTITY_ID: ENTITY_ID,
ATTR_SOUND_OUTPUT: "external_speaker", ATTR_SOUND_OUTPUT: "external_speaker",
} }
await hass.services.async_call(DOMAIN, SERVICE_SELECT_SOUND_OUTPUT, data, True) response = await hass.services.async_call(
DOMAIN,
SERVICE_SELECT_SOUND_OUTPUT,
data,
True,
return_response=True,
)
await hass.async_block_till_done() await hass.async_block_till_done()
client.change_sound_output.assert_called_once_with("external_speaker") client.change_sound_output.assert_called_once_with("external_speaker")
assert response == snapshot
async def test_device_info_startup_off( async def test_device_info_startup_off(