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

Convert Android TV integration to async (#37510)

Browse Source 
* Convert Android TV integration to async

* pylint

* Remove unused test code

* Require async versions of androidtv and adb-shell

* Cleanup

* Remove commented out code

* Use constants SHELL_RESPONSE_OFF and SHELL_RESPONSE_STANDBY
This commit is contained in:
Jeff Irion 2020-07-05 12:13:08 -07:00 committed by GitHub
parent e3aa4679a0
commit 10893f6246
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 340 additions and 226 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,8 +3,8 @@
"name": "Android TV", "name": "Android TV",
"documentation": "https://www.home-assistant.io/integrations/androidtv", "documentation": "https://www.home-assistant.io/integrations/androidtv",
"requirements": [ "requirements": [
"adb-shell==0.1.3", "adb-shell[async]==0.2.0",
"androidtv==0.0.43", "androidtv[async]==0.0.45",
"pure-python-adb==0.2.2.dev0" "pure-python-adb==0.2.2.dev0"
], ],
"codeowners": ["@JeffLIrion"] "codeowners": ["@JeffLIrion"]

192
homeassistant/components/androidtv/media_player.py
View File

@ -5,15 +5,18 @@ import logging
import os import os
from adb_shell.auth.keygen import keygen from adb_shell.auth.keygen import keygen
from adb_shell.auth.sign_pythonrsa import PythonRSASigner
from adb_shell.exceptions import ( from adb_shell.exceptions import (
AdbTimeoutError,
InvalidChecksumError, InvalidChecksumError,
InvalidCommandError, InvalidCommandError,
InvalidResponseError, InvalidResponseError,
TcpTimeoutException, TcpTimeoutException,
) )
from androidtv import ha_state_detection_rules_validator, setup from androidtv import ha_state_detection_rules_validator
from androidtv.constants import APPS, KEYS from androidtv.constants import APPS, KEYS
from androidtv.exceptions import LockNotAcquiredException from androidtv.exceptions import LockNotAcquiredException
from androidtv.setup_async import setup
import voluptuous as vol import voluptuous as vol
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerEntity from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerEntity
@ -163,7 +166,7 @@ ANDROIDTV_STATES = {
def setup_androidtv(hass, config): def setup_androidtv(hass, config):
"""Generate an ADB key (if needed) and connect to the Android TV / Fire TV.""" """Generate an ADB key (if needed) and load it."""
adbkey = config.get(CONF_ADBKEY, hass.config.path(STORAGE_DIR, "androidtv_adbkey")) adbkey = config.get(CONF_ADBKEY, hass.config.path(STORAGE_DIR, "androidtv_adbkey"))
if CONF_ADB_SERVER_IP not in config: if CONF_ADB_SERVER_IP not in config:
# Use "adb_shell" (Python ADB implementation) # Use "adb_shell" (Python ADB implementation)
@ -171,24 +174,18 @@ def setup_androidtv(hass, config):
# Generate ADB key files # Generate ADB key files
keygen(adbkey) keygen(adbkey)
# Load the ADB key
with open(adbkey) as priv_key:
priv = priv_key.read()
signer = PythonRSASigner("", priv)
adb_log = f"using Python ADB implementation with adbkey='{adbkey}'" adb_log = f"using Python ADB implementation with adbkey='{adbkey}'"
else: else:
# Use "pure-python-adb" (communicate with ADB server) # Use "pure-python-adb" (communicate with ADB server)
signer = None
adb_log = f"using ADB server at {config[CONF_ADB_SERVER_IP]}:{config[CONF_ADB_SERVER_PORT]}" adb_log = f"using ADB server at {config[CONF_ADB_SERVER_IP]}:{config[CONF_ADB_SERVER_PORT]}"
aftv = setup( return adbkey, signer, adb_log
config[CONF_HOST],
config[CONF_PORT],
adbkey,
config.get(CONF_ADB_SERVER_IP, ""),
config[CONF_ADB_SERVER_PORT],
config[CONF_STATE_DETECTION_RULES],
config[CONF_DEVICE_CLASS],
10.0,
)
return aftv, adb_log
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
@ -201,7 +198,21 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
_LOGGER.warning("Platform already setup on %s, skipping", address) _LOGGER.warning("Platform already setup on %s, skipping", address)
return return
aftv, adb_log = await hass.async_add_executor_job(setup_androidtv, hass, config) adbkey, signer, adb_log = await hass.async_add_executor_job(
setup_androidtv, hass, config
)
aftv = await setup(
config[CONF_HOST],
config[CONF_PORT],
adbkey,
config.get(CONF_ADB_SERVER_IP, ""),
config[CONF_ADB_SERVER_PORT],
config[CONF_STATE_DETECTION_RULES],
config[CONF_DEVICE_CLASS],
10.0,
signer,
)
if not aftv.available: if not aftv.available:
# Determine the name that will be used for the device in the log # Determine the name that will be used for the device in the log
@ -246,7 +257,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
platform = entity_platform.current_platform.get() platform = entity_platform.current_platform.get()
def service_adb_command(service): async def service_adb_command(service):
"""Dispatch service calls to target entities.""" """Dispatch service calls to target entities."""
cmd = service.data[ATTR_COMMAND] cmd = service.data[ATTR_COMMAND]
entity_id = service.data[ATTR_ENTITY_ID] entity_id = service.data[ATTR_ENTITY_ID]
@ -257,7 +268,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
] ]
for target_device in target_devices: for target_device in target_devices:
output = target_device.adb_command(cmd) output = await target_device.adb_command(cmd)
# log the output, if there is any # log the output, if there is any
if output: if output:
@ -276,10 +287,10 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
) )
platform.async_register_entity_service( platform.async_register_entity_service(
SERVICE_LEARN_SENDEVENT, {}, "learn_sendevent", SERVICE_LEARN_SENDEVENT, {}, "learn_sendevent"
) )
def service_download(service): async def service_download(service):
"""Download a file from your Android TV / Fire TV device to your Home Assistant instance.""" """Download a file from your Android TV / Fire TV device to your Home Assistant instance."""
local_path = service.data[ATTR_LOCAL_PATH] local_path = service.data[ATTR_LOCAL_PATH]
if not hass.config.is_allowed_path(local_path): if not hass.config.is_allowed_path(local_path):
@ -294,7 +305,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
if dev.entity_id in entity_id if dev.entity_id in entity_id
][0] ][0]
target_device.adb_pull(local_path, device_path) await target_device.adb_pull(local_path, device_path)
hass.services.async_register( hass.services.async_register(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
@ -303,7 +314,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
schema=SERVICE_DOWNLOAD_SCHEMA, schema=SERVICE_DOWNLOAD_SCHEMA,
) )
def service_upload(service): async def service_upload(service):
"""Upload a file from your Home Assistant instance to an Android TV / Fire TV device.""" """Upload a file from your Home Assistant instance to an Android TV / Fire TV device."""
local_path = service.data[ATTR_LOCAL_PATH] local_path = service.data[ATTR_LOCAL_PATH]
if not hass.config.is_allowed_path(local_path): if not hass.config.is_allowed_path(local_path):
@ -319,7 +330,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
] ]
for target_device in target_devices: for target_device in target_devices:
target_device.adb_push(local_path, device_path) await target_device.adb_push(local_path, device_path)
hass.services.async_register( hass.services.async_register(
ANDROIDTV_DOMAIN, SERVICE_UPLOAD, service_upload, schema=SERVICE_UPLOAD_SCHEMA ANDROIDTV_DOMAIN, SERVICE_UPLOAD, service_upload, schema=SERVICE_UPLOAD_SCHEMA
@ -337,13 +348,13 @@ def adb_decorator(override_available=False):
"""Wrap the provided ADB method and catch exceptions.""" """Wrap the provided ADB method and catch exceptions."""
@functools.wraps(func) @functools.wraps(func)
def _adb_exception_catcher(self, *args, **kwargs): async def _adb_exception_catcher(self, *args, **kwargs):
"""Call an ADB-related method and catch exceptions.""" """Call an ADB-related method and catch exceptions."""
if not self.available and not override_available: if not self.available and not override_available:
return None return None
try: try:
return func(self, *args, **kwargs) return await func(self, *args, **kwargs)
except LockNotAcquiredException: except LockNotAcquiredException:
# If the ADB lock could not be acquired, skip this command # If the ADB lock could not be acquired, skip this command
_LOGGER.info( _LOGGER.info(
@ -356,7 +367,7 @@ def adb_decorator(override_available=False):
"establishing attempt in the next update. Error: %s", "establishing attempt in the next update. Error: %s",
err, err,
) )
self.aftv.adb_close() await self.aftv.adb_close()
self._available = False # pylint: disable=protected-access self._available = False # pylint: disable=protected-access
return None return None
@ -403,6 +414,7 @@ class ADBDevice(MediaPlayerEntity):
if not self.aftv.adb_server_ip: if not self.aftv.adb_server_ip:
# Using "adb_shell" (Python ADB implementation) # Using "adb_shell" (Python ADB implementation)
self.exceptions = ( self.exceptions = (
AdbTimeoutError,
AttributeError, AttributeError,
BrokenPipeError, BrokenPipeError,
ConnectionResetError, ConnectionResetError,
@ -479,64 +491,60 @@ class ADBDevice(MediaPlayerEntity):
"""Return the device unique id.""" """Return the device unique id."""
return self._unique_id return self._unique_id
@adb_decorator()
async def async_get_media_image(self): async def async_get_media_image(self):
"""Fetch current playing image.""" """Fetch current playing image."""
if not self._screencap or self.state in [STATE_OFF, None] or not self.available: if not self._screencap or self.state in [STATE_OFF, None] or not self.available:
return None, None return None, None
media_data = await self.hass.async_add_executor_job(self.get_raw_media_data) media_data = await self.aftv.adb_screencap()
if media_data: if media_data:
return media_data, "image/png" return media_data, "image/png"
return None, None return None, None
@adb_decorator() @adb_decorator()
def get_raw_media_data(self): async def async_media_play(self):
"""Raw image data."""
return self.aftv.adb_screencap()
@adb_decorator()
def media_play(self):
"""Send play command.""" """Send play command."""
self.aftv.media_play() await self.aftv.media_play()
@adb_decorator() @adb_decorator()
def media_pause(self): async def async_media_pause(self):
"""Send pause command.""" """Send pause command."""
self.aftv.media_pause() await self.aftv.media_pause()
@adb_decorator() @adb_decorator()
def media_play_pause(self): async def async_media_play_pause(self):
"""Send play/pause command.""" """Send play/pause command."""
self.aftv.media_play_pause() await self.aftv.media_play_pause()
@adb_decorator() @adb_decorator()
def turn_on(self): async def async_turn_on(self):
"""Turn on the device.""" """Turn on the device."""
if self.turn_on_command: if self.turn_on_command:
self.aftv.adb_shell(self.turn_on_command) await self.aftv.adb_shell(self.turn_on_command)
else: else:
self.aftv.turn_on() await self.aftv.turn_on()
@adb_decorator() @adb_decorator()
def turn_off(self): async def async_turn_off(self):
"""Turn off the device.""" """Turn off the device."""
if self.turn_off_command: if self.turn_off_command:
self.aftv.adb_shell(self.turn_off_command) await self.aftv.adb_shell(self.turn_off_command)
else: else:
self.aftv.turn_off() await self.aftv.turn_off()
@adb_decorator() @adb_decorator()
def media_previous_track(self): async def async_media_previous_track(self):
"""Send previous track command (results in rewind).""" """Send previous track command (results in rewind)."""
self.aftv.media_previous_track() await self.aftv.media_previous_track()
@adb_decorator() @adb_decorator()
def media_next_track(self): async def async_media_next_track(self):
"""Send next track command (results in fast-forward).""" """Send next track command (results in fast-forward)."""
self.aftv.media_next_track() await self.aftv.media_next_track()
@adb_decorator() @adb_decorator()
def select_source(self, source): async def async_select_source(self, source):
"""Select input source. """Select input source.
If the source starts with a '!', then it will close the app instead of If the source starts with a '!', then it will close the app instead of
@ -544,62 +552,58 @@ class ADBDevice(MediaPlayerEntity):
""" """
if isinstance(source, str): if isinstance(source, str):
if not source.startswith("!"): if not source.startswith("!"):
self.aftv.launch_app(self._app_name_to_id.get(source, source)) await self.aftv.launch_app(self._app_name_to_id.get(source, source))
else: else:
source_ = source[1:].lstrip() source_ = source[1:].lstrip()
self.aftv.stop_app(self._app_name_to_id.get(source_, source_)) await self.aftv.stop_app(self._app_name_to_id.get(source_, source_))
@adb_decorator() @adb_decorator()
def adb_command(self, cmd): async def adb_command(self, cmd):
"""Send an ADB command to an Android TV / Fire TV device.""" """Send an ADB command to an Android TV / Fire TV device."""
key = self._keys.get(cmd) key = self._keys.get(cmd)
if key: if key:
self.aftv.adb_shell(f"input keyevent {key}") await self.aftv.adb_shell(f"input keyevent {key}")
self._adb_response = None
self.schedule_update_ha_state()
return return
if cmd == "GET_PROPERTIES": if cmd == "GET_PROPERTIES":
self._adb_response = str(self.aftv.get_properties_dict()) self._adb_response = str(await self.aftv.get_properties_dict())
self.schedule_update_ha_state() self.async_write_ha_state()
return self._adb_response return self._adb_response
try: try:
response = self.aftv.adb_shell(cmd) response = await self.aftv.adb_shell(cmd)
except UnicodeDecodeError: except UnicodeDecodeError:
self._adb_response = None
self.schedule_update_ha_state()
return return
if isinstance(response, str) and response.strip(): if isinstance(response, str) and response.strip():
self._adb_response = response.strip() self._adb_response = response.strip()
else: self.async_write_ha_state()
self._adb_response = None
self.schedule_update_ha_state()
return self._adb_response return self._adb_response
@adb_decorator() @adb_decorator()
def learn_sendevent(self): async def learn_sendevent(self):
"""Translate a key press on a remote to ADB 'sendevent' commands.""" """Translate a key press on a remote to ADB 'sendevent' commands."""
output = self.aftv.learn_sendevent() output = await self.aftv.learn_sendevent()
if output: if output:
self._adb_response = output self._adb_response = output
self.schedule_update_ha_state() self.async_write_ha_state()
msg = f"Output from service '{SERVICE_LEARN_SENDEVENT}' from {self.entity_id}: '{output}'" msg = f"Output from service '{SERVICE_LEARN_SENDEVENT}' from {self.entity_id}: '{output}'"
self.hass.components.persistent_notification.create(msg, title="Android TV") self.hass.components.persistent_notification.async_create(
msg, title="Android TV",
)
_LOGGER.info("%s", msg) _LOGGER.info("%s", msg)
@adb_decorator() @adb_decorator()
def adb_pull(self, local_path, device_path): async def adb_pull(self, local_path, device_path):
"""Download a file from your Android TV / Fire TV device to your Home Assistant instance.""" """Download a file from your Android TV / Fire TV device to your Home Assistant instance."""
self.aftv.adb_pull(local_path, device_path) await self.aftv.adb_pull(local_path, device_path)
@adb_decorator() @adb_decorator()
def adb_push(self, local_path, device_path): async def adb_push(self, local_path, device_path):
"""Upload a file from your Home Assistant instance to an Android TV / Fire TV device.""" """Upload a file from your Home Assistant instance to an Android TV / Fire TV device."""
self.aftv.adb_push(local_path, device_path) await self.aftv.adb_push(local_path, device_path)
class AndroidTVDevice(ADBDevice): class AndroidTVDevice(ADBDevice):
@ -632,17 +636,12 @@ class AndroidTVDevice(ADBDevice):
self._volume_level = None self._volume_level = None
@adb_decorator(override_available=True) @adb_decorator(override_available=True)
def update(self): async def async_update(self):
"""Update the device state and, if necessary, re-connect.""" """Update the device state and, if necessary, re-connect."""
# Check if device is disconnected. # Check if device is disconnected.
if not self._available: if not self._available:
# Try to connect # Try to connect
self._available = self.aftv.adb_connect(always_log_errors=False) self._available = await self.aftv.adb_connect(always_log_errors=False)
# To be safe, wait until the next update to run ADB commands if
# using the Python ADB implementation.
if not self.aftv.adb_server_ip:
return
# If the ADB connection is not intact, don't update. # If the ADB connection is not intact, don't update.
if not self._available: if not self._available:
@ -656,7 +655,7 @@ class AndroidTVDevice(ADBDevice):
_, _,
self._is_volume_muted, self._is_volume_muted,
self._volume_level, self._volume_level,
) = self.aftv.update(self._get_sources) ) = await self.aftv.update(self._get_sources)
self._state = ANDROIDTV_STATES.get(state) self._state = ANDROIDTV_STATES.get(state)
if self._state is None: if self._state is None:
@ -689,53 +688,50 @@ class AndroidTVDevice(ADBDevice):
return self._volume_level return self._volume_level
@adb_decorator() @adb_decorator()
def media_stop(self): async def async_media_stop(self):
"""Send stop command.""" """Send stop command."""
self.aftv.media_stop() await self.aftv.media_stop()
@adb_decorator() @adb_decorator()
def mute_volume(self, mute): async def async_mute_volume(self, mute):
"""Mute the volume.""" """Mute the volume."""
self.aftv.mute_volume() await self.aftv.mute_volume()
@adb_decorator() @adb_decorator()
def set_volume_level(self, volume): async def async_set_volume_level(self, volume):
"""Set the volume level.""" """Set the volume level."""
self.aftv.set_volume_level(volume) await self.aftv.set_volume_level(volume)
@adb_decorator() @adb_decorator()
def volume_down(self): async def async_volume_down(self):
"""Send volume down command.""" """Send volume down command."""
self._volume_level = self.aftv.volume_down(self._volume_level) self._volume_level = await self.aftv.volume_down(self._volume_level)
@adb_decorator() @adb_decorator()
def volume_up(self): async def async_volume_up(self):
"""Send volume up command.""" """Send volume up command."""
self._volume_level = self.aftv.volume_up(self._volume_level) self._volume_level = await self.aftv.volume_up(self._volume_level)
class FireTVDevice(ADBDevice): class FireTVDevice(ADBDevice):
"""Representation of a Fire TV device.""" """Representation of a Fire TV device."""
@adb_decorator(override_available=True) @adb_decorator(override_available=True)
def update(self): async def async_update(self):
"""Update the device state and, if necessary, re-connect.""" """Update the device state and, if necessary, re-connect."""
# Check if device is disconnected. # Check if device is disconnected.
if not self._available: if not self._available:
# Try to connect # Try to connect
self._available = self.aftv.adb_connect(always_log_errors=False) self._available = await self.aftv.adb_connect(always_log_errors=False)
# To be safe, wait until the next update to run ADB commands if
# using the Python ADB implementation.
if not self.aftv.adb_server_ip:
return
# If the ADB connection is not intact, don't update. # If the ADB connection is not intact, don't update.
if not self._available: if not self._available:
return return
# Get the `state`, `current_app`, and `running_apps`. # Get the `state`, `current_app`, and `running_apps`.
state, self._current_app, running_apps = self.aftv.update(self._get_sources) state, self._current_app, running_apps = await self.aftv.update(
self._get_sources
)
self._state = ANDROIDTV_STATES.get(state) self._state = ANDROIDTV_STATES.get(state)
if self._state is None: if self._state is None:
@ -758,6 +754,6 @@ class FireTVDevice(ADBDevice):
return SUPPORT_FIRETV return SUPPORT_FIRETV
@adb_decorator() @adb_decorator()
def media_stop(self): async def async_media_stop(self):
"""Send stop (back) command.""" """Send stop (back) command."""
self.aftv.back() await self.aftv.back()

4
requirements_all.txt
View File

@ -130,7 +130,7 @@ adafruit-circuitpython-bmp280==3.1.1
adafruit-circuitpython-mcp230xx==2.2.2 adafruit-circuitpython-mcp230xx==2.2.2
# homeassistant.components.androidtv # homeassistant.components.androidtv
adb-shell==0.1.3 adb-shell[async]==0.2.0
# homeassistant.components.alarmdecoder # homeassistant.components.alarmdecoder
adext==0.3 adext==0.3
@ -249,7 +249,7 @@ ambiclimate==0.2.1
amcrest==1.7.0 amcrest==1.7.0
# homeassistant.components.androidtv # homeassistant.components.androidtv
androidtv==0.0.43 androidtv[async]==0.0.45
# homeassistant.components.anel_pwrctrl # homeassistant.components.anel_pwrctrl
anel_pwrctrl-homeassistant==0.0.1.dev2 anel_pwrctrl-homeassistant==0.0.1.dev2

4
requirements_test_all.txt
View File

@ -45,7 +45,7 @@ YesssSMS==0.4.1
abodepy==0.19.0 abodepy==0.19.0
# homeassistant.components.androidtv # homeassistant.components.androidtv
adb-shell==0.1.3 adb-shell[async]==0.2.0
# homeassistant.components.adguard # homeassistant.components.adguard
adguardhome==0.4.2 adguardhome==0.4.2
@ -131,7 +131,7 @@ airly==0.0.2
ambiclimate==0.2.1 ambiclimate==0.2.1
# homeassistant.components.androidtv # homeassistant.components.androidtv
androidtv==0.0.43 androidtv[async]==0.0.45
# homeassistant.components.apns # homeassistant.components.apns
apns2==0.3.0 apns2==0.3.0

136
tests/components/androidtv/patchers.py
View File

@ -2,134 +2,148 @@
from tests.async_mock import mock_open, patch from tests.async_mock import mock_open, patch
KEY_PYTHON = "python"
KEY_SERVER = "server"
class AdbDeviceTcpFake: ADB_DEVICE_TCP_ASYNC_FAKE = "AdbDeviceTcpAsyncFake"
"""A fake of the `adb_shell.adb_device.AdbDeviceTcp` class.""" DEVICE_ASYNC_FAKE = "DeviceAsyncFake"
class AdbDeviceTcpAsyncFake:
"""A fake of the `adb_shell.adb_device_async.AdbDeviceTcpAsync` class."""
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
"""Initialize a fake `adb_shell.adb_device.AdbDeviceTcp` instance.""" """Initialize a fake `adb_shell.adb_device_async.AdbDeviceTcpAsync` instance."""
self.available = False self.available = False
def close(self): async def close(self):
"""Close the socket connection.""" """Close the socket connection."""
self.available = False self.available = False
def connect(self, *args, **kwargs): async def connect(self, *args, **kwargs):
"""Try to connect to a device.""" """Try to connect to a device."""
raise NotImplementedError raise NotImplementedError
def shell(self, cmd): async def shell(self, cmd, *args, **kwargs):
"""Send an ADB shell command.""" """Send an ADB shell command."""
return None return None
class ClientFakeSuccess: class ClientAsyncFakeSuccess:
"""A fake of the `ppadb.client.Client` class when the connection and shell commands succeed.""" """A fake of the `ClientAsync` class when the connection and shell commands succeed."""
def __init__(self, host="127.0.0.1", port=5037): def __init__(self, host="127.0.0.1", port=5037):
"""Initialize a `ClientFakeSuccess` instance.""" """Initialize a `ClientAsyncFakeSuccess` instance."""
self._devices = [] self._devices = []
def devices(self): async def device(self, serial):
"""Get a list of the connected devices.""" """Mock the `ClientAsync.device` method when the device is connected via ADB."""
return self._devices device = DeviceAsyncFake(serial)
def device(self, serial):
"""Mock the `Client.device` method when the device is connected via ADB."""
device = DeviceFake(serial)
self._devices.append(device) self._devices.append(device)
return device return device
class ClientFakeFail: class ClientAsyncFakeFail:
"""A fake of the `ppadb.client.Client` class when the connection and shell commands fail.""" """A fake of the `ClientAsync` class when the connection and shell commands fail."""
def __init__(self, host="127.0.0.1", port=5037): def __init__(self, host="127.0.0.1", port=5037):
"""Initialize a `ClientFakeFail` instance.""" """Initialize a `ClientAsyncFakeFail` instance."""
self._devices = [] self._devices = []
def devices(self): async def device(self, serial):
"""Get a list of the connected devices.""" """Mock the `ClientAsync.device` method when the device is not connected via ADB."""
return self._devices
def device(self, serial):
"""Mock the `Client.device` method when the device is not connected via ADB."""
self._devices = [] self._devices = []
return None
class DeviceFake: class DeviceAsyncFake:
"""A fake of the `ppadb.device.Device` class.""" """A fake of the `DeviceAsync` class."""
def __init__(self, host): def __init__(self, host):
"""Initialize a `DeviceFake` instance.""" """Initialize a `DeviceAsyncFake` instance."""
self.host = host self.host = host
def get_serial_no(self): async def shell(self, cmd):
"""Get the serial number for the device (IP:PORT)."""
return self.host
def shell(self, cmd):
"""Send an ADB shell command.""" """Send an ADB shell command."""
raise NotImplementedError raise NotImplementedError
def patch_connect(success): def patch_connect(success):
"""Mock the `adb_shell.adb_device.AdbDeviceTcp` and `ppadb.client.Client` classes.""" """Mock the `adb_shell.adb_device_async.AdbDeviceTcpAsync` and `ClientAsync` classes."""
def connect_success_python(self, *args, **kwargs): async def connect_success_python(self, *args, **kwargs):
"""Mock the `AdbDeviceTcpFake.connect` method when it succeeds.""" """Mock the `AdbDeviceTcpAsyncFake.connect` method when it succeeds."""
self.available = True self.available = True
def connect_fail_python(self, *args, **kwargs): async def connect_fail_python(self, *args, **kwargs):
"""Mock the `AdbDeviceTcpFake.connect` method when it fails.""" """Mock the `AdbDeviceTcpAsyncFake.connect` method when it fails."""
raise OSError raise OSError
if success: if success:
return { return {
"python": patch( KEY_PYTHON: patch(
f"{__name__}.AdbDeviceTcpFake.connect", connect_success_python f"{__name__}.{ADB_DEVICE_TCP_ASYNC_FAKE}.connect",
connect_success_python,
),
KEY_SERVER: patch(
"androidtv.adb_manager.adb_manager_async.ClientAsync",
ClientAsyncFakeSuccess,
), ),
"server": patch("androidtv.adb_manager.Client", ClientFakeSuccess),
} }
return { return {
"python": patch(f"{__name__}.AdbDeviceTcpFake.connect", connect_fail_python), KEY_PYTHON: patch(
"server": patch("androidtv.adb_manager.Client", ClientFakeFail), f"{__name__}.{ADB_DEVICE_TCP_ASYNC_FAKE}.connect", connect_fail_python
),
KEY_SERVER: patch(
"androidtv.adb_manager.adb_manager_async.ClientAsync", ClientAsyncFakeFail
),
} }
def patch_shell(response=None, error=False): def patch_shell(response=None, error=False):
"""Mock the `AdbDeviceTcpFake.shell` and `DeviceFake.shell` methods.""" """Mock the `AdbDeviceTcpAsyncFake.shell` and `DeviceAsyncFake.shell` methods."""
def shell_success(self, cmd): async def shell_success(self, cmd, *args, **kwargs):
"""Mock the `AdbDeviceTcpFake.shell` and `DeviceFake.shell` methods when they are successful.""" """Mock the `AdbDeviceTcpAsyncFake.shell` and `DeviceAsyncFake.shell` methods when they are successful."""
self.shell_cmd = cmd self.shell_cmd = cmd
return response return response
def shell_fail_python(self, cmd): async def shell_fail_python(self, cmd, *args, **kwargs):
"""Mock the `AdbDeviceTcpFake.shell` method when it fails.""" """Mock the `AdbDeviceTcpAsyncFake.shell` method when it fails."""
self.shell_cmd = cmd self.shell_cmd = cmd
raise AttributeError raise AttributeError
def shell_fail_server(self, cmd): async def shell_fail_server(self, cmd):
"""Mock the `DeviceFake.shell` method when it fails.""" """Mock the `DeviceAsyncFake.shell` method when it fails."""
self.shell_cmd = cmd self.shell_cmd = cmd
raise ConnectionResetError raise ConnectionResetError
if not error: if not error:
return { return {
"python": patch(f"{__name__}.AdbDeviceTcpFake.shell", shell_success), KEY_PYTHON: patch(
"server": patch(f"{__name__}.DeviceFake.shell", shell_success), f"{__name__}.{ADB_DEVICE_TCP_ASYNC_FAKE}.shell", shell_success
),
KEY_SERVER: patch(f"{__name__}.{DEVICE_ASYNC_FAKE}.shell", shell_success),
} }
return { return {
"python": patch(f"{__name__}.AdbDeviceTcpFake.shell", shell_fail_python), KEY_PYTHON: patch(
"server": patch(f"{__name__}.DeviceFake.shell", shell_fail_server), f"{__name__}.{ADB_DEVICE_TCP_ASYNC_FAKE}.shell", shell_fail_python
),
KEY_SERVER: patch(f"{__name__}.{DEVICE_ASYNC_FAKE}.shell", shell_fail_server),
} }
PATCH_ADB_DEVICE_TCP = patch("androidtv.adb_manager.AdbDeviceTcp", AdbDeviceTcpFake) PATCH_ADB_DEVICE_TCP = patch(
PATCH_ANDROIDTV_OPEN = patch("androidtv.adb_manager.open", mock_open()) "androidtv.adb_manager.adb_manager_async.AdbDeviceTcpAsync", AdbDeviceTcpAsyncFake
)
PATCH_ANDROIDTV_OPEN = patch(
"homeassistant.components.androidtv.media_player.open", mock_open()
)
PATCH_KEYGEN = patch("homeassistant.components.androidtv.media_player.keygen") PATCH_KEYGEN = patch("homeassistant.components.androidtv.media_player.keygen")
PATCH_SIGNER = patch("androidtv.adb_manager.PythonRSASigner") PATCH_SIGNER = patch(
"homeassistant.components.androidtv.media_player.PythonRSASigner",
return_value="signer for testing",
)
def isfile(filepath): def isfile(filepath):
@ -144,7 +158,7 @@ PATCH_ACCESS = patch("os.access", return_value=True)
def patch_firetv_update(state, current_app, running_apps): def patch_firetv_update(state, current_app, running_apps):
"""Patch the `FireTV.update()` method.""" """Patch the `FireTV.update()` method."""
return patch( return patch(
"androidtv.firetv.FireTV.update", "androidtv.firetv.firetv_async.FireTVAsync.update",
return_value=(state, current_app, running_apps), return_value=(state, current_app, running_apps),
) )
@ -154,7 +168,7 @@ def patch_androidtv_update(
): ):
"""Patch the `AndroidTV.update()` method.""" """Patch the `AndroidTV.update()` method."""
return patch( return patch(
"androidtv.androidtv.AndroidTV.update", "androidtv.androidtv.androidtv_async.AndroidTVAsync.update",
return_value=( return_value=(
state, state,
current_app, current_app,
@ -166,5 +180,5 @@ def patch_androidtv_update(
) )
PATCH_LAUNCH_APP = patch("androidtv.basetv.BaseTV.launch_app") PATCH_LAUNCH_APP = patch("androidtv.basetv.basetv_async.BaseTVAsync.launch_app")
PATCH_STOP_APP = patch("androidtv.basetv.BaseTV.stop_app") PATCH_STOP_APP = patch("androidtv.basetv.basetv_async.BaseTVAsync.stop_app")

226
tests/components/androidtv/test_media_player.py
View File

@ -13,16 +13,32 @@ from homeassistant.components.androidtv.media_player import (
CONF_ADBKEY, CONF_ADBKEY,
CONF_APPS, CONF_APPS,
CONF_EXCLUDE_UNNAMED_APPS, CONF_EXCLUDE_UNNAMED_APPS,
CONF_TURN_OFF_COMMAND,
CONF_TURN_ON_COMMAND,
KEYS, KEYS,
SERVICE_ADB_COMMAND, SERVICE_ADB_COMMAND,
SERVICE_DOWNLOAD, SERVICE_DOWNLOAD,
SERVICE_LEARN_SENDEVENT, SERVICE_LEARN_SENDEVENT,
SERVICE_UPLOAD, SERVICE_UPLOAD,
) )
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE, ATTR_INPUT_SOURCE,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN, DOMAIN,
SERVICE_MEDIA_NEXT_TRACK,
SERVICE_MEDIA_PAUSE,
SERVICE_MEDIA_PLAY,
SERVICE_MEDIA_PLAY_PAUSE,
SERVICE_MEDIA_PREVIOUS_TRACK,
SERVICE_MEDIA_STOP,
SERVICE_SELECT_SOURCE, SERVICE_SELECT_SOURCE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_SET,
SERVICE_VOLUME_UP,
) )
from homeassistant.components.websocket_api.const import TYPE_RESULT from homeassistant.components.websocket_api.const import TYPE_RESULT
from homeassistant.const import ( from homeassistant.const import (
@ -31,7 +47,6 @@ from homeassistant.const import (
CONF_HOST, CONF_HOST,
CONF_NAME, CONF_NAME,
CONF_PLATFORM, CONF_PLATFORM,
SERVICE_VOLUME_SET,
STATE_OFF, STATE_OFF,
STATE_PLAYING, STATE_PLAYING,
STATE_STANDBY, STATE_STANDBY,
@ -39,9 +54,11 @@ from homeassistant.const import (
) )
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from . import patchers
from tests.async_mock import patch from tests.async_mock import patch
from tests.components.androidtv import patchers
SHELL_RESPONSE_OFF = ""
SHELL_RESPONSE_STANDBY = "1"
# Android TV device with Python ADB implementation # Android TV device with Python ADB implementation
CONFIG_ANDROIDTV_PYTHON_ADB = { CONFIG_ANDROIDTV_PYTHON_ADB = {
@ -113,7 +130,7 @@ async def _test_reconnect(hass, caplog, config):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[ ], patchers.patch_shell(SHELL_RESPONSE_OFF)[
patch_key patch_key
], patchers.PATCH_KEYGEN, patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER: ], patchers.PATCH_KEYGEN, patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
@ -141,23 +158,11 @@ async def _test_reconnect(hass, caplog, config):
assert caplog.record_tuples[1][1] == logging.WARNING assert caplog.record_tuples[1][1] == logging.WARNING
caplog.set_level(logging.DEBUG) caplog.set_level(logging.DEBUG)
with patchers.patch_connect(True)[patch_key], patchers.patch_shell("1")[ with patchers.patch_connect(True)[patch_key], patchers.patch_shell(
patch_key SHELL_RESPONSE_STANDBY
], patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER: )[patch_key], patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER:
# Update 1 will reconnect
await hass.helpers.entity_component.async_update_entity(entity_id) await hass.helpers.entity_component.async_update_entity(entity_id)
# If using an ADB server, the state will get updated; otherwise, the
# state will be the last known state
state = hass.states.get(entity_id)
if patch_key == "server":
assert state.state == STATE_STANDBY
else:
assert state.state == STATE_OFF
# Update 2 will update the state, regardless of which ADB connection
# method is used
await hass.helpers.entity_component.async_update_entity(entity_id)
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state is not None assert state is not None
assert state.state == STATE_STANDBY assert state.state == STATE_STANDBY
@ -185,7 +190,7 @@ async def _test_adb_shell_returns_none(hass, config):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[ ], patchers.patch_shell(SHELL_RESPONSE_OFF)[
patch_key patch_key
], patchers.PATCH_KEYGEN, patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER: ], patchers.PATCH_KEYGEN, patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
@ -294,7 +299,7 @@ async def test_setup_with_adbkey(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[ ], patchers.patch_shell(SHELL_RESPONSE_OFF)[
patch_key patch_key
], patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER, patchers.PATCH_ISFILE, patchers.PATCH_ACCESS: ], patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER, patchers.PATCH_ISFILE, patchers.PATCH_ACCESS:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
@ -311,13 +316,13 @@ async def _test_sources(hass, config0):
config[DOMAIN][CONF_APPS] = { config[DOMAIN][CONF_APPS] = {
"com.app.test1": "TEST 1", "com.app.test1": "TEST 1",
"com.app.test3": None, "com.app.test3": None,
"com.app.test4": "", "com.app.test4": SHELL_RESPONSE_OFF,
} }
patch_key, entity_id = _setup(config) patch_key, entity_id = _setup(config)
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
await hass.helpers.entity_component.async_update_entity(entity_id) await hass.helpers.entity_component.async_update_entity(entity_id)
@ -392,13 +397,13 @@ async def _test_exclude_sources(hass, config0, expected_sources):
config[DOMAIN][CONF_APPS] = { config[DOMAIN][CONF_APPS] = {
"com.app.test1": "TEST 1", "com.app.test1": "TEST 1",
"com.app.test3": None, "com.app.test3": None,
"com.app.test4": "", "com.app.test4": SHELL_RESPONSE_OFF,
} }
patch_key, entity_id = _setup(config) patch_key, entity_id = _setup(config)
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
await hass.helpers.entity_component.async_update_entity(entity_id) await hass.helpers.entity_component.async_update_entity(entity_id)
@ -467,7 +472,7 @@ async def _test_select_source(hass, config0, source, expected_arg, method_patch)
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
await hass.helpers.entity_component.async_update_entity(entity_id) await hass.helpers.entity_component.async_update_entity(entity_id)
@ -669,7 +674,7 @@ async def _test_setup_fail(hass, config):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(False)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(False)[
patch_key patch_key
], patchers.patch_shell("")[ ], patchers.patch_shell(SHELL_RESPONSE_OFF)[
patch_key patch_key
], patchers.PATCH_KEYGEN, patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER: ], patchers.PATCH_KEYGEN, patchers.PATCH_ANDROIDTV_OPEN, patchers.PATCH_SIGNER:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
@ -704,7 +709,7 @@ async def test_setup_two_devices(hass):
patch_key = "server" patch_key = "server"
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, config) assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -721,7 +726,7 @@ async def test_setup_same_device_twice(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
@ -731,7 +736,7 @@ async def test_setup_same_device_twice(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -744,12 +749,12 @@ async def test_adb_command(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
with patch( with patch(
"androidtv.basetv.BaseTV.adb_shell", return_value=response "androidtv.basetv.basetv_async.BaseTVAsync.adb_shell", return_value=response
) as patch_shell: ) as patch_shell:
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
@ -772,12 +777,12 @@ async def test_adb_command_unicode_decode_error(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
with patch( with patch(
"androidtv.basetv.BaseTV.adb_shell", "androidtv.basetv.basetv_async.BaseTVAsync.adb_shell",
side_effect=UnicodeDecodeError("utf-8", response, 0, len(response), "TEST"), side_effect=UnicodeDecodeError("utf-8", response, 0, len(response), "TEST"),
): ):
await hass.services.async_call( await hass.services.async_call(
@ -802,12 +807,12 @@ async def test_adb_command_key(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
with patch( with patch(
"androidtv.basetv.BaseTV.adb_shell", return_value=response "androidtv.basetv.basetv_async.BaseTVAsync.adb_shell", return_value=response
) as patch_shell: ) as patch_shell:
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
@ -831,12 +836,13 @@ async def test_adb_command_get_properties(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
with patch( with patch(
"androidtv.androidtv.AndroidTV.get_properties_dict", return_value=response "androidtv.androidtv.androidtv_async.AndroidTVAsync.get_properties_dict",
return_value=response,
) as patch_get_props: ) as patch_get_props:
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
@ -859,12 +865,13 @@ async def test_learn_sendevent(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
with patch( with patch(
"androidtv.basetv.BaseTV.learn_sendevent", return_value=response "androidtv.basetv.basetv_async.BaseTVAsync.learn_sendevent",
return_value=response,
) as patch_learn_sendevent: ) as patch_learn_sendevent:
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
@ -885,26 +892,27 @@ async def test_update_lock_not_acquired(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
with patchers.patch_shell("")[patch_key]: with patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
await hass.helpers.entity_component.async_update_entity(entity_id) await hass.helpers.entity_component.async_update_entity(entity_id)
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state is not None assert state is not None
assert state.state == STATE_OFF assert state.state == STATE_OFF
with patch( with patch(
"androidtv.androidtv.AndroidTV.update", side_effect=LockNotAcquiredException "androidtv.androidtv.androidtv_async.AndroidTVAsync.update",
side_effect=LockNotAcquiredException,
): ):
with patchers.patch_shell("1")[patch_key]: with patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key]:
await hass.helpers.entity_component.async_update_entity(entity_id) await hass.helpers.entity_component.async_update_entity(entity_id)
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state is not None assert state is not None
assert state.state == STATE_OFF assert state.state == STATE_OFF
with patchers.patch_shell("1")[patch_key]: with patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key]:
await hass.helpers.entity_component.async_update_entity(entity_id) await hass.helpers.entity_component.async_update_entity(entity_id)
state = hass.states.get(entity_id) state = hass.states.get(entity_id)
assert state is not None assert state is not None
@ -919,12 +927,12 @@ async def test_download(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
# Failed download because path is not whitelisted # Failed download because path is not whitelisted
with patch("androidtv.basetv.BaseTV.adb_pull") as patch_pull: with patch("androidtv.basetv.basetv_async.BaseTVAsync.adb_pull") as patch_pull:
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
SERVICE_DOWNLOAD, SERVICE_DOWNLOAD,
@ -938,9 +946,9 @@ async def test_download(hass):
patch_pull.assert_not_called() patch_pull.assert_not_called()
# Successful download # Successful download
with patch("androidtv.basetv.BaseTV.adb_pull") as patch_pull, patch.object( with patch(
hass.config, "is_allowed_path", return_value=True "androidtv.basetv.basetv_async.BaseTVAsync.adb_pull"
): ) as patch_pull, patch.object(hass.config, "is_allowed_path", return_value=True):
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
SERVICE_DOWNLOAD, SERVICE_DOWNLOAD,
@ -962,12 +970,12 @@ async def test_upload(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
# Failed upload because path is not whitelisted # Failed upload because path is not whitelisted
with patch("androidtv.basetv.BaseTV.adb_push") as patch_push: with patch("androidtv.basetv.basetv_async.BaseTVAsync.adb_push") as patch_push:
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
SERVICE_UPLOAD, SERVICE_UPLOAD,
@ -981,9 +989,9 @@ async def test_upload(hass):
patch_push.assert_not_called() patch_push.assert_not_called()
# Successful upload # Successful upload
with patch("androidtv.basetv.BaseTV.adb_push") as patch_push, patch.object( with patch(
hass.config, "is_allowed_path", return_value=True "androidtv.basetv.basetv_async.BaseTVAsync.adb_push"
): ) as patch_push, patch.object(hass.config, "is_allowed_path", return_value=True):
await hass.services.async_call( await hass.services.async_call(
ANDROIDTV_DOMAIN, ANDROIDTV_DOMAIN,
SERVICE_UPLOAD, SERVICE_UPLOAD,
@ -1003,17 +1011,17 @@ async def test_androidtv_volume_set(hass):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
with patch( with patch(
"androidtv.basetv.BaseTV.set_volume_level", return_value=0.5 "androidtv.basetv.basetv_async.BaseTVAsync.set_volume_level", return_value=0.5
) as patch_set_volume_level: ) as patch_set_volume_level:
await hass.services.async_call( await hass.services.async_call(
DOMAIN, DOMAIN,
SERVICE_VOLUME_SET, SERVICE_VOLUME_SET,
{ATTR_ENTITY_ID: entity_id, "volume_level": 0.5}, {ATTR_ENTITY_ID: entity_id, ATTR_MEDIA_VOLUME_LEVEL: 0.5},
blocking=True, blocking=True,
) )
@ -1029,7 +1037,7 @@ async def test_get_image(hass, hass_ws_client):
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[ with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key patch_key
], patchers.patch_shell("")[patch_key]: ], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER) assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -1038,7 +1046,9 @@ async def test_get_image(hass, hass_ws_client):
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
with patch("androidtv.basetv.BaseTV.adb_screencap", return_value=b"image"): with patch(
"androidtv.basetv.basetv_async.BaseTVAsync.adb_screencap", return_value=b"image"
):
await client.send_json( await client.send_json(
{"id": 5, "type": "media_player_thumbnail", "entity_id": entity_id} {"id": 5, "type": "media_player_thumbnail", "entity_id": entity_id}
) )
@ -1050,3 +1060,97 @@ async def test_get_image(hass, hass_ws_client):
assert msg["success"] assert msg["success"]
assert msg["result"]["content_type"] == "image/png" assert msg["result"]["content_type"] == "image/png"
assert msg["result"]["content"] == base64.b64encode(b"image").decode("utf-8") assert msg["result"]["content"] == base64.b64encode(b"image").decode("utf-8")
async def _test_service(
hass,
entity_id,
ha_service_name,
androidtv_method,
additional_service_data=None,
return_value=None,
):
"""Test generic Android TV media player entity service."""
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}", return_value=return_value
) as service_call:
await hass.services.async_call(
DOMAIN, ha_service_name, service_data=service_data, blocking=True,
)
assert service_call.called
async def test_services_androidtv(hass):
"""Test media player services for an Android TV device."""
patch_key, entity_id = _setup(CONFIG_ANDROIDTV_ADB_SERVER)
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[patch_key]:
with patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(
hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER
)
await hass.async_block_till_done()
with patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key]:
await _test_service(
hass, entity_id, SERVICE_MEDIA_NEXT_TRACK, "media_next_track"
)
await _test_service(hass, entity_id, SERVICE_MEDIA_PAUSE, "media_pause")
await _test_service(hass, entity_id, SERVICE_MEDIA_PLAY, "media_play")
await _test_service(
hass, entity_id, SERVICE_MEDIA_PLAY_PAUSE, "media_play_pause"
)
await _test_service(
hass, entity_id, SERVICE_MEDIA_PREVIOUS_TRACK, "media_previous_track"
)
await _test_service(hass, entity_id, SERVICE_MEDIA_STOP, "media_stop")
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_VOLUME_DOWN, "volume_down", return_value=0.1
)
await _test_service(
hass,
entity_id,
SERVICE_VOLUME_MUTE,
"mute_volume",
{ATTR_MEDIA_VOLUME_MUTED: False},
)
await _test_service(
hass,
entity_id,
SERVICE_VOLUME_SET,
"set_volume_level",
{ATTR_MEDIA_VOLUME_LEVEL: 0.5},
0.5,
)
await _test_service(
hass, entity_id, SERVICE_VOLUME_UP, "volume_up", return_value=0.2
)
async def test_services_firetv(hass):
"""Test media player services for a Fire TV device."""
patch_key, entity_id = _setup(CONFIG_FIRETV_ADB_SERVER)
config = CONFIG_FIRETV_ADB_SERVER.copy()
config[DOMAIN][CONF_TURN_OFF_COMMAND] = "test off"
config[DOMAIN][CONF_TURN_ON_COMMAND] = "test on"
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[patch_key]:
with patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
with patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key]:
await _test_service(hass, entity_id, SERVICE_MEDIA_STOP, "back")
await _test_service(hass, entity_id, SERVICE_TURN_OFF, "adb_shell")
await _test_service(hass, entity_id, SERVICE_TURN_ON, "adb_shell")