jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add media_player platform to Android TV Remote (#91677)

Browse Source
This commit is contained in:
Artem Draft 2023-05-06 17:18:34 +03:00 committed by GitHub
parent 053eaad2bd
commit d816da9355
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 702 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CODEOWNERS
View File

@ -80,8 +80,8 @@ build.json @home-assistant/supervisor
/tests/components/android_ip_webcam/ @engrbm87
/homeassistant/components/androidtv/ @JeffLIrion @ollo69
/tests/components/androidtv/ @JeffLIrion @ollo69
/homeassistant/components/androidtv_remote/ @tronikos
/tests/components/androidtv_remote/ @tronikos
/homeassistant/components/androidtv_remote/ @tronikos @Drafteed
/tests/components/androidtv_remote/ @tronikos @Drafteed
/homeassistant/components/anova/ @Lash-L
/tests/components/anova/ @Lash-L
/homeassistant/components/anthemav/ @hyralex

25
homeassistant/components/androidtv_remote/__init__.py
View File

@ -1,6 +1,8 @@
"""The Android TV Remote integration."""
from __future__ import annotations
import logging
from androidtvremote2 import (
AndroidTVRemote,
CannotConnect,
@ -9,20 +11,37 @@ from androidtvremote2 import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from .const import DOMAIN
from .helpers import create_api
PLATFORMS: list[Platform] = [Platform.REMOTE]
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[Platform] = [Platform.MEDIA_PLAYER, Platform.REMOTE]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Android TV Remote from a config entry."""
api = create_api(hass, entry.data[CONF_HOST])
@callback
def is_available_updated(is_available: bool) -> None:
if is_available:
_LOGGER.info(
"Reconnected to %s at %s", entry.data[CONF_NAME], entry.data[CONF_HOST]
)
else:
_LOGGER.warning(
"Disconnected from %s at %s",
entry.data[CONF_NAME],
entry.data[CONF_HOST],
)
api.add_is_available_updated_callback(is_available_updated)
try:
await api.async_connect()
except InvalidAuth as exc:

84
homeassistant/components/androidtv_remote/entity.py Normal file
View File

@ -0,0 +1,84 @@
"""Base entity for Android TV Remote."""
from __future__ import annotations
from androidtvremote2 import AndroidTVRemote, ConnectionClosed
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import DOMAIN
class AndroidTVRemoteBaseEntity(Entity):
"""Android TV Remote Base Entity."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(self, api: AndroidTVRemote, config_entry: ConfigEntry) -> None:
"""Initialize the entity."""
self._api = api
self._host = config_entry.data[CONF_HOST]
self._name = config_entry.data[CONF_NAME]
self._attr_unique_id = config_entry.unique_id
self._attr_is_on = api.is_on
device_info = api.device_info
assert config_entry.unique_id
assert device_info
self._attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, config_entry.data[CONF_MAC])},
identifiers={(DOMAIN, config_entry.unique_id)},
name=self._name,
manufacturer=device_info["manufacturer"],
model=device_info["model"],
)
@callback
def _is_available_updated(self, is_available: bool) -> None:
"""Update the state when the device is ready to receive commands or is unavailable."""
self._attr_available = is_available
self.async_write_ha_state()
@callback
def _is_on_updated(self, is_on: bool) -> None:
"""Update the state when device turns on or off."""
self._attr_is_on = is_on
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
self._api.add_is_available_updated_callback(self._is_available_updated)
self._api.add_is_on_updated_callback(self._is_on_updated)
async def async_will_remove_from_hass(self) -> None:
"""Remove callbacks."""
self._api.remove_is_available_updated_callback(self._is_available_updated)
self._api.remove_is_on_updated_callback(self._is_on_updated)
def _send_key_command(self, key_code: str, direction: str = "SHORT") -> None:
"""Send a key press to Android TV.
This does not block; it buffers the data and arranges for it to be sent out asynchronously.
"""
try:
self._api.send_key_command(key_code, direction)
except ConnectionClosed as exc:
raise HomeAssistantError(
"Connection to Android TV device is closed"
) from exc
def _send_launch_app_command(self, app_link: str) -> None:
"""Launch an app on Android TV.
This does not block; it buffers the data and arranges for it to be sent out asynchronously.
"""
try:
self._api.send_launch_app_command(app_link)
except ConnectionClosed as exc:
raise HomeAssistantError(
"Connection to Android TV device is closed"
) from exc

