mirror of
https://github.com/home-assistant/core.git
synced 2025-12-26 22:18:40 +00:00
Register 'androidtv.download' and 'androidtv.upload' services (#30086)
* Add tests * Add FileSync test * Fill in services.yaml for 'androidtv.adb_filesync' service * Update example paths in services.yaml * Bump androidtv to 0.0.37 * Bump androidtv to 0.0.37 * Bump androidtv to 0.0.37 * Import LockNotAcquiredException * Import LockNotAcquiredException from androidtv.exceptions * Rename 'host' to 'address' * Add a logging statement when an ADB command is skipped * Check hass.config.is_allowed_path(local_path) * Add return * Fix pylint * Reduce duplicated code (AndroidTVDevice vs. FireTVDevice) * Split 'adb_filesync' service into 'download' and 'upload' services * Don't use '.get()' for required data; return if the services are already registered * Replace "command" with ATTR_COMMAND * Don't try to connect to a device if it is a duplicate
This commit is contained in:
Jeff Irion
committed by
Paulus Schoutsen
Paulus Schoutsen
parent
e88bfda2a8
commit
5ec5df77cc
@@ -1,11 +1,21 @@
|
||||
"""The tests for the androidtv platform."""
|
||||
import logging
|
||||
from unittest.mock import patch
|
||||
|
||||
from androidtv.exceptions import LockNotAcquiredException
|
||||
|
||||
from homeassistant.components.androidtv.media_player import (
|
||||
ANDROIDTV_DOMAIN,
|
||||
ATTR_COMMAND,
|
||||
ATTR_DEVICE_PATH,
|
||||
ATTR_LOCAL_PATH,
|
||||
CONF_ADB_SERVER_IP,
|
||||
CONF_ADBKEY,
|
||||
CONF_APPS,
|
||||
KEYS,
|
||||
SERVICE_ADB_COMMAND,
|
||||
SERVICE_DOWNLOAD,
|
||||
SERVICE_UPLOAD,
|
||||
)
|
||||
from homeassistant.components.media_player.const import (
|
||||
ATTR_INPUT_SOURCE,
|
||||
@@ -70,7 +80,7 @@ CONFIG_FIRETV_ADB_SERVER = {
|
||||
}
|
||||
|
||||
|
||||
def _setup(hass, config):
|
||||
def _setup(config):
|
||||
"""Perform common setup tasks for the tests."""
|
||||
if CONF_ADB_SERVER_IP not in config[DOMAIN]:
|
||||
patch_key = "python"
|
||||
@@ -93,7 +103,7 @@ async def _test_reconnect(hass, caplog, config):
|
||||
|
||||
https://developers.home-assistant.io/docs/en/integration_quality_scale_index.html
|
||||
"""
|
||||
patch_key, entity_id = _setup(hass, config)
|
||||
patch_key, entity_id = _setup(config)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
@@ -164,7 +174,7 @@ async def _test_adb_shell_returns_none(hass, config):
|
||||
|
||||
The state should be `None` and the device should be unavailable.
|
||||
"""
|
||||
patch_key, entity_id = _setup(hass, config)
|
||||
patch_key, entity_id = _setup(config)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
@@ -272,7 +282,7 @@ async def test_setup_with_adbkey(hass):
|
||||
"""Test that setup succeeds when using an ADB key."""
|
||||
config = CONFIG_ANDROIDTV_PYTHON_ADB.copy()
|
||||
config[DOMAIN][CONF_ADBKEY] = hass.config.path("user_provided_adbkey")
|
||||
patch_key, entity_id = _setup(hass, config)
|
||||
patch_key, entity_id = _setup(config)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
@@ -290,7 +300,7 @@ async def _test_sources(hass, config0):
|
||||
"""Test that sources (i.e., apps) are handled correctly for Android TV and Fire TV devices."""
|
||||
config = config0.copy()
|
||||
config[DOMAIN][CONF_APPS] = {"com.app.test1": "TEST 1"}
|
||||
patch_key, entity_id = _setup(hass, config)
|
||||
patch_key, entity_id = _setup(config)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
@@ -362,7 +372,7 @@ async def _test_select_source(hass, config0, source, expected_arg, method_patch)
|
||||
"""Test that the methods for launching and stopping apps are called correctly when selecting a source."""
|
||||
config = config0.copy()
|
||||
config[DOMAIN][CONF_APPS] = {"com.app.test1": "TEST 1"}
|
||||
patch_key, entity_id = _setup(hass, config)
|
||||
patch_key, entity_id = _setup(config)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
@@ -519,7 +529,7 @@ async def test_firetv_select_source_stop_app_id_no_name(hass):
|
||||
|
||||
async def _test_setup_fail(hass, config):
|
||||
"""Test that the entity is not created when the ADB connection is not established."""
|
||||
patch_key, entity_id = _setup(hass, config)
|
||||
patch_key, entity_id = _setup(config)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(False)[
|
||||
patch_key
|
||||
@@ -569,14 +579,216 @@ async def test_setup_two_devices(hass):
|
||||
|
||||
async def test_setup_same_device_twice(hass):
|
||||
"""Test that setup succeeds with a duplicated config entry."""
|
||||
patch_key, entity_id = _setup(CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
|
||||
assert hass.services.has_service(ANDROIDTV_DOMAIN, SERVICE_ADB_COMMAND)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
|
||||
async def test_adb_command(hass):
|
||||
"""Test sending a command via the `androidtv.adb_command` service."""
|
||||
patch_key, entity_id = _setup(CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
command = "test command"
|
||||
response = "test response"
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
with patch(
|
||||
"androidtv.basetv.BaseTV.adb_shell", return_value=response
|
||||
) as patch_shell:
|
||||
await hass.services.async_call(
|
||||
ANDROIDTV_DOMAIN,
|
||||
SERVICE_ADB_COMMAND,
|
||||
{ATTR_ENTITY_ID: entity_id, ATTR_COMMAND: command},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
patch_shell.assert_called_with(command)
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
assert state.attributes["adb_response"] == response
|
||||
|
||||
|
||||
async def test_adb_command_key(hass):
|
||||
"""Test sending a key command via the `androidtv.adb_command` service."""
|
||||
patch_key = "server"
|
||||
entity_id = "media_player.android_tv"
|
||||
command = "HOME"
|
||||
response = None
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
with patch(
|
||||
"androidtv.basetv.BaseTV.adb_shell", return_value=response
|
||||
) as patch_shell:
|
||||
await hass.services.async_call(
|
||||
ANDROIDTV_DOMAIN,
|
||||
SERVICE_ADB_COMMAND,
|
||||
{ATTR_ENTITY_ID: entity_id, ATTR_COMMAND: command},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
patch_shell.assert_called_with(f"input keyevent {KEYS[command]}")
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
assert state.attributes["adb_response"] is None
|
||||
|
||||
|
||||
async def test_adb_command_get_properties(hass):
|
||||
"""Test sending the "GET_PROPERTIES" command via the `androidtv.adb_command` service."""
|
||||
patch_key = "server"
|
||||
entity_id = "media_player.android_tv"
|
||||
command = "GET_PROPERTIES"
|
||||
response = {"test key": "test value"}
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
with patch(
|
||||
"androidtv.androidtv.AndroidTV.get_properties_dict", return_value=response
|
||||
) as patch_get_props:
|
||||
await hass.services.async_call(
|
||||
ANDROIDTV_DOMAIN,
|
||||
SERVICE_ADB_COMMAND,
|
||||
{ATTR_ENTITY_ID: entity_id, ATTR_COMMAND: command},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
patch_get_props.assert_called()
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
assert state.attributes["adb_response"] == str(response)
|
||||
|
||||
|
||||
async def test_update_lock_not_acquired(hass):
|
||||
"""Test that the state does not get updated when a `LockNotAcquiredException` is raised."""
|
||||
patch_key, entity_id = _setup(CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
with patchers.patch_shell("")[patch_key]:
|
||||
await hass.helpers.entity_component.async_update_entity(entity_id)
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
with patch(
|
||||
"androidtv.androidtv.AndroidTV.update", side_effect=LockNotAcquiredException
|
||||
):
|
||||
with patchers.patch_shell("1")[patch_key]:
|
||||
await hass.helpers.entity_component.async_update_entity(entity_id)
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
with patchers.patch_shell("1")[patch_key]:
|
||||
await hass.helpers.entity_component.async_update_entity(entity_id)
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
assert state.state == STATE_IDLE
|
||||
|
||||
|
||||
async def test_download(hass):
|
||||
"""Test the `androidtv.download` service."""
|
||||
patch_key, entity_id = _setup(CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
device_path = "device/path"
|
||||
local_path = "local/path"
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
# Failed download because path is not whitelisted
|
||||
with patch("androidtv.basetv.BaseTV.adb_pull") as patch_pull:
|
||||
await hass.services.async_call(
|
||||
ANDROIDTV_DOMAIN,
|
||||
SERVICE_DOWNLOAD,
|
||||
{
|
||||
ATTR_ENTITY_ID: entity_id,
|
||||
ATTR_DEVICE_PATH: device_path,
|
||||
ATTR_LOCAL_PATH: local_path,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
patch_pull.assert_not_called()
|
||||
|
||||
# Successful download
|
||||
with patch("androidtv.basetv.BaseTV.adb_pull") as patch_pull, patch.object(
|
||||
hass.config, "is_allowed_path", return_value=True
|
||||
):
|
||||
await hass.services.async_call(
|
||||
ANDROIDTV_DOMAIN,
|
||||
SERVICE_DOWNLOAD,
|
||||
{
|
||||
ATTR_ENTITY_ID: entity_id,
|
||||
ATTR_DEVICE_PATH: device_path,
|
||||
ATTR_LOCAL_PATH: local_path,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
patch_pull.assert_called_with(local_path, device_path)
|
||||
|
||||
|
||||
async def test_upload(hass):
|
||||
"""Test the `androidtv.upload` service."""
|
||||
patch_key, entity_id = _setup(CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
device_path = "device/path"
|
||||
local_path = "local/path"
|
||||
|
||||
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
|
||||
patch_key
|
||||
], patchers.patch_shell("")[patch_key]:
|
||||
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
|
||||
|
||||
# Failed upload because path is not whitelisted
|
||||
with patch("androidtv.basetv.BaseTV.adb_push") as patch_push:
|
||||
await hass.services.async_call(
|
||||
ANDROIDTV_DOMAIN,
|
||||
SERVICE_UPLOAD,
|
||||
{
|
||||
ATTR_ENTITY_ID: entity_id,
|
||||
ATTR_DEVICE_PATH: device_path,
|
||||
ATTR_LOCAL_PATH: local_path,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
patch_push.assert_not_called()
|
||||
|
||||
# Successful upload
|
||||
with patch("androidtv.basetv.BaseTV.adb_push") as patch_push, patch.object(
|
||||
hass.config, "is_allowed_path", return_value=True
|
||||
):
|
||||
await hass.services.async_call(
|
||||
ANDROIDTV_DOMAIN,
|
||||
SERVICE_UPLOAD,
|
||||
{
|
||||
ATTR_ENTITY_ID: entity_id,
|
||||
ATTR_DEVICE_PATH: device_path,
|
||||
ATTR_LOCAL_PATH: local_path,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
patch_push.assert_called_with(local_path, device_path)
|
||||
|
||||
Reference in New Issue
Block a user