From 00052e850b9b5ef4b472dc0046e17649ed7968df Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Fri, 14 Jan 2022 20:48:45 -0800 Subject: [PATCH 1/9] Split `get_properties()` for Fire TV --- androidtv/firetv/base_firetv.py | 87 -------------------------------- androidtv/firetv/firetv_async.py | 31 +++++------- androidtv/firetv/firetv_sync.py | 30 +++++------ tests/test_firetv_async.py | 3 -- tests/test_firetv_sync.py | 3 -- 5 files changed, 25 insertions(+), 129 deletions(-) diff --git a/androidtv/firetv/base_firetv.py b/androidtv/firetv/base_firetv.py index 74c7536..6a64179 100644 --- a/androidtv/firetv/base_firetv.py +++ b/androidtv/firetv/base_firetv.py @@ -39,10 +39,6 @@ def __init__(self, host, port=5555, adbkey="", adb_server_ip="", adb_server_port def _fill_in_commands(self): """Fill in commands that are specific to Fire TV devices.""" - self._cmd_get_properties_lazy_running_apps = constants.CMD_FIRETV_PROPERTIES_LAZY_RUNNING_APPS - self._cmd_get_properties_lazy_no_running_apps = constants.CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS - self._cmd_get_properties_not_lazy_running_apps = constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_RUNNING_APPS - self._cmd_get_properties_not_lazy_no_running_apps = constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_NO_RUNNING_APPS self._cmd_current_app = constants.CMD_CURRENT_APP self._cmd_launch_app = constants.CMD_LAUNCH_APP @@ -223,86 +219,3 @@ def _update(self, screen_on, awake, wake_lock_size, current_app, media_session_s state = constants.STATE_PAUSED return state, current_app, running_apps, hdmi_input - - # ======================================================================= # - # # - # Properties # - # # - # ======================================================================= # - def _get_properties(self, output, get_running_apps=True): - """Get the properties needed for Home Assistant updates. - - This will send one of the following ADB commands: - - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_LAZY_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_NO_RUNNING_APPS` - - Parameters - ---------- - output : str, None - The output of the ADB command used to retrieve the properties - get_running_apps : bool - Whether or not to get the ``running_apps`` property - - Returns - ------- - screen_on : bool, None - Whether or not the device is on, or ``None`` if it was not determined - awake : bool, None - Whether or not the device is awake (screensaver is not running), or ``None`` if it was not determined - wake_lock_size : int, None - The size of the current wake lock, or ``None`` if it was not determined - current_app : str, None - The current app property, or ``None`` if it was not determined - media_session_state : int, None - The state from the output of ``dumpsys media_session``, or ``None`` if it was not determined - running_apps : list, None - A list of the running apps, or ``None`` if it was not determined - hdmi_input : str, None - The HDMI input, or ``None`` if it could not be determined - - """ - # ADB command was unsuccessful - if output is None: - return None, None, None, None, None, None, None - - # `screen_on` property - if not output: - return False, False, -1, None, None, None, None - screen_on = output[0] == "1" - - # `awake` property - if len(output) < 2: - return screen_on, False, -1, None, None, None, None - awake = output[1] == "1" - - lines = output.strip().splitlines() - - # `wake_lock_size` property - if len(lines[0]) < 3: - return screen_on, awake, -1, None, None, None, None - wake_lock_size = self._wake_lock_size(lines[0]) - - # `current_app` property - if len(lines) < 2: - return screen_on, awake, wake_lock_size, None, None, None, None - current_app = self._current_app(lines[1]) - - # `media_session_state` property - if len(lines) < 3: - return screen_on, awake, wake_lock_size, current_app, None, None, None - media_session_state = self._media_session_state(lines[2], current_app) - - # HDMI input property - if len(lines) < 4: - return screen_on, awake, wake_lock_size, current_app, media_session_state, None, None - hdmi_input = self._get_hdmi_input(lines[3]) - - # `running_apps` property - if not get_running_apps or len(lines) < 5: - return screen_on, awake, wake_lock_size, current_app, media_session_state, None, hdmi_input - running_apps = self._running_apps(lines[4:]) - - return screen_on, awake, wake_lock_size, current_app, media_session_state, running_apps, hdmi_input diff --git a/androidtv/firetv/firetv_async.py b/androidtv/firetv/firetv_async.py index 4d05961..2787ba7 100644 --- a/androidtv/firetv/firetv_async.py +++ b/androidtv/firetv/firetv_async.py @@ -100,13 +100,6 @@ async def update(self, get_running_apps=True, lazy=True): async def get_properties(self, get_running_apps=True, lazy=False): """Get the properties needed for Home Assistant updates. - This will send one of the following ADB commands: - - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_LAZY_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_NO_RUNNING_APPS` - Parameters ---------- get_running_apps : bool @@ -132,19 +125,21 @@ async def get_properties(self, get_running_apps=True, lazy=False): The HDMI input, or ``None`` if it could not be determined """ - if lazy: - if get_running_apps: - output = await self._adb.shell(self._cmd_get_properties_lazy_running_apps) - else: - output = await self._adb.shell(self._cmd_get_properties_lazy_no_running_apps) + screen_on, awake, wake_lock_size = await self.screen_on_awake_wake_lock_size() + if lazy and not (screen_on and awake): + return screen_on, awake, wake_lock_size, None, None, None, None + + # Left off here + current_app, media_session_state = await self.current_app_media_session_state() + + if get_running_apps: + running_apps = await self.running_apps() else: - if get_running_apps: - output = await self._adb.shell(self._cmd_get_properties_not_lazy_running_apps) - else: - output = await self._adb.shell(self._cmd_get_properties_not_lazy_no_running_apps) - _LOGGER.debug("Fire TV %s:%d `get_properties` response: %s", self.host, self.port, output) + running_apps = None + + hdmi_input = await self.get_hdmi_input() - return self._get_properties(output, get_running_apps) + return screen_on, awake, wake_lock_size, current_app, media_session_state, running_apps, hdmi_input async def get_properties_dict(self, get_running_apps=True, lazy=True): """Get the properties needed for Home Assistant updates and return them as a dictionary. diff --git a/androidtv/firetv/firetv_sync.py b/androidtv/firetv/firetv_sync.py index 6bcb272..66cb194 100644 --- a/androidtv/firetv/firetv_sync.py +++ b/androidtv/firetv/firetv_sync.py @@ -100,13 +100,6 @@ def update(self, get_running_apps=True, lazy=True): def get_properties(self, get_running_apps=True, lazy=False): """Get the properties needed for Home Assistant updates. - This will send one of the following ADB commands: - - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_LAZY_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_RUNNING_APPS` - * :py:const:`androidtv.constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_NO_RUNNING_APPS` - Parameters ---------- get_running_apps : bool @@ -132,19 +125,20 @@ def get_properties(self, get_running_apps=True, lazy=False): The HDMI input, or ``None`` if it could not be determined """ - if lazy: - if get_running_apps: - output = self._adb.shell(self._cmd_get_properties_lazy_running_apps) - else: - output = self._adb.shell(self._cmd_get_properties_lazy_no_running_apps) + screen_on, awake, wake_lock_size = self.screen_on_awake_wake_lock_size() + if lazy and not (screen_on and awake): + return screen_on, awake, wake_lock_size, None, None, None, None + + current_app, media_session_state = self.current_app_media_session_state() + + if get_running_apps: + running_apps = self.running_apps() else: - if get_running_apps: - output = self._adb.shell(self._cmd_get_properties_not_lazy_running_apps) - else: - output = self._adb.shell(self._cmd_get_properties_not_lazy_no_running_apps) - _LOGGER.debug("Fire TV %s:%d `get_properties` response: %s", self.host, self.port, output) + running_apps = None + + hdmi_input = self.get_hdmi_input() - return self._get_properties(output, get_running_apps) + return screen_on, awake, wake_lock_size, current_app, media_session_state, running_apps, hdmi_input def get_properties_dict(self, get_running_apps=True, lazy=True): """Get the properties needed for Home Assistant updates and return them as a dictionary. diff --git a/tests/test_firetv_async.py b/tests/test_firetv_async.py index e1e5154..e4f88be 100644 --- a/tests/test_firetv_async.py +++ b/tests/test_firetv_async.py @@ -24,9 +24,6 @@ async def setUp(self): ], async_patchers.patch_shell("")[self.PATCH_KEY]: self.ftv = FireTVAsync("HOST", 5555) await self.ftv.adb_connect() - self.assertEqual( - self.ftv._cmd_get_properties_lazy_no_running_apps, constants.CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS - ) @awaiter async def test_turn_on_off(self): diff --git a/tests/test_firetv_sync.py b/tests/test_firetv_sync.py index cc86c10..e43336e 100644 --- a/tests/test_firetv_sync.py +++ b/tests/test_firetv_sync.py @@ -190,9 +190,6 @@ def setUp(self): ]: self.ftv = FireTVSync("HOST", 5555) self.ftv.adb_connect() - self.assertEqual( - self.ftv._cmd_get_properties_lazy_no_running_apps, constants.CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS - ) def test_turn_on_off(self): """Test that the ``FireTVSync.turn_on`` and ``FireTVSync.turn_off`` methods work correctly.""" From 0d594ac1e14f4ee6b6ccda150e6e9f49bf494a4d Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 07:54:19 -0800 Subject: [PATCH 2/9] Remove test_homeassistant_sync.py --- tests/test_homeassistant_sync.py | 632 ------------------------------- 1 file changed, 632 deletions(-) delete mode 100644 tests/test_homeassistant_sync.py diff --git a/tests/test_homeassistant_sync.py b/tests/test_homeassistant_sync.py deleted file mode 100644 index 8c7187f..0000000 --- a/tests/test_homeassistant_sync.py +++ /dev/null @@ -1,632 +0,0 @@ -import functools -import logging -import sys -import unittest - -try: - from unittest.mock import patch -except ImportError: - from mock import patch - -sys.path.insert(0, "..") - -from adb_shell.exceptions import ( - InvalidChecksumError, - InvalidCommandError, - InvalidResponseError, - TcpTimeoutException, -) -from androidtv import setup -from androidtv.constants import APPS, KEYS, STATE_IDLE, STATE_OFF, STATE_PAUSED, STATE_PLAYING, STATE_STANDBY -from androidtv.exceptions import LockNotAcquiredException - -from . import patchers - - -_LOGGER = logging.getLogger(__name__) - -SUPPORT_ANDROIDTV = 12345 -SUPPORT_FIRETV = 12345 - -DIRECTION_PULL = "pull" -DIRECTION_PUSH = "push" - - -class MediaPlayerDevice(object): - _unique_id = None - - @staticmethod - def schedule_update_ha_state(): - pass - - -# =========================================================================== # -# # -# media_player.py # -# # -# =========================================================================== # - - -# Translate from `AndroidTV` / `FireTV` reported state to HA state. -ANDROIDTV_STATES = { - "off": STATE_OFF, - "idle": STATE_IDLE, - "standby": STATE_STANDBY, - "playing": STATE_PLAYING, - "paused": STATE_PAUSED, -} - - -def adb_decorator(override_available=False): - """Wrap ADB methods and catch exceptions. - Allows for overriding the available status of the ADB connection via the - `override_available` parameter. - """ - - def _adb_decorator(func): - """Wrap the provided ADB method and catch exceptions.""" - - @functools.wraps(func) - def _adb_exception_catcher(self, *args, **kwargs): - """Call an ADB-related method and catch exceptions.""" - if not self.available and not override_available: - return None - - try: - return func(self, *args, **kwargs) - except LockNotAcquiredException: - # If the ADB lock could not be acquired, skip this command - return - except self.exceptions as err: - _LOGGER.error( - "Failed to execute an ADB command. ADB connection re-" - "establishing attempt in the next update. Error: %s", - err, - ) - self.aftv.adb_close() - self._available = False # pylint: disable=protected-access - return None - - return _adb_exception_catcher - - return _adb_decorator - - -class ADBDevice(MediaPlayerDevice): - """Representation of an Android TV or Fire TV device.""" - - def __init__(self, aftv, name, apps, get_sources, turn_on_command, turn_off_command): - """Initialize the Android TV / Fire TV device.""" - self.aftv = aftv - self._name = name - self._app_id_to_name = APPS.copy() - self._app_id_to_name.update(apps) - self._app_name_to_id = {value: key for key, value in self._app_id_to_name.items()} - self._get_sources = get_sources - self._keys = KEYS - - self._device_properties = self.aftv.device_properties - self._unique_id = self._device_properties.get("serialno") - - self.turn_on_command = turn_on_command - self.turn_off_command = turn_off_command - - # ADB exceptions to catch - if not self.aftv.adb_server_ip: - # Using "adb_shell" (Python ADB implementation) - self.exceptions = ( - AttributeError, - BrokenPipeError, - TypeError, - ValueError, - InvalidChecksumError, - InvalidCommandError, - InvalidResponseError, - TcpTimeoutException, - ) - else: - # Using "pure-python-adb" (communicate with ADB server) - self.exceptions = (ConnectionResetError, RuntimeError) - - # Property attributes - self._adb_response = None - self._available = True - self._current_app = None - self._sources = None - self._state = None - - @property - def app_id(self): - """Return the current app.""" - return self._current_app - - @property - def app_name(self): - """Return the friendly name of the current app.""" - return self._app_id_to_name.get(self._current_app, self._current_app) - - @property - def available(self): - """Return whether or not the ADB connection is valid.""" - return self._available - - @property - def device_state_attributes(self): - """Provide the last ADB command's response as an attribute.""" - return {"adb_response": self._adb_response} - - @property - def name(self): - """Return the device name.""" - return self._name - - @property - def should_poll(self): - """Device should be polled.""" - return True - - @property - def source(self): - """Return the current app.""" - return self._app_id_to_name.get(self._current_app, self._current_app) - - @property - def source_list(self): - """Return a list of running apps.""" - return self._sources - - @property - def state(self): - """Return the state of the player.""" - return self._state - - @property - def unique_id(self): - """Return the device unique id.""" - return self._unique_id - - @adb_decorator() - def media_play(self): - """Send play command.""" - self.aftv.media_play() - - @adb_decorator() - def media_pause(self): - """Send pause command.""" - self.aftv.media_pause() - - @adb_decorator() - def media_play_pause(self): - """Send play/pause command.""" - self.aftv.media_play_pause() - - @adb_decorator() - def turn_on(self): - """Turn on the device.""" - if self.turn_on_command: - self.aftv.adb_shell(self.turn_on_command) - else: - self.aftv.turn_on() - - @adb_decorator() - def turn_off(self): - """Turn off the device.""" - if self.turn_off_command: - self.aftv.adb_shell(self.turn_off_command) - else: - self.aftv.turn_off() - - @adb_decorator() - def media_previous_track(self): - """Send previous track command (results in rewind).""" - self.aftv.media_previous_track() - - @adb_decorator() - def media_next_track(self): - """Send next track command (results in fast-forward).""" - self.aftv.media_next_track() - - @adb_decorator() - def select_source(self, source): - """Select input source. - If the source starts with a '!', then it will close the app instead of - opening it. - """ - if isinstance(source, str): - if not source.startswith("!"): - self.aftv.launch_app(self._app_name_to_id.get(source, source)) - else: - source_ = source[1:].lstrip() - self.aftv.stop_app(self._app_name_to_id.get(source_, source_)) - - @adb_decorator() - def adb_command(self, cmd): - """Send an ADB command to an Android TV / Fire TV device.""" - key = self._keys.get(cmd) - if key: - self.aftv.adb_shell("input keyevent {}".format(key)) - self._adb_response = None - self.schedule_update_ha_state() - return - - if cmd == "GET_PROPERTIES": - self._adb_response = str(self.aftv.get_properties_dict()) - self.schedule_update_ha_state() - return self._adb_response - - response = self.aftv.adb_shell(cmd) - if isinstance(response, str) and response.strip(): - self._adb_response = response.strip() - else: - self._adb_response = None - - self.schedule_update_ha_state() - return self._adb_response - - @adb_decorator() - def adb_filesync(self, direction, local_path, device_path): - """Transfer a file between your HA instance and an Android TV / Fire TV device.""" - if direction == DIRECTION_PULL: - self.aftv.adb_pull(local_path, device_path) - else: - self.aftv.adb_push(local_path, device_path) - - -class AndroidTVDevice(ADBDevice): - """Representation of an Android TV device.""" - - def __init__(self, aftv, name, apps, get_sources, turn_on_command, turn_off_command): - """Initialize the Android TV device.""" - super().__init__(aftv, name, apps, get_sources, turn_on_command, turn_off_command) - - self._is_volume_muted = None - self._volume_level = None - - @adb_decorator(override_available=True) - def update(self): - """Update the device state and, if necessary, re-connect.""" - # Check if device is disconnected. - if not self._available: - # Try to connect - self._available = 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 not self._available: - return - - # Get the updated state and attributes. - (state, self._current_app, running_apps, _, self._is_volume_muted, self._volume_level, _) = self.aftv.update( - self._get_sources - ) - - self._state = ANDROIDTV_STATES.get(state) - if self._state is None: - self._available = False - - if running_apps: - self._sources = [self._app_id_to_name.get(app_id, app_id) for app_id in running_apps] - else: - self._sources = None - - @property - def is_volume_muted(self): - """Boolean if volume is currently muted.""" - return self._is_volume_muted - - @property - def supported_features(self): - """Flag media player features that are supported.""" - return SUPPORT_ANDROIDTV - - @property - def volume_level(self): - """Return the volume level.""" - return self._volume_level - - @adb_decorator() - def media_stop(self): - """Send stop command.""" - self.aftv.media_stop() - - @adb_decorator() - def mute_volume(self, mute): - """Mute the volume.""" - self.aftv.mute_volume() - - @adb_decorator() - def volume_down(self): - """Send volume down command.""" - self._volume_level = self.aftv.volume_down(self._volume_level) - - @adb_decorator() - def volume_up(self): - """Send volume up command.""" - self._volume_level = self.aftv.volume_up(self._volume_level) - - -class FireTVDevice(ADBDevice): - """Representation of a Fire TV device.""" - - @adb_decorator(override_available=True) - def update(self): - """Update the device state and, if necessary, re-connect.""" - # Check if device is disconnected. - if not self._available: - # Try to connect - self._available = 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 not self._available: - return - - # Get the `state`, `current_app`, and `running_apps`. - state, self._current_app, running_apps, _ = self.aftv.update(self._get_sources) - - self._state = ANDROIDTV_STATES.get(state) - if self._state is None: - self._available = False - - if running_apps: - self._sources = [self._app_id_to_name.get(app_id, app_id) for app_id in running_apps] - else: - self._sources = None - - @property - def supported_features(self): - """Flag media player features that are supported.""" - return SUPPORT_FIRETV - - @adb_decorator() - def media_stop(self): - """Send stop (back) command.""" - self.aftv.back() - - -# =========================================================================== # -# # -# test_media_player.py # -# # -# =========================================================================== # - - -@unittest.skipIf(sys.version_info.major == 2, "Test requires Python 3") -class TestAndroidTVPythonImplementation(unittest.TestCase): - """Test the androidtv media player for an Android TV device.""" - - PATCH_KEY = "python" - - def setUp(self): - """Set up an `AndroidTVDevice` media player.""" - with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[ - self.PATCH_KEY - ]: - aftv = setup("HOST", 5555, device_class="androidtv") - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, True, None, None) - - def test_reconnect(self): - """Test that the error and reconnection attempts are logged correctly. - - "Handles device/service unavailable. Log a warning once when - unavailable, log once when reconnected." - - https://developers.home-assistant.io/docs/en/integration_quality_scale_index.html - """ - with self.assertLogs(level=logging.WARNING) as logs: - with patchers.patch_connect(False)[self.PATCH_KEY], patchers.patch_shell(error=True)[self.PATCH_KEY]: - for _ in range(5): - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) - - assert len(logs.output) == 2 - assert logs.output[0].startswith("ERROR") - assert logs.output[1].startswith("WARNING") - - with self.assertLogs(level=logging.DEBUG) as logs: - with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[self.PATCH_KEY]: - # Update 1 will reconnect - self.aftv.update() - self.assertTrue(self.aftv.available) - - # Update 2 will update the state - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) - - assert ( - "ADB connection to {}:{} successfully established".format(self.aftv.aftv.host, self.aftv.aftv.port) - in logs.output[0] - ) - - def test_adb_shell_returns_none(self): - """Test the case that the ADB shell command returns `None`. - - The state should be `None` and the device should be unavailable. - """ - with patchers.patch_shell(None)[self.PATCH_KEY]: - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) - - with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[self.PATCH_KEY]: - # Update 1 will reconnect - self.aftv.update() - self.assertTrue(self.aftv.available) - - # Update 2 will update the state - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) - - -@unittest.skipIf(sys.version_info.major == 2, "Test requires Python 3") -class TestAndroidTVServerImplementation(unittest.TestCase): - """Test the androidtv media player for an Android TV device.""" - - PATCH_KEY = "server" - - def setUp(self): - """Set up an `AndroidTVDevice` media player.""" - with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[self.PATCH_KEY]: - aftv = setup("HOST", 5555, adb_server_ip="ADB_SERVER_IP", device_class="androidtv") - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, True, None, None) - - def test_reconnect(self): - """Test that the error and reconnection attempts are logged correctly. - - "Handles device/service unavailable. Log a warning once when - unavailable, log once when reconnected." - - https://developers.home-assistant.io/docs/en/integration_quality_scale_index.html - """ - with self.assertLogs(level=logging.WARNING) as logs: - with patchers.patch_connect(False)[self.PATCH_KEY], patchers.patch_shell(error=True)[self.PATCH_KEY]: - for _ in range(5): - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) - - assert len(logs.output) == 2 - assert logs.output[0].startswith("ERROR") - assert logs.output[1].startswith("WARNING") - - with self.assertLogs(level=logging.DEBUG) as logs: - with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[self.PATCH_KEY]: - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) - - assert ( - "ADB connection to {}:{} via ADB server {}:{} successfully established".format( - self.aftv.aftv.host, - self.aftv.aftv.port, - self.aftv.aftv.adb_server_ip, - self.aftv.aftv.adb_server_port, - ) - in logs.output[0] - ) - - def test_adb_shell_returns_none(self): - """Test the case that the ADB shell command returns `None`. - - The state should be `None` and the device should be unavailable. - """ - with patchers.patch_shell(None)[self.PATCH_KEY]: - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) - - with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[self.PATCH_KEY]: - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) - - -@unittest.skipIf(sys.version_info.major == 2, "Test requires Python 3") -class TestFireTVPythonImplementation(TestAndroidTVPythonImplementation): - """Test the androidtv media player for a Fire TV device.""" - - def setUp(self): - """Set up a `FireTVDevice` media player.""" - with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[ - self.PATCH_KEY - ]: - aftv = setup("HOST", 5555, device_class="firetv") - self.aftv = FireTVDevice(aftv, "Fake Fire TV", {}, True, None, None) - - -@unittest.skipIf(sys.version_info.major == 2, "Test requires Python 3") -class TestFireTVServerImplementation(TestAndroidTVServerImplementation): - """Test the androidtv media player for a Fire TV device.""" - - def setUp(self): - """Set up a `FireTVDevice` media player.""" - with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("")[self.PATCH_KEY]: - aftv = setup("HOST", 5555, adb_server_ip="ADB_SERVER_IP", device_class="firetv") - self.aftv = FireTVDevice(aftv, "Fake Fire TV", {}, True, None, None) - - -@unittest.skipIf(sys.version_info.major == 2, "Test requires Python 3") -class TestADBCommandAndFileSync(unittest.TestCase): - """Test ADB and FileSync services.""" - - def test_adb_command(self): - """Test sending a command via the `androidtv.adb_command` service.""" - patch_key = "server" - command = "test command" - response = "test response" - - with patchers.patch_connect(True)[patch_key], patchers.patch_shell("")[patch_key]: - aftv = setup("HOST", 5555, adb_server_ip="ADB_SERVER_IP", device_class="androidtv") - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, True, None, None) - - with patch("androidtv.basetv.basetv_sync.BaseTVSync.adb_shell", return_value=response) as patch_shell: - self.aftv.adb_command(command) - - patch_shell.assert_called_with(command) - assert self.aftv._adb_response == response - - def test_adb_command_key(self): - """Test sending a key command via the `androidtv.adb_command` service.""" - patch_key = "server" - command = "HOME" - response = None - - with patchers.patch_connect(True)[patch_key], patchers.patch_shell("")[patch_key]: - aftv = setup("HOST", 5555, adb_server_ip="ADB_SERVER_IP", device_class="androidtv") - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, True, None, None) - - with patch("androidtv.basetv.basetv_sync.BaseTVSync.adb_shell", return_value=response) as patch_shell: - self.aftv.adb_command(command) - - patch_shell.assert_called_with("input keyevent {}".format(self.aftv._keys[command])) - assert self.aftv._adb_response is None - - def test_adb_command_get_properties(self): - """Test sending the "GET_PROPERTIES" command via the `androidtv.adb_command` service.""" - patch_key = "server" - command = "GET_PROPERTIES" - response = {"key": "value"} - - with patchers.patch_connect(True)[patch_key], patchers.patch_shell("")[patch_key]: - aftv = setup("HOST", 5555, adb_server_ip="ADB_SERVER_IP", device_class="androidtv") - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, True, None, None) - - with patch( - "androidtv.androidtv.androidtv_sync.AndroidTVSync.get_properties_dict", return_value=response - ) as patch_get_props: - self.aftv.adb_command(command) - - assert patch_get_props.called - assert self.aftv._adb_response == str(response) - - def test_update_lock_not_acquired(self): - """Test that the state does not get updated when a `LockNotAcquiredException` is raised.""" - patch_key = "server" - - with patchers.patch_connect(True)[patch_key], patchers.patch_shell("")[patch_key]: - aftv = setup("HOST", 5555, adb_server_ip="ADB_SERVER_IP", device_class="androidtv") - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, True, None, None) - - with patchers.patch_shell("")[patch_key]: - self.aftv.update() - assert self.aftv.state == STATE_OFF - - with patch("androidtv.androidtv.androidtv_sync.AndroidTVSync.update", side_effect=LockNotAcquiredException): - with patchers.patch_shell("1")[patch_key]: - self.aftv.update() - assert self.aftv.state == STATE_OFF - - with patchers.patch_shell("1")[patch_key]: - self.aftv.update() - assert self.aftv.state == STATE_STANDBY From a737800da9ab16a83e9e76ff049f4568c38c5f66 Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 07:57:57 -0800 Subject: [PATCH 3/9] Fix test_update --- tests/test_firetv_async.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_firetv_async.py b/tests/test_firetv_async.py index e4f88be..64ec603 100644 --- a/tests/test_firetv_async.py +++ b/tests/test_firetv_async.py @@ -93,7 +93,7 @@ async def test_get_properties(self): async def test_update(self): """Check that the ``update`` method works correctly.""" with async_patchers.patch_shell(None)[self.PATCH_KEY]: - with patch_calls(self.ftv, self.ftv._get_properties) as patched: + with patch_calls(self.ftv, self.ftv.get_properties) as patched: await self.ftv.update() assert patched.called From d413542ded2bcacc383db6f7bdde3a1acdab0972 Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 12:41:06 -0800 Subject: [PATCH 4/9] Fix async get_properties test --- tests/test_firetv_async.py | 49 ++++++++++++++++++++++++++++++++++---- 1 file changed, 44 insertions(+), 5 deletions(-) diff --git a/tests/test_firetv_async.py b/tests/test_firetv_async.py index 64ec603..cfd201f 100644 --- a/tests/test_firetv_async.py +++ b/tests/test_firetv_async.py @@ -83,11 +83,50 @@ async def test_running_apps(self): async def test_get_properties(self): """Check that ``get_properties()`` works correctly.""" with async_patchers.patch_shell(None)[self.PATCH_KEY]: - for get_running_apps in [True, False]: - for lazy in [True, False]: - with patch_calls(self.ftv, self.ftv._get_properties) as patched: - await self.ftv.get_properties_dict(get_running_apps, lazy) - assert patched.called + with patch_calls( + self.ftv, self.ftv.screen_on_awake_wake_lock_size + ) as screen_on_awake_wake_lock_size, patch_calls( + self.ftv, self.ftv.current_app_media_session_state + ) as current_app_media_session_state, patch_calls( + self.ftv, self.ftv.running_apps + ) as running_apps, patch_calls( + self.ftv, self.ftv.get_hdmi_input + ) as get_hdmi_input: + await self.ftv.get_properties(lazy=True) + assert screen_on_awake_wake_lock_size.called + assert not current_app_media_session_state.called + assert not running_apps.called + assert not get_hdmi_input.called + + with patch_calls( + self.ftv, self.ftv.screen_on_awake_wake_lock_size + ) as screen_on_awake_wake_lock_size, patch_calls( + self.ftv, self.ftv.current_app_media_session_state + ) as current_app_media_session_state, patch_calls( + self.ftv, self.ftv.running_apps + ) as running_apps, patch_calls( + self.ftv, self.ftv.get_hdmi_input + ) as get_hdmi_input: + await self.ftv.get_properties(lazy=False, get_running_apps=True) + assert screen_on_awake_wake_lock_size.called + assert current_app_media_session_state.called + assert running_apps.called + assert get_hdmi_input.called + + with patch_calls( + self.ftv, self.ftv.screen_on_awake_wake_lock_size + ) as screen_on_awake_wake_lock_size, patch_calls( + self.ftv, self.ftv.current_app_media_session_state + ) as current_app_media_session_state, patch_calls( + self.ftv, self.ftv.running_apps + ) as running_apps, patch_calls( + self.ftv, self.ftv.get_hdmi_input + ) as get_hdmi_input: + await self.ftv.get_properties(lazy=False, get_running_apps=False) + assert screen_on_awake_wake_lock_size.called + assert current_app_media_session_state.called + assert not running_apps.called + assert get_hdmi_input.called @awaiter async def test_update(self): From 71e4e981a86413f2e03939f5bf2e5d8f6d26be62 Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 13:00:18 -0800 Subject: [PATCH 5/9] Fix tests --- androidtv/basetv/basetv.py | 13 ++-- tests/test_basetv_async.py | 2 +- tests/test_basetv_sync.py | 2 +- tests/test_firetv_async.py | 8 +++ tests/test_firetv_sync.py | 119 +++++++++++++++++-------------------- 5 files changed, 72 insertions(+), 72 deletions(-) diff --git a/androidtv/basetv/basetv.py b/androidtv/basetv/basetv.py index f95c9d8..45f47f9 100644 --- a/androidtv/basetv/basetv.py +++ b/androidtv/basetv/basetv.py @@ -543,15 +543,18 @@ def _screen_on_awake_wake_lock_size(output): Returns ------- - bool - Whether or not the device is on - bool - Whether or not the device is awake (screensaver is not running) + bool, None + Whether or not the device is on, or ``None`` if it could not be determined + bool, None + Whether or not the device is awake (screensaver is not running), or ``None`` if it could not be determined int, None The size of the current wake lock, or ``None`` if it could not be determined """ - if not output: + if output is None: + return None, None, None + + if output == "": return False, False, None screen_on = output[0] == "1" diff --git a/tests/test_basetv_async.py b/tests/test_basetv_async.py index a2a185c..36b84ed 100644 --- a/tests/test_basetv_async.py +++ b/tests/test_basetv_async.py @@ -408,7 +408,7 @@ async def test_screen_on(self): async def test_screen_on_awake_wake_lock_size(self): """Check that the ``screen_on_awake_wake_lock_size`` property works correctly.""" with async_patchers.patch_shell(None)[self.PATCH_KEY]: - self.assertTupleEqual(await self.btv.screen_on_awake_wake_lock_size(), (False, False, None)) + self.assertTupleEqual(await self.btv.screen_on_awake_wake_lock_size(), (None, None, None)) with async_patchers.patch_shell("")[self.PATCH_KEY]: self.assertTupleEqual(await self.btv.screen_on_awake_wake_lock_size(), (False, False, None)) diff --git a/tests/test_basetv_sync.py b/tests/test_basetv_sync.py index c782558..b6160ce 100644 --- a/tests/test_basetv_sync.py +++ b/tests/test_basetv_sync.py @@ -565,7 +565,7 @@ def test_screen_on(self): def test_screen_on_awake_wake_lock_size(self): """Check that the ``screen_on_awake_wake_lock_size`` property works correctly.""" with patchers.patch_shell(None)[self.PATCH_KEY]: - self.assertTupleEqual(self.btv.screen_on_awake_wake_lock_size(), (False, False, None)) + self.assertTupleEqual(self.btv.screen_on_awake_wake_lock_size(), (None, None, None)) with patchers.patch_shell("")[self.PATCH_KEY]: self.assertTupleEqual(self.btv.screen_on_awake_wake_lock_size(), (False, False, None)) diff --git a/tests/test_firetv_async.py b/tests/test_firetv_async.py index cfd201f..90fc000 100644 --- a/tests/test_firetv_async.py +++ b/tests/test_firetv_async.py @@ -128,6 +128,14 @@ async def test_get_properties(self): assert not running_apps.called assert get_hdmi_input.called + @awaiter + async def test_get_properties_dict(self): + """Check that ``get_properties_dict()`` works correctly.""" + with async_patchers.patch_shell(None)[self.PATCH_KEY]: + with patch_calls(self.ftv, self.ftv.get_properties) as get_properties: + await self.ftv.get_properties_dict() + assert get_properties.called + @awaiter async def test_update(self): """Check that the ``update`` method works correctly.""" diff --git a/tests/test_firetv_sync.py b/tests/test_firetv_sync.py index e43336e..57592cb 100644 --- a/tests/test_firetv_sync.py +++ b/tests/test_firetv_sync.py @@ -251,76 +251,65 @@ def test_running_apps(self): def test_get_properties(self): """Check that ``get_properties()`` works correctly.""" with patchers.patch_shell(None)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT_NONE) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT1)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT1) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT2)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT2) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3A)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3A) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3B)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3B) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3C)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3C) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3D)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3D) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3E)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3E) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3E)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True, get_running_apps=False) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3E) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3E)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=False, get_running_apps=False) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3E) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3F)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT3F) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT4)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT4) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT4)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True, get_running_apps=False) - self.assertDictEqual(properties, GET_PROPERTIES_DICT4) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT5)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=True) - self.assertDictEqual(properties, GET_PROPERTIES_DICT5) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT5)[self.PATCH_KEY]: - properties = self.ftv.get_properties_dict(lazy=False) - self.assertDictEqual(properties, GET_PROPERTIES_DICT5) + with patchers.patch_calls( + self.ftv, self.ftv.screen_on_awake_wake_lock_size + ) as screen_on_awake_wake_lock_size, patchers.patch_calls( + self.ftv, self.ftv.current_app_media_session_state + ) as current_app_media_session_state, patchers.patch_calls( + self.ftv, self.ftv.running_apps + ) as running_apps, patchers.patch_calls( + self.ftv, self.ftv.get_hdmi_input + ) as get_hdmi_input: + self.ftv.get_properties(lazy=True) + assert screen_on_awake_wake_lock_size.called + assert not current_app_media_session_state.called + assert not running_apps.called + assert not get_hdmi_input.called + + with patchers.patch_calls( + self.ftv, self.ftv.screen_on_awake_wake_lock_size + ) as screen_on_awake_wake_lock_size, patchers.patch_calls( + self.ftv, self.ftv.current_app_media_session_state + ) as current_app_media_session_state, patchers.patch_calls( + self.ftv, self.ftv.running_apps + ) as running_apps, patchers.patch_calls( + self.ftv, self.ftv.get_hdmi_input + ) as get_hdmi_input: + self.ftv.get_properties(lazy=False, get_running_apps=True) + assert screen_on_awake_wake_lock_size.called + assert current_app_media_session_state.called + assert running_apps.called + assert get_hdmi_input.called + + with patchers.patch_calls( + self.ftv, self.ftv.screen_on_awake_wake_lock_size + ) as screen_on_awake_wake_lock_size, patchers.patch_calls( + self.ftv, self.ftv.current_app_media_session_state + ) as current_app_media_session_state, patchers.patch_calls( + self.ftv, self.ftv.running_apps + ) as running_apps, patchers.patch_calls( + self.ftv, self.ftv.get_hdmi_input + ) as get_hdmi_input: + self.ftv.get_properties(lazy=False, get_running_apps=False) + assert screen_on_awake_wake_lock_size.called + assert current_app_media_session_state.called + assert not running_apps.called + assert get_hdmi_input.called + + def test_get_properties_dict(self): + """Check that ``get_properties_dict()`` works correctly.""" + with patchers.patch_shell(None)[self.PATCH_KEY]: + with patchers.patch_calls(self.ftv, self.ftv.get_properties) as get_properties: + self.ftv.get_properties_dict() + assert get_properties.called def test_update(self): """Check that the ``update`` method works correctly.""" with patchers.patch_connect(False)[self.PATCH_KEY]: self.ftv.adb_connect() - state = self.ftv.update() - self.assertTupleEqual(state, STATE_NONE) - + # state = self.ftv.update() + self.assertTupleEqual(self.ftv.update(), STATE_NONE) + return with patchers.patch_connect(True)[self.PATCH_KEY]: self.assertTrue(self.ftv.adb_connect()) From 3bef6eb89480b60cee7a73781a4cc371b51b1b01 Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 13:26:40 -0800 Subject: [PATCH 6/9] Cleanup --- androidtv/firetv/firetv_async.py | 1 - tests/test_firetv_sync.py | 41 +------------------------------- 2 files changed, 1 insertion(+), 41 deletions(-) diff --git a/androidtv/firetv/firetv_async.py b/androidtv/firetv/firetv_async.py index 2787ba7..202eeeb 100644 --- a/androidtv/firetv/firetv_async.py +++ b/androidtv/firetv/firetv_async.py @@ -129,7 +129,6 @@ async def get_properties(self, get_running_apps=True, lazy=False): if lazy and not (screen_on and awake): return screen_on, awake, wake_lock_size, None, None, None, None - # Left off here current_app, media_session_state = await self.current_app_media_session_state() if get_running_apps: diff --git a/tests/test_firetv_sync.py b/tests/test_firetv_sync.py index 57592cb..c1b8be8 100644 --- a/tests/test_firetv_sync.py +++ b/tests/test_firetv_sync.py @@ -307,47 +307,8 @@ def test_update(self): """Check that the ``update`` method works correctly.""" with patchers.patch_connect(False)[self.PATCH_KEY]: self.ftv.adb_connect() - # state = self.ftv.update() - self.assertTupleEqual(self.ftv.update(), STATE_NONE) - return - with patchers.patch_connect(True)[self.PATCH_KEY]: - self.assertTrue(self.ftv.adb_connect()) - - with patchers.patch_shell(None)[self.PATCH_KEY]: - state = self.ftv.update() - self.assertTupleEqual(state, STATE_NONE) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT1)[self.PATCH_KEY]: - state = self.ftv.update() - self.assertTupleEqual(state, STATE1) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT2)[self.PATCH_KEY]: - state = self.ftv.update() - self.assertTupleEqual(state, STATE2) - - with patchers.patch_shell(GET_PROPERTIES_OUTPUT3)[self.PATCH_KEY]: - state = self.ftv.update() - self.assertTupleEqual(state, STATE3) - self.ftv._state_detection_rules = STATE_DETECTION_RULES1 - state = self.ftv.update() - self.assertEqual(state[0], constants.STATE_OFF) - - self.ftv._state_detection_rules = STATE_DETECTION_RULES2 - state = self.ftv.update() - self.assertEqual(state[0], constants.STATE_OFF) - - self.ftv._state_detection_rules = STATE_DETECTION_RULES3 - state = self.ftv.update() - self.assertEqual(state[0], constants.STATE_STANDBY) - - self.ftv._state_detection_rules = STATE_DETECTION_RULES4 - state = self.ftv.update() - self.assertEqual(state[0], constants.STATE_PAUSED) - - self.ftv._state_detection_rules = STATE_DETECTION_RULES5 - state = self.ftv.update() - self.assertEqual(state[0], constants.STATE_IDLE) + self.assertTupleEqual(self.ftv.update(), STATE_NONE) def assertUpdate(self, get_properties, update): """Check that the results of the `update` method are as expected.""" From 476bfe50a803b9f2b695dd8cedaaa9901030252d Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 13:29:44 -0800 Subject: [PATCH 7/9] Cleanup --- androidtv/constants.py | 74 ----------------------------------------- tests/test_constants.py | 24 ------------- 2 files changed, 98 deletions(-) diff --git a/androidtv/constants.py b/androidtv/constants.py index 42731ee..942e4e0 100644 --- a/androidtv/constants.py +++ b/androidtv/constants.py @@ -274,80 +274,6 @@ + CMD_STREAM_MUSIC ) -#: Get the properties for a Fire TV device (``lazy=True, get_running_apps=True``); see :py:meth:`androidtv.firetv.firetv_sync.FireTVSync.get_properties` and :py:meth:`androidtv.firetv.firetv_async.FireTVAsync.get_properties` -CMD_FIRETV_PROPERTIES_LAZY_RUNNING_APPS = ( - CMD_SCREEN_ON - + CMD_SUCCESS1 - + " && " - + CMD_AWAKE - + CMD_SUCCESS1 - + " && " - + CMD_WAKE_LOCK_SIZE - + " && " - + CMD_CURRENT_APP - + " && (" - + CMD_MEDIA_SESSION_STATE - + " || echo) && (" - + CMD_HDMI_INPUT - + " || echo) && " - + CMD_RUNNING_APPS_FIRETV -) - -#: Get the properties for a Fire TV device (``lazy=True, get_running_apps=False``); see :py:meth:`androidtv.firetv.firetv_sync.FireTVSync.get_properties` and :py:meth:`androidtv.firetv.firetv_async.FireTVAsync.get_properties` -CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS = ( - CMD_SCREEN_ON - + CMD_SUCCESS1 - + " && " - + CMD_AWAKE - + CMD_SUCCESS1 - + " && " - + CMD_WAKE_LOCK_SIZE - + " && " - + CMD_CURRENT_APP - + " && (" - + CMD_MEDIA_SESSION_STATE - + " || echo) && (" - + CMD_HDMI_INPUT - + " || echo)" -) - -#: Get the properties for a Fire TV device (``lazy=False, get_running_apps=True``); see :py:meth:`androidtv.firetv.firetv_sync.FireTVSync.get_properties` and :py:meth:`androidtv.firetv.firetv_async.FireTVAsync.get_properties` -CMD_FIRETV_PROPERTIES_NOT_LAZY_RUNNING_APPS = ( - CMD_SCREEN_ON - + CMD_SUCCESS1_FAILURE0 - + " && " - + CMD_AWAKE - + CMD_SUCCESS1_FAILURE0 - + " && " - + CMD_WAKE_LOCK_SIZE - + " && " - + CMD_CURRENT_APP - + " && (" - + CMD_MEDIA_SESSION_STATE - + " || echo) && (" - + CMD_HDMI_INPUT - + " || echo) && " - + CMD_RUNNING_APPS_FIRETV -) - -#: Get the properties for a Fire TV device (``lazy=False, get_running_apps=False``); see :py:meth:`androidtv.firetv.firetv_sync.FireTVSync.get_properties` and :py:meth:`androidtv.firetv.firetv_async.FireTVAsync.get_properties` -CMD_FIRETV_PROPERTIES_NOT_LAZY_NO_RUNNING_APPS = ( - CMD_SCREEN_ON - + CMD_SUCCESS1_FAILURE0 - + " && " - + CMD_AWAKE - + CMD_SUCCESS1_FAILURE0 - + " && " - + CMD_WAKE_LOCK_SIZE - + " && " - + CMD_CURRENT_APP - + " && (" - + CMD_MEDIA_SESSION_STATE - + " || echo) && (" - + CMD_HDMI_INPUT - + " || echo)" -) - # `getprop` commands CMD_MANUFACTURER = "getprop ro.product.manufacturer" CMD_MODEL = "getprop ro.product.model" diff --git a/tests/test_constants.py b/tests/test_constants.py index 04fba4f..afe9dcc 100644 --- a/tests/test_constants.py +++ b/tests/test_constants.py @@ -119,30 +119,6 @@ def test_constants(self): r"(dumpsys power | grep 'Display Power' | grep -q 'state=ON' || dumpsys power | grep -q 'mScreenOn=true') && echo -e '1\c' || echo -e '0\c' && dumpsys power | grep mWakefulness | grep -q Awake && echo -e '1\c' || echo -e '0\c' && (dumpsys audio | grep paused | grep -qv 'Buffer Queue' && echo -e '1\c' || (dumpsys audio | grep started | grep -qv 'Buffer Queue' && echo '2\c' || echo '0\c')) && dumpsys power | grep Locks | grep 'size=' && CURRENT_APP=$(dumpsys activity a . | grep mResumedActivity) && CURRENT_APP=${CURRENT_APP#*ActivityRecord{* * } && CURRENT_APP=${CURRENT_APP#*{* * } && CURRENT_APP=${CURRENT_APP%%/*} && CURRENT_APP=${CURRENT_APP%\}*} && echo $CURRENT_APP && (dumpsys media_session | grep -A 100 'Sessions Stack' | grep -A 100 $CURRENT_APP | grep -m 1 'state=PlaybackState {' || echo) && (dumpsys activity starter | grep -E -o '(ExternalTv|HDMI)InputService/HW[0-9]' -m 1 | grep -o 'HW[0-9]' || echo) && dumpsys audio | grep '\- STREAM_MUSIC:' -A 11", ) - # CMD_FIRETV_PROPERTIES_LAZY_RUNNING_APPS - self.assertEqual( - constants.CMD_FIRETV_PROPERTIES_LAZY_RUNNING_APPS, - r"(dumpsys power | grep 'Display Power' | grep -q 'state=ON' || dumpsys power | grep -q 'mScreenOn=true') && echo -e '1\c' && dumpsys power | grep mWakefulness | grep -q Awake && echo -e '1\c' && dumpsys power | grep Locks | grep 'size=' && CURRENT_APP=$(dumpsys window windows | grep -E 'mCurrentFocus|mFocusedApp') && CURRENT_APP=${CURRENT_APP#*ActivityRecord{* * } && CURRENT_APP=${CURRENT_APP#*{* * } && CURRENT_APP=${CURRENT_APP%%/*} && CURRENT_APP=${CURRENT_APP%\}*} && echo $CURRENT_APP && (dumpsys media_session | grep -A 100 'Sessions Stack' | grep -A 100 $CURRENT_APP | grep -m 1 'state=PlaybackState {' || echo) && (dumpsys activity starter | grep -E -o '(ExternalTv|HDMI)InputService/HW[0-9]' -m 1 | grep -o 'HW[0-9]' || echo) && ps | grep u0_a", - ) - - # CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS - self.assertEqual( - constants.CMD_FIRETV_PROPERTIES_LAZY_NO_RUNNING_APPS, - r"(dumpsys power | grep 'Display Power' | grep -q 'state=ON' || dumpsys power | grep -q 'mScreenOn=true') && echo -e '1\c' && dumpsys power | grep mWakefulness | grep -q Awake && echo -e '1\c' && dumpsys power | grep Locks | grep 'size=' && CURRENT_APP=$(dumpsys window windows | grep -E 'mCurrentFocus|mFocusedApp') && CURRENT_APP=${CURRENT_APP#*ActivityRecord{* * } && CURRENT_APP=${CURRENT_APP#*{* * } && CURRENT_APP=${CURRENT_APP%%/*} && CURRENT_APP=${CURRENT_APP%\}*} && echo $CURRENT_APP && (dumpsys media_session | grep -A 100 'Sessions Stack' | grep -A 100 $CURRENT_APP | grep -m 1 'state=PlaybackState {' || echo) && (dumpsys activity starter | grep -E -o '(ExternalTv|HDMI)InputService/HW[0-9]' -m 1 | grep -o 'HW[0-9]' || echo)", - ) - - # CMD_FIRETV_PROPERTIES_NOT_LAZY_RUNNING_APPS - self.assertEqual( - constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_RUNNING_APPS, - r"(dumpsys power | grep 'Display Power' | grep -q 'state=ON' || dumpsys power | grep -q 'mScreenOn=true') && echo -e '1\c' || echo -e '0\c' && dumpsys power | grep mWakefulness | grep -q Awake && echo -e '1\c' || echo -e '0\c' && dumpsys power | grep Locks | grep 'size=' && CURRENT_APP=$(dumpsys window windows | grep -E 'mCurrentFocus|mFocusedApp') && CURRENT_APP=${CURRENT_APP#*ActivityRecord{* * } && CURRENT_APP=${CURRENT_APP#*{* * } && CURRENT_APP=${CURRENT_APP%%/*} && CURRENT_APP=${CURRENT_APP%\}*} && echo $CURRENT_APP && (dumpsys media_session | grep -A 100 'Sessions Stack' | grep -A 100 $CURRENT_APP | grep -m 1 'state=PlaybackState {' || echo) && (dumpsys activity starter | grep -E -o '(ExternalTv|HDMI)InputService/HW[0-9]' -m 1 | grep -o 'HW[0-9]' || echo) && ps | grep u0_a", - ) - - # CMD_FIRETV_PROPERTIES_NOT_LAZY_NO_RUNNING_APPS - self.assertEqual( - constants.CMD_FIRETV_PROPERTIES_NOT_LAZY_NO_RUNNING_APPS, - r"(dumpsys power | grep 'Display Power' | grep -q 'state=ON' || dumpsys power | grep -q 'mScreenOn=true') && echo -e '1\c' || echo -e '0\c' && dumpsys power | grep mWakefulness | grep -q Awake && echo -e '1\c' || echo -e '0\c' && dumpsys power | grep Locks | grep 'size=' && CURRENT_APP=$(dumpsys window windows | grep -E 'mCurrentFocus|mFocusedApp') && CURRENT_APP=${CURRENT_APP#*ActivityRecord{* * } && CURRENT_APP=${CURRENT_APP#*{* * } && CURRENT_APP=${CURRENT_APP%%/*} && CURRENT_APP=${CURRENT_APP%\}*} && echo $CURRENT_APP && (dumpsys media_session | grep -A 100 'Sessions Stack' | grep -A 100 $CURRENT_APP | grep -m 1 'state=PlaybackState {' || echo) && (dumpsys activity starter | grep -E -o '(ExternalTv|HDMI)InputService/HW[0-9]' -m 1 | grep -o 'HW[0-9]' || echo)", - ) - def test_current_app_extraction_atv_launcher(self): dumpsys_output = """ mCurrentFocus=Window{e74bb23 u0 com.google.android.tvlauncher/com.google.android.tvlauncher.MainActivity} From 23f0bb6986c506299cf2c00663fd0dd8b1fe3138 Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 13:32:39 -0800 Subject: [PATCH 8/9] Cleanup --- tests/test_firetv_sync.py | 135 -------------------------------------- 1 file changed, 135 deletions(-) diff --git a/tests/test_firetv_sync.py b/tests/test_firetv_sync.py index c1b8be8..6f26a21 100644 --- a/tests/test_firetv_sync.py +++ b/tests/test_firetv_sync.py @@ -25,45 +25,10 @@ RUNNING_APPS_LIST = ["com.netflix.ninja", "com.amazon.device.controllermanager"] -GET_PROPERTIES_OUTPUT1 = "" -GET_PROPERTIES_DICT1 = { - "screen_on": False, - "awake": False, - "wake_lock_size": -1, - "current_app": None, - "media_session_state": None, - "running_apps": None, - "hdmi_input": None, -} STATE1 = (constants.STATE_OFF, None, None, None) -GET_PROPERTIES_OUTPUT2 = "1" -GET_PROPERTIES_DICT2 = { - "screen_on": True, - "awake": False, - "wake_lock_size": -1, - "current_app": None, - "media_session_state": None, - "running_apps": None, - "hdmi_input": None, -} STATE2 = (constants.STATE_STANDBY, None, None, None) -GET_PROPERTIES_OUTPUT3 = """11Wake Locks: size=2 -com.amazon.tv.launcher - - -u0_a2 17243 197 998628 24932 ffffffff 00000000 S com.amazon.device.controllermanager -u0_a2 17374 197 995368 20764 ffffffff 00000000 S com.amazon.device.controllermanager:BluetoothReceiver""" -GET_PROPERTIES_DICT3 = { - "screen_on": True, - "awake": True, - "wake_lock_size": 2, - "current_app": "com.amazon.tv.launcher", - "media_session_state": None, - "running_apps": ["com.amazon.device.controllermanager", "com.amazon.device.controllermanager:BluetoothReceiver"], - "hdmi_input": None, -} STATE3 = ( constants.STATE_IDLE, "com.amazon.tv.launcher", @@ -71,106 +36,6 @@ None, ) -GET_PROPERTIES_OUTPUT3A = GET_PROPERTIES_OUTPUT3[0] -GET_PROPERTIES_OUTPUT3B = GET_PROPERTIES_OUTPUT3[:2] -GET_PROPERTIES_OUTPUT3C = GET_PROPERTIES_OUTPUT3.splitlines()[0] -GET_PROPERTIES_OUTPUT3D = "\n".join(GET_PROPERTIES_OUTPUT3.splitlines()[:2]) -GET_PROPERTIES_OUTPUT3E = "\n".join(GET_PROPERTIES_OUTPUT3.splitlines()[:3]) -GET_PROPERTIES_OUTPUT3F = "\n".join(GET_PROPERTIES_OUTPUT3.splitlines()[:4]) + "HW2" - -GET_PROPERTIES_DICT3A = { - "screen_on": True, - "awake": False, - "wake_lock_size": -1, - "current_app": None, - "media_session_state": None, - "running_apps": None, - "hdmi_input": None, -} -GET_PROPERTIES_DICT3B = { - "screen_on": True, - "awake": True, - "wake_lock_size": -1, - "current_app": None, - "media_session_state": None, - "running_apps": None, - "hdmi_input": None, -} -GET_PROPERTIES_DICT3C = { - "screen_on": True, - "awake": True, - "wake_lock_size": 2, - "current_app": None, - "media_session_state": None, - "running_apps": None, - "hdmi_input": None, -} -GET_PROPERTIES_DICT3D = { - "screen_on": True, - "awake": True, - "wake_lock_size": 2, - "current_app": "com.amazon.tv.launcher", - "media_session_state": None, - "running_apps": None, - "hdmi_input": None, -} -GET_PROPERTIES_DICT3E = { - "screen_on": True, - "awake": True, - "wake_lock_size": 2, - "current_app": "com.amazon.tv.launcher", - "media_session_state": None, - "running_apps": None, - "hdmi_input": None, -} -GET_PROPERTIES_DICT3F = { - "screen_on": True, - "awake": True, - "wake_lock_size": 2, - "current_app": "com.amazon.tv.launcher", - "media_session_state": None, - "running_apps": None, - "hdmi_input": "HW2", -} - -GET_PROPERTIES_OUTPUT4 = """11Wake Locks: size=2 -com.amazon.tv.launcher -state=PlaybackState {state=2, position=0, buffered position=0, speed=0.0, updated=65749, actions=240640, custom actions=[], active item id=-1, error=null}""" -GET_PROPERTIES_DICT4 = { - "screen_on": True, - "awake": True, - "wake_lock_size": 2, - "current_app": "com.amazon.tv.launcher", - "media_session_state": 2, - "running_apps": None, - "hdmi_input": None, -} - -GET_PROPERTIES_OUTPUT5 = """11Wake Locks: size=2 -com.amazon.tv.launcher -state=PlaybackState {state=2, position=0, buffered position=0, speed=0.0, updated=65749, actions=240640, custom actions=[], active item id=-1, error=null} - -u0_a2 17243 197 998628 24932 ffffffff 00000000 S com.amazon.device.controllermanager -u0_a2 17374 197 995368 20764 ffffffff 00000000 S com.amazon.device.controllermanager:BluetoothReceiver""" -GET_PROPERTIES_DICT5 = { - "screen_on": True, - "awake": True, - "wake_lock_size": 2, - "current_app": "com.amazon.tv.launcher", - "media_session_state": 2, - "running_apps": ["com.amazon.device.controllermanager", "com.amazon.device.controllermanager:BluetoothReceiver"], - "hdmi_input": None, -} - -GET_PROPERTIES_DICT_NONE = { - "screen_on": None, - "awake": None, - "wake_lock_size": None, - "media_session_state": None, - "current_app": None, - "running_apps": None, - "hdmi_input": None, -} STATE_NONE = (None, None, None, None) STATE_DETECTION_RULES1 = {"com.amazon.tv.launcher": ["off"]} From 4e25aa0321c68f52db98fa29bdeda933a1dadcdc Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Sat, 15 Jan 2022 13:33:21 -0800 Subject: [PATCH 9/9] Cleanup --- tests/test_firetv_sync.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/tests/test_firetv_sync.py b/tests/test_firetv_sync.py index 6f26a21..7dc61c0 100644 --- a/tests/test_firetv_sync.py +++ b/tests/test_firetv_sync.py @@ -25,17 +25,6 @@ RUNNING_APPS_LIST = ["com.netflix.ninja", "com.amazon.device.controllermanager"] -STATE1 = (constants.STATE_OFF, None, None, None) - -STATE2 = (constants.STATE_STANDBY, None, None, None) - -STATE3 = ( - constants.STATE_IDLE, - "com.amazon.tv.launcher", - ["com.amazon.device.controllermanager", "com.amazon.device.controllermanager:BluetoothReceiver"], - None, -) - STATE_NONE = (None, None, None, None) STATE_DETECTION_RULES1 = {"com.amazon.tv.launcher": ["off"]}