4
homeassistant/components/androidtv_remote/manifest.json
View File

@ -1,13 +1,13 @@
{
"domain": "androidtv_remote",
"name": "Android TV Remote",
"codeowners": ["@tronikos"],
"codeowners": ["@tronikos", "@Drafteed"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/androidtv_remote",
"integration_type": "device",
"iot_class": "local_push",
"loggers": ["androidtvremote2"],
"quality_scale": "platinum",
"requirements": ["androidtvremote2==0.0.7"],
"requirements": ["androidtvremote2==0.0.8"],
"zeroconf": ["_androidtvremote2._tcp.local."]
}

198
homeassistant/components/androidtv_remote/media_player.py Normal file
View File

@ -0,0 +1,198 @@
"""Media player support for Android TV Remote."""
from __future__ import annotations
import asyncio
from typing import Any
from androidtvremote2 import AndroidTVRemote, ConnectionClosed
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .entity import AndroidTVRemoteBaseEntity
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Android TV media player entity based on a config entry."""
api: AndroidTVRemote = hass.data[DOMAIN][config_entry.entry_id]
async_add_entities([AndroidTVRemoteMediaPlayerEntity(api, config_entry)])
class AndroidTVRemoteMediaPlayerEntity(AndroidTVRemoteBaseEntity, MediaPlayerEntity):
"""Android TV Remote Media Player Entity."""
_attr_assumed_state = True
_attr_device_class = MediaPlayerDeviceClass.TV
_attr_supported_features = (
MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.PREVIOUS_TRACK
| MediaPlayerEntityFeature.NEXT_TRACK
| MediaPlayerEntityFeature.TURN_ON
| MediaPlayerEntityFeature.TURN_OFF
| MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.PLAY_MEDIA
)
def __init__(self, api: AndroidTVRemote, config_entry: ConfigEntry) -> None:
"""Initialize the entity."""
super().__init__(api, config_entry)
# This task is needed to create a job that sends a key press
# sequence that can be canceled if concurrency occurs
self._channel_set_task: asyncio.Task | None = None
def _update_current_app(self, current_app: str) -> None:
"""Update current app info."""
self._attr_app_id = current_app
self._attr_app_name = current_app
def _update_volume_info(self, volume_info: dict[str, str | bool]) -> None:
"""Update volume info."""
if volume_info.get("max"):
self._attr_volume_level = int(volume_info["level"]) / int(
volume_info["max"]
)
self._attr_is_volume_muted = bool(volume_info["muted"])
else:
self._attr_volume_level = None
self._attr_is_volume_muted = None
@callback
def _current_app_updated(self, current_app: str) -> None:
"""Update the state when the current app changes."""
self._update_current_app(current_app)
self.async_write_ha_state()
@callback
def _volume_info_updated(self, volume_info: dict[str, str | bool]) -> None:
"""Update the state when the volume info changes."""
self._update_volume_info(volume_info)
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
await super().async_added_to_hass()
self._update_current_app(self._api.current_app)
self._update_volume_info(self._api.volume_info)
self._api.add_current_app_updated_callback(self._current_app_updated)
self._api.add_volume_info_updated_callback(self._volume_info_updated)
async def async_will_remove_from_hass(self) -> None:
"""Remove callbacks."""
await super().async_will_remove_from_hass()
self._api.remove_current_app_updated_callback(self._current_app_updated)
self._api.remove_volume_info_updated_callback(self._volume_info_updated)
@property
def state(self) -> MediaPlayerState:
"""Return the state of the device."""
if self._attr_is_on:
return MediaPlayerState.ON
return MediaPlayerState.OFF
async def async_turn_on(self) -> None:
"""Turn the Android TV on."""
if not self._attr_is_on:
self._send_key_command("POWER")
async def async_turn_off(self) -> None:
"""Turn the Android TV off."""
if self._attr_is_on:
self._send_key_command("POWER")
async def async_volume_up(self) -> None:
"""Turn volume up for media player."""
self._send_key_command("VOLUME_UP")
async def async_volume_down(self) -> None:
"""Turn volume down for media player."""
self._send_key_command("VOLUME_DOWN")
async def async_mute_volume(self, mute: bool) -> None:
"""Mute the volume."""
if mute != self.is_volume_muted:
self._send_key_command("VOLUME_MUTE")
async def async_media_play(self) -> None:
"""Send play command."""
self._send_key_command("MEDIA_PLAY")
async def async_media_pause(self) -> None:
"""Send pause command."""
self._send_key_command("MEDIA_PAUSE")
async def async_media_play_pause(self) -> None:
"""Send play/pause command."""
self._send_key_command("MEDIA_PLAY_PAUSE")
async def async_media_stop(self) -> None:
"""Send stop command."""
self._send_key_command("MEDIA_STOP")
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
self._send_key_command("MEDIA_PREVIOUS")
async def async_media_next_track(self) -> None:
"""Send next track command."""
self._send_key_command("MEDIA_NEXT")
async def async_play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
"""Play a piece of media."""
if media_type == MediaType.CHANNEL:
if not media_id.isnumeric():
raise ValueError(f"Channel must be numeric: {media_id}")
if self._channel_set_task:
self._channel_set_task.cancel()
self._channel_set_task = asyncio.create_task(
self._send_key_commands(list(media_id))
)
await self._channel_set_task
return
if media_type == MediaType.URL:
self._send_launch_app_command(media_id)
return
raise ValueError(f"Invalid media type: {media_type}")
async def _send_key_commands(
self, key_codes: list[str], delay_secs: float = 0.1
) -> None:
"""Send a key press sequence to Android TV.
The delay is necessary because device may ignore
some commands if we send the sequence without delay.
"""
try:
for key_code in key_codes:
self._api.send_key_command(key_code)
await asyncio.sleep(delay_secs)
except ConnectionClosed as exc:
raise HomeAssistantError(
"Connection to Android TV device is closed"
) from exc

100
homeassistant/components/androidtv_remote/remote.py
View File

@ -3,10 +3,9 @@ from __future__ import annotations
import asyncio
from collections.abc import Iterable
import logging
from typing import Any
from androidtvremote2 import AndroidTVRemote, ConnectionClosed
from androidtvremote2 import AndroidTVRemote
from homeassistant.components.remote import (
ATTR_ACTIVITY,
@ -20,17 +19,13 @@ from homeassistant.components.remote import (
RemoteEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .entity import AndroidTVRemoteBaseEntity
PARALLEL_UPDATES = 0
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
@ -43,62 +38,29 @@ async def async_setup_entry(
async_add_entities([AndroidTVRemoteEntity(api, config_entry)])
class AndroidTVRemoteEntity(RemoteEntity):
"""Representation of an Android TV Remote."""
class AndroidTVRemoteEntity(AndroidTVRemoteBaseEntity, RemoteEntity):
"""Android TV Remote Entity."""
_attr_has_entity_name = True
_attr_should_poll = False
_attr_supported_features = RemoteEntityFeature.ACTIVITY
def __init__(self, api: AndroidTVRemote, config_entry: ConfigEntry) -> None:
"""Initialize device."""
self._api = api
self._host = config_entry.data[CONF_HOST]
self._name = config_entry.data[CONF_NAME]
self._attr_unique_id = config_entry.unique_id
self._attr_supported_features = RemoteEntityFeature.ACTIVITY
self._attr_is_on = api.is_on
self._attr_current_activity = api.current_app
device_info = api.device_info
assert config_entry.unique_id
assert device_info
self._attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, config_entry.data[CONF_MAC])},
identifiers={(DOMAIN, config_entry.unique_id)},
name=self._name,
manufacturer=device_info["manufacturer"],
model=device_info["model"],
)
@callback
def _current_app_updated(self, current_app: str) -> None:
"""Update the state when the current app changes."""
self._attr_current_activity = current_app
self.async_write_ha_state()
@callback
def is_on_updated(is_on: bool) -> None:
self._attr_is_on = is_on
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
await super().async_added_to_hass()
@callback
def current_app_updated(current_app: str) -> None:
self._attr_current_activity = current_app
self.async_write_ha_state()
self._attr_current_activity = self._api.current_app
self._api.add_current_app_updated_callback(self._current_app_updated)
@callback
def is_available_updated(is_available: bool) -> None:
if is_available:
_LOGGER.info(
"Reconnected to %s at %s",
self._name,
self._host,
)
else:
_LOGGER.warning(
"Disconnected from %s at %s",
self._name,
self._host,
)
self._attr_available = is_available
self.async_write_ha_state()
async def async_will_remove_from_hass(self) -> None:
"""Remove callbacks."""
await super().async_will_remove_from_hass()
api.add_is_on_updated_callback(is_on_updated)
api.add_current_app_updated_callback(current_app_updated)
api.add_is_available_updated_callback(is_available_updated)
self._api.remove_current_app_updated_callback(self._current_app_updated)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the Android TV on."""
@ -128,27 +90,3 @@ class AndroidTVRemoteEntity(RemoteEntity):
else:
self._send_key_command(single_command, "SHORT")
await asyncio.sleep(delay_secs)
def _send_key_command(self, key_code: str, direction: str = "SHORT") -> None:
"""Send a key press to Android TV.
This does not block; it buffers the data and arranges for it to be sent out asynchronously.
"""
try:
self._api.send_key_command(key_code, direction)
except ConnectionClosed as exc:
raise HomeAssistantError(
"Connection to Android TV device is closed"
) from exc
def _send_launch_app_command(self, app_link: str) -> None:
"""Launch an app on Android TV.
This does not block; it buffers the data and arranges for it to be sent out asynchronously.
"""
try:
self._api.send_launch_app_command(app_link)
except ConnectionClosed as exc:
raise HomeAssistantError(
"Connection to Android TV device is closed"
) from exc

2
requirements_all.txt
View File

@ -333,7 +333,7 @@ amcrest==1.9.7
androidtv[async]==0.0.70
# homeassistant.components.androidtv_remote
androidtvremote2==0.0.7
androidtvremote2==0.0.8
# homeassistant.components.anel_pwrctrl
anel_pwrctrl-homeassistant==0.0.1.dev2

2
requirements_test_all.txt
View File

@ -308,7 +308,7 @@ ambiclimate==0.2.1
androidtv[async]==0.0.70
# homeassistant.components.androidtv_remote
androidtvremote2==0.0.7
androidtvremote2==0.0.8
# homeassistant.components.anova
anova-wifi==0.9.0

53
tests/components/androidtv_remote/conftest.py
View File

@ -1,5 +1,5 @@
"""Fixtures for the Android TV Remote integration tests."""
from collections.abc import Generator
from collections.abc import Callable, Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -42,6 +42,57 @@ def mock_api() -> Generator[None, MagicMock, None]:
"manufacturer": "My Android TV manufacturer",
"model": "My Android TV model",
}
is_on_updated_callbacks: list[Callable] = []
current_app_updated_callbacks: list[Callable] = []
volume_info_updated_callbacks: list[Callable] = []
is_available_updated_callbacks: list[Callable] = []
def mocked_add_is_on_updated_callback(callback: Callable):
is_on_updated_callbacks.append(callback)
def mocked_add_current_app_updated_callback(callback: Callable):
current_app_updated_callbacks.append(callback)
def mocked_add_volume_info_updated_callback(callback: Callable):
volume_info_updated_callbacks.append(callback)
def mocked_add_is_available_updated_callbacks(callback: Callable):
is_available_updated_callbacks.append(callback)
def mocked_is_on_updated(is_on: bool):
for callback in is_on_updated_callbacks:
callback(is_on)
def mocked_current_app_updated(current_app: str):
for callback in current_app_updated_callbacks:
callback(current_app)
def mocked_volume_info_updated(volume_info: dict[str, str | bool]):
for callback in volume_info_updated_callbacks:
callback(volume_info)
def mocked_is_available_updated(is_available: bool):
for callback in is_available_updated_callbacks:
callback(is_available)
mock_api.add_is_on_updated_callback.side_effect = (
mocked_add_is_on_updated_callback
)
mock_api.add_current_app_updated_callback.side_effect = (
mocked_add_current_app_updated_callback
)
mock_api.add_volume_info_updated_callback.side_effect = (
mocked_add_volume_info_updated_callback
)
mock_api.add_is_available_updated_callback.side_effect = (
mocked_add_is_available_updated_callbacks
)
mock_api._on_is_on_updated.side_effect = mocked_is_on_updated
mock_api._on_current_app_updated.side_effect = mocked_current_app_updated
mock_api._on_volume_info_updated.side_effect = mocked_volume_info_updated
mock_api._on_is_available_updated.side_effect = mocked_is_available_updated
yield mock_api

