Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ADB server fixes #157

Merged
merged 5 commits into from Apr 18, 2020
Merged

ADB server fixes #157

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
41b5025
Change 'self.available' -> 'self._available'
JeffLIrion Apr 18, 2020
753b30c
Make 'close()' set 'self._available' to False
JeffLIrion Apr 18, 2020
34e1c65
Add '_was_available' attribute
JeffLIrion Apr 18, 2020
d803dfc
Simplify 'available' property
JeffLIrion Apr 18, 2020
1c0e619
Remove inapplicable test
JeffLIrion Apr 18, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
57 changes: 14 additions & 43 deletions androidtv/adb_manager.py
View file
Open in desktop
Expand Up @@ -279,8 +279,9 @@ def __init__(self, host, port=5555, adb_server_ip='', adb_server_port=5037):
self._adb_client = None
self._adb_device = None

# keep track of whether the ADB connection is intact
# keep track of whether the ADB connection is/was intact
self._available = False
self._was_available = False

# use a lock to make sure that ADB commands don't overlap
self._adb_lock = threading.Lock()
Expand All @@ -295,52 +296,18 @@ def available(self):
Whether or not the ADB connection is intact

"""
if not self._adb_client:
return False

try:
# make sure the server is available
adb_devices = self._adb_client.devices()

# make sure the device is available
try:
# case 1: the device is currently available
host_port = '{}:{}'.format(self.host, self.port)
if any([host_port in dev.get_serial_no() for dev in adb_devices]):
if not self._available:
self._available = True
return True

# case 2: the device is not currently available
if self._available:
_LOGGER.error('ADB server is not connected to the device.')
self._available = False
return False

except RuntimeError:
if self._available:
_LOGGER.error('ADB device is unavailable; encountered an error when searching for device.')
self._available = False
return False

except RuntimeError:
if self._available:
_LOGGER.error('ADB server is unavailable.')
self._available = False
if not self._adb_client or not self._adb_device:
return False

except Exception as exc: # noqa pylint: disable=broad-except
if self._available:
_LOGGER.error('ADB server is unavailable, error: %s', exc)
self._available = False
return False
return self._available

def close(self):
"""Close the ADB server socket connection.

Currently, this doesn't do anything.
Currently, this doesn't do anything except set ``self._available = False``.

"""
self._available = False

def connect(self, always_log_errors=True):
"""Connect to an Android TV / Fire TV device.
Expand All @@ -367,29 +334,33 @@ def connect(self, always_log_errors=True):
if self._adb_device:
_LOGGER.debug("ADB connection to %s:%d via ADB server %s:%d successfully established", self.host, self.port, self.adb_server_ip, self.adb_server_port)
self._available = True
self._was_available = True
return True

# ADB connection attempt failed (without an exception)
if self._available or always_log_errors:
_LOGGER.warning("Couldn't connect to %s:%d via ADB server %s:%d", self.host, self.port, self.adb_server_ip, self.adb_server_port)
if self._was_available or always_log_errors:
_LOGGER.warning("Couldn't connect to %s:%d via ADB server %s:%d because the server is not connected to the device", self.host, self.port, self.adb_server_ip, self.adb_server_port)

self.close()
self._available = False
self._was_available = False
return False

# ADB connection attempt failed
except Exception as exc: # noqa pylint: disable=broad-except
if self._available or always_log_errors:
if self._was_available or always_log_errors:
_LOGGER.warning("Couldn't connect to %s:%d via ADB server %s:%d, error: %s", self.host, self.port, self.adb_server_ip, self.adb_server_port, exc)

self.close()
self._available = False
self._was_available = False
return False

except LockNotAcquiredException:
_LOGGER.warning("Couldn't connect to %s:%d via ADB server %s:%d because pure-python-adb lock not acquired.", self.host, self.port, self.adb_server_ip, self.adb_server_port)
self.close()
self._available = False
self._was_available = False
return False

def pull(self, local_path, device_path):
Expand Down Expand Up @@ -463,7 +434,7 @@ def shell(self, cmd):
The response from the device, if there is a response

"""
if not self._available:
if not self.available:
_LOGGER.debug("ADB command not sent to %s:%d via ADB server %s:%d because pure-python-adb connection is not established: %s", self.host, self.port, self.adb_server_ip, self.adb_server_port, cmd)
return None

Expand Down
33 changes: 0 additions & 33 deletions tests/test_adb_manager.py
View file
Open in desktop
Expand Up @@ -297,39 +297,6 @@ def setUp(self):
"""
self.adb = ADBServer('HOST', 5555, 'ADB_SERVER_IP')

def test_available(self):
"""Test that the ``available`` property works correctly.

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

with patchers.patch_connect(True)[self.PATCH_KEY], patch('{}.patchers.ClientFakeSuccess.devices'.format(__name__), side_effect=RuntimeError):
self.assertFalse(self.adb.available)

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

with patch('{}.patchers.DeviceFake.get_serial_no'.format(__name__), side_effect=RuntimeError):
self.assertFalse(self.adb.available)

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

with patch.object(self.adb._adb_client, 'devices', return_value=[]):
self.assertFalse(self.adb.available)

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

with patchers.PATCH_CONNECT_FAIL_CUSTOM_EXCEPTION[self.PATCH_KEY]:
self.assertFalse(self.adb.available)
self.assertFalse(self.adb._available)

def test_connect_fail_server(self):
"""Test that the ``connect`` method works correctly.

Expand Down