Add remote entity to AndroidTV (#103496)

* Add remote entity to AndroidTV

* Add tests for remote entity

* Requested changes on tests
This commit is contained in:
ollo69 2024-06-04 17:02:30 +02:00 committed by GitHub
parent b09f3eb313
commit 8610436948
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 249 additions and 1 deletions

View File

@ -61,7 +61,7 @@ ADB_PYTHON_EXCEPTIONS: tuple = (
)
ADB_TCP_EXCEPTIONS: tuple = (ConnectionResetError, RuntimeError)
PLATFORMS = [Platform.MEDIA_PLAYER]
PLATFORMS = [Platform.MEDIA_PLAYER, Platform.REMOTE]
RELOAD_OPTIONS = [CONF_STATE_DETECTION_RULES]
_INVALID_MACS = {"ff:ff:ff:ff:ff:ff"}

View File

@ -18,6 +18,7 @@ from homeassistant.const import (
CONF_HOST,
CONF_NAME,
)
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.entity import Entity
@ -87,6 +88,9 @@ def adb_decorator[_ADBDeviceT: AndroidTVEntity, **_P, _R](
await self.aftv.adb_close()
self._attr_available = False
return None
except ServiceValidationError:
# Service validation error is thrown because raised by remote services
raise
except Exception as err: # noqa: BLE001
# An unforeseen exception occurred. Close the ADB connection so that
# it doesn't happen over and over again.

View File

@ -0,0 +1,75 @@
"""Support for the AndroidTV remote."""
from __future__ import annotations
from collections.abc import Iterable
import logging
from typing import Any
from androidtv.constants import KEYS
from homeassistant.components.remote import ATTR_NUM_REPEATS, RemoteEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import CONF_TURN_OFF_COMMAND, CONF_TURN_ON_COMMAND, DOMAIN
from .entity import AndroidTVEntity, adb_decorator
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the AndroidTV remote from a config entry."""
async_add_entities([AndroidTVRemote(entry)])
class AndroidTVRemote(AndroidTVEntity, RemoteEntity):
"""Device that sends commands to a AndroidTV."""
_attr_name = None
_attr_should_poll = False
@adb_decorator()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the device."""
options = self._entry_runtime_data.dev_opt
if turn_on_cmd := options.get(CONF_TURN_ON_COMMAND):
await self.aftv.adb_shell(turn_on_cmd)
else:
await self.aftv.turn_on()
@adb_decorator()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the device."""
options = self._entry_runtime_data.dev_opt
if turn_off_cmd := options.get(CONF_TURN_OFF_COMMAND):
await self.aftv.adb_shell(turn_off_cmd)
else:
await self.aftv.turn_off()
@adb_decorator()
async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
"""Send a command to a device."""
num_repeats = kwargs[ATTR_NUM_REPEATS]
command_list = []
for cmd in command:
if key := KEYS.get(cmd):
command_list.append(f"input keyevent {key}")
else:
command_list.append(cmd)
for _ in range(num_repeats):
for cmd in command_list:
try:
await self.aftv.adb_shell(cmd)
except UnicodeDecodeError as ex:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="failed_send",
translation_placeholders={"cmd": cmd},
) from ex

View File

@ -103,5 +103,10 @@
"name": "Learn sendevent",
"description": "Translates a key press on a remote into ADB 'sendevent' commands. You must press one button on the remote within 8 seconds of calling this service."
}
},
"exceptions": {
"failed_send": {
"message": "Failed to send command {cmd}"
}
}
}

View File

@ -0,0 +1,164 @@
"""The tests for the androidtv remote platform."""
from typing import Any
from unittest.mock import call, patch
from androidtv.constants import KEYS
import pytest
from homeassistant.components.androidtv.const import (
CONF_TURN_OFF_COMMAND,
CONF_TURN_ON_COMMAND,
)
from homeassistant.components.remote import (
ATTR_NUM_REPEATS,
DOMAIN as REMOTE_DOMAIN,
SERVICE_SEND_COMMAND,
)
from homeassistant.const import (
ATTR_COMMAND,
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from . import patchers
from .common import (
CONFIG_ANDROID_DEFAULT,
CONFIG_FIRETV_DEFAULT,
SHELL_RESPONSE_OFF,
SHELL_RESPONSE_STANDBY,
setup_mock_entry,
)
from tests.common import MockConfigEntry
def _setup(config: dict[str, Any]) -> tuple[str, str, MockConfigEntry]:
"""Prepare mock entry for the media player tests."""
return setup_mock_entry(config, REMOTE_DOMAIN)
async def _test_service(
hass: HomeAssistant,
entity_id,
ha_service_name,
androidtv_method,
additional_service_data=None,
expected_call_args=None,
) -> None:
"""Test generic Android media player entity service."""
if expected_call_args is None:
expected_call_args = [None]
service_data = {ATTR_ENTITY_ID: entity_id}
if additional_service_data:
service_data.update(additional_service_data)
androidtv_patch = (
"androidtv.androidtv_async.AndroidTVAsync"
if "android" in entity_id
else "firetv.firetv_async.FireTVAsync"
)
with patch(f"androidtv.{androidtv_patch}.{androidtv_method}") as api_call:
await hass.services.async_call(
REMOTE_DOMAIN,
ha_service_name,
service_data=service_data,
blocking=True,
)
assert api_call.called
assert api_call.call_count == len(expected_call_args)
expected_calls = [call(s) if s else call() for s in expected_call_args]
assert api_call.call_args_list == expected_calls
@pytest.mark.parametrize("config", [CONFIG_ANDROID_DEFAULT, CONFIG_FIRETV_DEFAULT])
async def test_services_remote(hass: HomeAssistant, config) -> None:
"""Test services for remote entity."""
patch_key, entity_id, config_entry = _setup(config)
config_entry.add_to_hass(hass)
with patchers.patch_connect(True)[patch_key]:
with patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with (
patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key],
patchers.PATCH_SCREENCAP,
):
await _test_service(hass, entity_id, SERVICE_TURN_OFF, "turn_off")
await _test_service(hass, entity_id, SERVICE_TURN_ON, "turn_on")
await _test_service(
hass,
entity_id,
SERVICE_SEND_COMMAND,
"adb_shell",
{ATTR_COMMAND: ["BACK", "test"], ATTR_NUM_REPEATS: 2},
[
f"input keyevent {KEYS["BACK"]}",
"test",
f"input keyevent {KEYS["BACK"]}",
"test",
],
)
@pytest.mark.parametrize("config", [CONFIG_ANDROID_DEFAULT, CONFIG_FIRETV_DEFAULT])
async def test_services_remote_custom(hass: HomeAssistant, config) -> None:
"""Test services with custom options for remote entity."""
patch_key, entity_id, config_entry = _setup(config)
config_entry.add_to_hass(hass)
hass.config_entries.async_update_entry(
config_entry,
options={
CONF_TURN_OFF_COMMAND: "test off",
CONF_TURN_ON_COMMAND: "test on",
},
)
with patchers.patch_connect(True)[patch_key]:
with patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with (
patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key],
patchers.PATCH_SCREENCAP,
):
await _test_service(
hass, entity_id, SERVICE_TURN_OFF, "adb_shell", None, ["test off"]
)
await _test_service(
hass, entity_id, SERVICE_TURN_ON, "adb_shell", None, ["test on"]
)
async def test_remote_unicode_decode_error(hass: HomeAssistant) -> None:
"""Test sending a command via the send_command remote service that raises a UnicodeDecodeError exception."""
patch_key, entity_id, config_entry = _setup(CONFIG_ANDROID_DEFAULT)
config_entry.add_to_hass(hass)
response = b"test response"
with patchers.patch_connect(True)[patch_key]:
with patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with patch(
"androidtv.basetv.basetv_async.BaseTVAsync.adb_shell",
side_effect=UnicodeDecodeError("utf-8", response, 0, len(response), "TEST"),
) as api_call:
try:
await hass.services.async_call(
REMOTE_DOMAIN,
SERVICE_SEND_COMMAND,
service_data={ATTR_ENTITY_ID: entity_id, ATTR_COMMAND: "BACK"},
blocking=True,
)
pytest.fail("Exception not raised")
except ServiceValidationError:
assert api_call.call_count == 1