314
tests/components/androidtv_remote/test_media_player.py Normal file
View File

@ -0,0 +1,314 @@
"""Tests for the Android TV Remote remote platform."""
from unittest.mock import MagicMock, call
from androidtvremote2 import ConnectionClosed
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from tests.common import MockConfigEntry
MEDIA_PLAYER_ENTITY = "media_player.my_android_tv"
async def test_media_player_receives_push_updates(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test the Android TV Remote media player receives push updates and state is updated."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_api._on_is_on_updated(False)
assert hass.states.is_state(MEDIA_PLAYER_ENTITY, STATE_OFF)
mock_api._on_is_on_updated(True)
assert hass.states.is_state(MEDIA_PLAYER_ENTITY, STATE_ON)
mock_api._on_current_app_updated("com.google.android.tvlauncher")
assert (
hass.states.get(MEDIA_PLAYER_ENTITY).attributes.get("app_id")
== "com.google.android.tvlauncher"
)
assert (
hass.states.get(MEDIA_PLAYER_ENTITY).attributes.get("app_name")
== "com.google.android.tvlauncher"
)
mock_api._on_volume_info_updated({"level": 35, "muted": False, "max": 100})
assert hass.states.get(MEDIA_PLAYER_ENTITY).attributes.get("volume_level") == 0.35
mock_api._on_volume_info_updated({"level": 50, "muted": True, "max": 100})
assert hass.states.get(MEDIA_PLAYER_ENTITY).attributes.get("volume_level") == 0.50
assert hass.states.get(MEDIA_PLAYER_ENTITY).attributes.get("is_volume_muted")
mock_api._on_volume_info_updated({"level": 0, "muted": False, "max": 0})
assert hass.states.get(MEDIA_PLAYER_ENTITY).attributes.get("volume_level") is None
assert (
hass.states.get(MEDIA_PLAYER_ENTITY).attributes.get("is_volume_muted") is None
)
mock_api._on_is_available_updated(False)
assert hass.states.is_state(MEDIA_PLAYER_ENTITY, STATE_UNAVAILABLE)
mock_api._on_is_available_updated(True)
assert hass.states.is_state(MEDIA_PLAYER_ENTITY, STATE_ON)
async def test_media_player_toggles(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test the Android TV Remote media player toggles."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.services.async_call(
"media_player",
"turn_off",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api._on_is_on_updated(False)
mock_api.send_key_command.assert_called_with("POWER", "SHORT")
assert await hass.services.async_call(
"media_player",
"turn_on",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api._on_is_on_updated(True)
mock_api.send_key_command.assert_called_with("POWER", "SHORT")
async def test_media_player_volume(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test the Android TV Remote media player up/down/mute volume."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.services.async_call(
"media_player",
"volume_up",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api._on_volume_info_updated({"level": 10, "muted": False, "max": 100})
mock_api.send_key_command.assert_called_with("VOLUME_UP", "SHORT")
assert await hass.services.async_call(
"media_player",
"volume_down",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api._on_volume_info_updated({"level": 9, "muted": False, "max": 100})
mock_api.send_key_command.assert_called_with("VOLUME_DOWN", "SHORT")
assert await hass.services.async_call(
"media_player",
"volume_mute",
{"entity_id": MEDIA_PLAYER_ENTITY, "is_volume_muted": True},
blocking=True,
)
mock_api._on_volume_info_updated({"level": 9, "muted": True, "max": 100})
mock_api.send_key_command.assert_called_with("VOLUME_MUTE", "SHORT")
assert await hass.services.async_call(
"media_player",
"volume_mute",
{"entity_id": MEDIA_PLAYER_ENTITY, "is_volume_muted": False},
blocking=True,
)
mock_api._on_volume_info_updated({"level": 9, "muted": False, "max": 100})
mock_api.send_key_command.assert_called_with("VOLUME_MUTE", "SHORT")
async def test_media_player_controls(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test the Android TV Remote media player play/pause/stop/next/prev."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.services.async_call(
"media_player",
"media_play",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api.send_key_command.assert_called_with("MEDIA_PLAY", "SHORT")
assert await hass.services.async_call(
"media_player",
"media_pause",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api.send_key_command.assert_called_with("MEDIA_PAUSE", "SHORT")
assert await hass.services.async_call(
"media_player",
"media_play_pause",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api.send_key_command.assert_called_with("MEDIA_PLAY_PAUSE", "SHORT")
assert await hass.services.async_call(
"media_player",
"media_stop",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api.send_key_command.assert_called_with("MEDIA_STOP", "SHORT")
assert await hass.services.async_call(
"media_player",
"media_previous_track",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api.send_key_command.assert_called_with("MEDIA_PREVIOUS", "SHORT")
assert await hass.services.async_call(
"media_player",
"media_next_track",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api.send_key_command.assert_called_with("MEDIA_NEXT", "SHORT")
async def test_media_player_play_media(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test the Android TV Remote media player play_media."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.services.async_call(
"media_player",
"play_media",
{
"entity_id": MEDIA_PLAYER_ENTITY,
"media_content_type": "channel",
"media_content_id": "45",
},
blocking=True,
)
assert mock_api.send_key_command.mock_calls == [
call("4"),
call("5"),
]
# Test that set channel task has been canceled
mock_api.send_key_command.reset_mock()
await hass.services.async_call(
"media_player",
"play_media",
{
"entity_id": MEDIA_PLAYER_ENTITY,
"media_content_type": "channel",
"media_content_id": "7777",
},
blocking=False,
)
await hass.services.async_call(
"media_player",
"play_media",
{
"entity_id": MEDIA_PLAYER_ENTITY,
"media_content_type": "channel",
"media_content_id": "11",
},
blocking=True,
)
assert mock_api.send_key_command.call_count == 2
assert await hass.services.async_call(
"media_player",
"play_media",
{
"entity_id": MEDIA_PLAYER_ENTITY,
"media_content_type": "url",
"media_content_id": "https://www.youtube.com",
},
blocking=True,
)
mock_api.send_launch_app_command.assert_called_with("https://www.youtube.com")
with pytest.raises(ValueError):
assert await hass.services.async_call(
"media_player",
"play_media",
{
"entity_id": MEDIA_PLAYER_ENTITY,
"media_content_type": "channel",
"media_content_id": "abc",
},
blocking=True,
)
with pytest.raises(ValueError):
assert await hass.services.async_call(
"media_player",
"play_media",
{
"entity_id": MEDIA_PLAYER_ENTITY,
"media_content_type": "music",
"media_content_id": "invalid",
},
blocking=True,
)
async def test_media_player_connection_closed(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test media_player raise HomeAssistantError if ConnectionClosed."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_api.send_key_command.side_effect = ConnectionClosed()
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
"media_player",
"media_pause",
{"entity_id": MEDIA_PLAYER_ENTITY},
blocking=True,
)
mock_api.send_launch_app_command.side_effect = ConnectionClosed()
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
"media_player",
"play_media",
{
"entity_id": MEDIA_PLAYER_ENTITY,
"media_content_type": "channel",
"media_content_id": "1",
},
blocking=True,
)

47
tests/components/androidtv_remote/test_remote.py
View File

@ -1,5 +1,4 @@
"""Tests for the Android TV Remote remote platform."""
from collections.abc import Callable
from unittest.mock import MagicMock, call
from androidtvremote2 import ConnectionClosed
@ -19,49 +18,25 @@ async def test_remote_receives_push_updates(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test the Android TV Remote receives push updates and state is updated."""
is_on_updated_callback: Callable | None = None
current_app_updated_callback: Callable | None = None
is_available_updated_callback: Callable | None = None
def mocked_add_is_on_updated_callback(callback: Callable):
nonlocal is_on_updated_callback
is_on_updated_callback = callback
def mocked_add_current_app_updated_callback(callback: Callable):
nonlocal current_app_updated_callback
current_app_updated_callback = callback
def mocked_add_is_available_updated_callback(callback: Callable):
nonlocal is_available_updated_callback
is_available_updated_callback = callback
mock_api.add_is_on_updated_callback.side_effect = mocked_add_is_on_updated_callback
mock_api.add_current_app_updated_callback.side_effect = (
mocked_add_current_app_updated_callback
)
mock_api.add_is_available_updated_callback.side_effect = (
mocked_add_is_available_updated_callback
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
is_on_updated_callback(False)
mock_api._on_is_on_updated(False)
assert hass.states.is_state(REMOTE_ENTITY, STATE_OFF)
is_on_updated_callback(True)
mock_api._on_is_on_updated(True)
assert hass.states.is_state(REMOTE_ENTITY, STATE_ON)
current_app_updated_callback("activity1")
mock_api._on_current_app_updated("activity1")
assert (
hass.states.get(REMOTE_ENTITY).attributes.get("current_activity") == "activity1"
)
is_available_updated_callback(False)
mock_api._on_is_available_updated(False)
assert hass.states.is_state(REMOTE_ENTITY, STATE_UNAVAILABLE)
is_available_updated_callback(True)
mock_api._on_is_available_updated(True)
assert hass.states.is_state(REMOTE_ENTITY, STATE_ON)
@ -69,14 +44,6 @@ async def test_remote_toggles(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: MagicMock
) -> None:
"""Test the Android TV Remote toggles."""
is_on_updated_callback: Callable | None = None
def mocked_add_is_on_updated_callback(callback: Callable):
nonlocal is_on_updated_callback
is_on_updated_callback = callback
mock_api.add_is_on_updated_callback.side_effect = mocked_add_is_on_updated_callback
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
@ -87,7 +54,7 @@ async def test_remote_toggles(
{"entity_id": REMOTE_ENTITY},
blocking=True,
)
is_on_updated_callback(False)
mock_api._on_is_on_updated(False)
mock_api.send_key_command.assert_called_with("POWER", "SHORT")
@ -97,7 +64,7 @@ async def test_remote_toggles(
{"entity_id": REMOTE_ENTITY},
blocking=True,
)
is_on_updated_callback(True)
mock_api._on_is_on_updated(True)
mock_api.send_key_command.assert_called_with("POWER", "SHORT")
assert mock_api.send_key_command.call_count == 2