Skip to content

Commit

Permalink
Merge bc483ec into 7608b72
Browse files Browse the repository at this point in the history
  • Loading branch information
@JeffLIrion
JeffLIrion committed Sep 23, 2019
2 parents 7608b72 + bc483ec commit d1e5c5a
Show file tree
Hide file tree
Showing 9 changed files with 125 additions and 86 deletions.
30 changes: 22 additions & 8 deletions androidtv/adb_manager.py
View file
Expand Up @@ -11,8 +11,8 @@
import sys
import threading

from adb.adb_commands import AdbCommands
from adb.sign_pythonrsa import PythonRSASigner
from adb_shell.adb_device import AdbDevice
from adb_shell.auth.sign_pythonrsa import PythonRSASigner
from adb_messenger.client import Client

_LOGGER = logging.getLogger(__name__)
Expand All @@ -38,7 +38,7 @@ class ADBPython(object):
def __init__(self, host, adbkey=''):
self.host = host
self.adbkey = adbkey
self._adb = None
self._adb = AdbDevice(serial=self.host)

# keep track of whether the ADB connection is intact
self._available = False
Expand All @@ -56,7 +56,14 @@ def available(self):
Whether or not the ADB connection is intact
"""
return bool(self._adb)
return self._adb.available

def close(self):
"""Close the ADB socket connection.
"""
self._adb.close()
self._available = False

def connect(self, always_log_errors=True):
"""Connect to an Android TV / Fire TV device.
Expand Down Expand Up @@ -93,9 +100,9 @@ def connect(self, always_log_errors=True):
signer = PythonRSASigner(pub, priv)

# Connect to the device
self._adb = AdbCommands().ConnectDevice(serial=self.host, rsa_keys=[signer], default_timeout_ms=9000)
self._adb.connect(rsa_keys=[signer], timeout_s=9000)
else:
self._adb = AdbCommands().ConnectDevice(serial=self.host, default_timeout_ms=9000)
self._adb.connect(timeout_s=9000)

# ADB connection successfully established
self._available = True
Expand All @@ -108,7 +115,7 @@ def connect(self, always_log_errors=True):
_LOGGER.warning("Couldn't connect to host %s, error: %s", self.host, serr.strerror)

# ADB connection attempt failed
self._adb = None
self._adb.close()
self._available = False

finally:
Expand Down Expand Up @@ -138,7 +145,7 @@ def shell(self, cmd):
if self._adb_lock.acquire(**LOCK_KWARGS): # pylint: disable=unexpected-keyword-arg
_LOGGER.debug("Sending command to %s via python-adb: %s", self.host, cmd)
try:
return self._adb.Shell(cmd)
return self._adb.shell(cmd)
finally:
self._adb_lock.release()
else:
Expand Down Expand Up @@ -218,6 +225,13 @@ def available(self):
self._available = False
return False

def close(self):
"""Close the ADB server socket connection.
Currently, this doesn't do anything.
"""

def connect(self, always_log_errors=True):
"""Connect to an Android TV / Fire TV device.
Expand Down
2 changes: 1 addition & 1 deletion setup.py
View file
Expand Up @@ -9,7 +9,7 @@
author='Jeff Irion',
author_email='jefflirion@users.noreply.github.com',
packages=['androidtv'],
install_requires=['rsa', 'adb-homeassistant>=1.3.2', 'pure-python-adb-homeassistant>=0.1.7.dev0'],
install_requires=['adb-shell>=0.0.1', 'pure-python-adb-homeassistant>=0.1.7.dev0', 'pyasn1', 'rsa'],
classifiers=[
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
Expand Down
57 changes: 29 additions & 28 deletions tests/patchers.py
View file
Expand Up @@ -8,32 +8,22 @@
from mock import patch


class AdbCommandsFake(object):
"""A fake of the `adb.adb_commands.AdbCommands` class."""
class AdbDeviceFake(object):
"""A fake of the `adb_shell.adb_commands.AdbDevice` class."""
def __init__(self, *args, **kwargs):
self.available = False

def ConnectDevice(self, *args, **kwargs):
def connect(self, *args, **kwargs):
"""Try to connect to a device."""
raise NotImplementedError

def Shell(self, cmd):
def shell(self, cmd):
"""Send an ADB shell command."""
raise NotImplementedError


class AdbCommandsFakeSuccess(AdbCommandsFake):
"""A fake of the `adb.adb_commands.AdbCommands` class when the connection attempt succeeds."""

def ConnectDevice(self, *args, **kwargs):
"""Successfully connect to a device."""
return self


class AdbCommandsFakeFail(AdbCommandsFake):
"""A fake of the `adb.adb_commands.AdbCommands` class when the connection attempt fails."""
return None

def ConnectDevice(self, *args, **kwargs):
"""Fail to connect to a device."""
raise socket_error
def close(self):
"""Close the socket connection."""
self.available = False


class ClientFakeSuccess(object):
Expand Down Expand Up @@ -85,23 +75,31 @@ def shell(self, cmd):


def patch_connect(success):
"""Mock the `adb.adb_commands.AdbCommands` and `adb_messenger.client.Client` classes."""
"""Mock the `adb_shell.adb_device.AdbDevice` and `adb_messenger.client.Client` classes."""

def connect_success_python(self, *args, **kwargs):
"""Mock the `AdbDeviceFake.connect` method when it succeeds."""
self.available = True

def connect_fail_python(self, *args, **kwargs):
"""Mock the `AdbDeviceFake.connect` method when it fails."""
raise socket_error

if success:
return {'python': patch('androidtv.adb_manager.AdbCommands', AdbCommandsFakeSuccess), 'server': patch('androidtv.adb_manager.Client', ClientFakeSuccess)}
return {'python': patch('androidtv.adb_manager.AdbCommands', AdbCommandsFakeFail), 'server': patch('androidtv.adb_manager.Client', ClientFakeFail)}
return {'python': patch('{}.AdbDeviceFake.connect'.format(__name__), connect_success_python), 'server': patch('androidtv.adb_manager.Client', ClientFakeSuccess)}
return {'python': patch('{}.AdbDeviceFake.connect'.format(__name__), connect_fail_python), 'server': patch('androidtv.adb_manager.Client', ClientFakeFail)}


def patch_shell(response=None, error=False):
"""Mock the `AdbCommandsFake.Shell` and `DeviceFake.shell` methods."""
"""Mock the `AdbDeviceFake.shell` and `DeviceFake.shell` methods."""

def shell_success(self, cmd):
"""Mock the `AdbCommandsFake.Shell` and `DeviceFake.shell` methods when they are successful."""
"""Mock the `AdbDeviceFake.shell` and `DeviceFake.shell` methods when they are successful."""
self.shell_cmd = cmd
return response

def shell_fail_python(self, cmd):
"""Mock the `AdbCommandsFake.Shell` method when it fails."""
"""Mock the `AdbDeviceFake.shell` method when it fails."""
self.shell_cmd = cmd
raise AttributeError

Expand All @@ -111,5 +109,8 @@ def shell_fail_server(self, cmd):
raise ConnectionResetError

if not error:
return {'python': patch('{}.AdbCommandsFake.Shell'.format(__name__), shell_success), 'server': patch('{}.DeviceFake.shell'.format(__name__), shell_success)}
return {'python': patch('{}.AdbCommandsFake.Shell'.format(__name__), shell_fail_python), 'server': patch('{}.DeviceFake.shell'.format(__name__), shell_fail_server)}
return {'python': patch('{}.AdbDeviceFake.shell'.format(__name__), shell_success), 'server': patch('{}.DeviceFake.shell'.format(__name__), shell_success)}
return {'python': patch('{}.AdbDeviceFake.shell'.format(__name__), shell_fail_python), 'server': patch('{}.DeviceFake.shell'.format(__name__), shell_fail_server)}


patch_adb_device = patch('androidtv.adb_manager.AdbDevice', AdbDeviceFake)
29 changes: 26 additions & 3 deletions tests/test_adb_manager.py
View file
Expand Up @@ -72,7 +72,8 @@ def setUp(self):
"""Create an `ADBPython` instance.
"""
self.adb = ADBPython('IP:PORT')
with patchers.patch_adb_device, patchers.patch_connect(True)[self.PATCH_KEY]:
self.adb = ADBPython('IP:5555')

def test_connect_success(self):
"""Test when the connect attempt is successful.
Expand Down Expand Up @@ -123,7 +124,7 @@ def setUp(self):
"""Create an `ADBServer` instance.
"""
self.adb = ADBServer('IP:PORT', 'ADB_SERVER_IP')
self.adb = ADBServer('IP:5555', 'ADB_SERVER_IP')

def test_available(self):
"""Test that the ``available`` property works correctly.
Expand Down Expand Up @@ -171,7 +172,8 @@ def setUp(self):
"""Create an `ADBPython` instance.
"""
self.adb = ADBPython('IP:PORT', 'adbkey')
with patchers.patch_adb_device, patchers.patch_connect(True)[self.PATCH_KEY]:
self.adb = ADBPython('IP:5555', 'adbkey')

def test_connect_success_with_priv_key(self):
"""Test when the connect attempt is successful when using a private key.
Expand All @@ -190,3 +192,24 @@ def test_connect_success_with_priv_pub_key(self):
self.assertTrue(self.adb.connect())
self.assertTrue(self.adb.available)
self.assertTrue(self.adb._available)


class TestADBPythonClose(unittest.TestCase):
"""Test the `ADBPython.close` method."""

PATCH_KEY = 'python'

def test_close(self):
"""Test the `ADBPython.close` method.
"""
with patchers.patch_adb_device, patchers.patch_connect(True)[self.PATCH_KEY]:
self.adb = ADBPython('IP:5555')

with patchers.patch_connect(True)[self.PATCH_KEY]:
self.assertTrue(self.adb.connect())
self.assertTrue(self.adb.available)
self.assertTrue(self.adb._available)

self.adb.close()
self.assertFalse(self.adb.available)
self.assertFalse(self.adb._available)
16 changes: 8 additions & 8 deletions tests/test_androidtv.py
View file
Expand Up @@ -513,8 +513,8 @@ class TestAndroidTVPython(unittest.TestCase):
ADB_ATTR = '_adb'

def setUp(self):
with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.atv = AndroidTV('IP:PORT')
with patchers.patch_adb_device, patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.atv = AndroidTV('IP:5555')

def test_turn_on_off(self):
"""Test that the ``AndroidTV.turn_on`` and ``AndroidTV.turn_off`` methods work correctly.
Expand Down Expand Up @@ -982,7 +982,7 @@ class TestAndroidTVServer(TestAndroidTVPython):

def setUp(self):
with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.atv = AndroidTV('IP:PORT', adb_server_ip='ADB_SERVER_IP')
self.atv = AndroidTV('IP:5555', adb_server_ip='ADB_SERVER_IP')


class TestStateDetectionRulesValidator(unittest.TestCase):
Expand All @@ -992,11 +992,11 @@ def test_state_detection_rules_validator(self):
"""
with patchers.patch_connect(True)['python'], patchers.patch_shell('')['python']:
# Make sure that no error is raised when the state detection rules are valid
atv1 = AndroidTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES1)
atv2 = AndroidTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES2)
atv3 = AndroidTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES3)
atv4 = AndroidTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES4)
atv5 = AndroidTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES5)
atv1 = AndroidTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES1)
atv2 = AndroidTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES2)
atv3 = AndroidTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES3)
atv4 = AndroidTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES4)
atv5 = AndroidTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES5)


if __name__ == "__main__":
Expand Down
36 changes: 18 additions & 18 deletions tests/test_basetv.py
View file
Expand Up @@ -87,8 +87,8 @@ class TestBaseTVPython(unittest.TestCase):
ADB_ATTR = '_adb'

def setUp(self):
with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.btv = BaseTV('IP:PORT')
with patchers.patch_adb_device, patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.btv = BaseTV('IP:5555')

def test_available(self):
"""Test that the available property works correctly.
Expand Down Expand Up @@ -394,24 +394,24 @@ def test_state_detection_rules_validator(self):
"""
with patchers.patch_connect(True)['python'], patchers.patch_shell('')['python']:
# Make sure that no error is raised when the state detection rules are valid
btv1 = BaseTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES1)
btv2 = BaseTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES2)
btv3 = BaseTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES3)
btv4 = BaseTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES4)
btv5 = BaseTV('IP:PORT', state_detection_rules=STATE_DETECTION_RULES5)
btv1 = BaseTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES1)
btv2 = BaseTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES2)
btv3 = BaseTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES3)
btv4 = BaseTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES4)
btv5 = BaseTV('IP:5555', state_detection_rules=STATE_DETECTION_RULES5)

# Make sure that an error is raised when the state detection rules are invalid
self.assertRaises(TypeError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID1)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID2)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID3)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID4)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID5)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID6)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID7)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID8)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID9)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID10)
self.assertRaises(KeyError, BaseTV, 'IP:PORT', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID11)
self.assertRaises(TypeError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID1)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID2)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID3)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID4)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID5)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID6)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID7)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID8)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID9)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID10)
self.assertRaises(KeyError, BaseTV, 'IP:5555', '', '', 5037, state_detection_rules=STATE_DETECTION_RULES_INVALID11)

def test_wake_lock_size(self):
"""Check that the ``wake_lock_size`` property works correctly.
Expand Down
6 changes: 3 additions & 3 deletions tests/test_firetv.py
View file
Expand Up @@ -128,8 +128,8 @@ class TestFireTVPython(unittest.TestCase):
PATCH_KEY = 'python'

def setUp(self):
with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.ftv = FireTV('IP:PORT')
with patchers.patch_adb_device, patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.ftv = FireTV('IP:5555')

def test_turn_on_off(self):
"""Test that the ``FireTV.turn_on`` and ``FireTV.turn_off`` methods work correctly.
Expand Down Expand Up @@ -414,7 +414,7 @@ class TestFireTVServer(TestFireTVPython):

def setUp(self):
with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell('')[self.PATCH_KEY]:
self.ftv = FireTV('IP:PORT', adb_server_ip='ADB_SERVER_PORT')
self.ftv = FireTV('IP:5555', adb_server_ip='ADB_SERVER_PORT')


if __name__ == "__main__":
Expand Down

0 comments on commit d1e5c5a

Please sign in to comment.