diff --git a/androidtv/adb_manager.py b/androidtv/adb_manager.py index 8c56af7e..8e4bd853 100644 --- a/androidtv/adb_manager.py +++ b/androidtv/adb_manager.py @@ -141,6 +141,58 @@ def connect(self, always_log_errors=True, auth_timeout_s=DEFAULT_AUTH_TIMEOUT_S) self._available = False return False + def pull(self, local_path, device_path): + """Pull a file from the device using the Python ADB implementation. + + Parameters + ---------- + local_path : str + The path where the file will be saved + device_path : str + The file on the device that will be pulled + + """ + if not self.available: + _LOGGER.debug("ADB command not sent to %s because adb-shell connection is not established: pull(%s, %s)", self.host, local_path, device_path) + return + + if self._adb_lock.acquire(**LOCK_KWARGS): # pylint: disable=unexpected-keyword-arg + _LOGGER.debug("Sending command to %s via adb-shell: pull(%s, %s)", self.host, local_path, device_path) + try: + self._adb.pull(device_path, local_path) + finally: + self._adb_lock.release() + + # Lock could not be acquired + _LOGGER.warning("ADB command not sent to %s because adb-shell lock not acquired: pull(%s, %s)", self.host, local_path, device_path) + return + + def push(self, local_path, device_path): + """Push a file to the device using the Python ADB implementation. + + Parameters + ---------- + local_path : str + The file that will be pushed to the device + device_path : str + The path where the file will be saved on the device + + """ + if not self.available: + _LOGGER.debug("ADB command not sent to %s because adb-shell connection is not established: push(%s, %s)", self.host, local_path, device_path) + return + + if self._adb_lock.acquire(**LOCK_KWARGS): # pylint: disable=unexpected-keyword-arg + _LOGGER.debug("Sending command to %s via adb-shell: push(%s, %s)", self.host, local_path, device_path) + try: + self._adb.push(local_path, device_path) + finally: + self._adb_lock.release() + + # Lock could not be acquired + _LOGGER.warning("ADB command not sent to %s because adb-shell lock not acquired: push(%s, %s)", self.host, local_path, device_path) + return + def shell(self, cmd): """Send an ADB command using the Python ADB implementation. @@ -309,6 +361,58 @@ def connect(self, always_log_errors=True): self._available = False return False + def pull(self, local_path, device_path): + """Pull a file from the device using an ADB server. + + Parameters + ---------- + local_path : str + The path where the file will be saved + device_path : str + The file on the device that will be pulled + + """ + if not self.available: + _LOGGER.debug("ADB command not sent to %s via ADB server %s:%s because pure-python-adb connection is not established: pull(%s, %s)", self.host, self.adb_server_ip, self.adb_server_port, local_path, device_path) + return + + if self._adb_lock.acquire(**LOCK_KWARGS): # pylint: disable=unexpected-keyword-arg + _LOGGER.debug("Sending command to %s via ADB server %s:%s: pull(%s, %s)", self.host, self.adb_server_ip, self.adb_server_port, local_path, device_path) + try: + self._adb_device.pull(device_path, local_path) + finally: + self._adb_lock.release() + + # Lock could not be acquired + _LOGGER.warning("ADB command not sent to %s via ADB server %s:%s: pull(%s, %s)", self.host, self.adb_server_ip, self.adb_server_port, local_path, device_path) + return + + def push(self, local_path, device_path): + """Push a file to the device using an ADB server. + + Parameters + ---------- + local_path : str + The file that will be pushed to the device + device_path : str + The path where the file will be saved on the device + + """ + if not self.available: + _LOGGER.debug("ADB command not sent to %s via ADB server %s:%s because pure-python-adb connection is not established: push(%s, %s)", self.host, self.adb_server_ip, self.adb_server_port, local_path, device_path) + return + + if self._adb_lock.acquire(**LOCK_KWARGS): # pylint: disable=unexpected-keyword-arg + _LOGGER.debug("Sending command to %s via ADB server %s:%s: push(%s, %s)", self.host, self.adb_server_ip, self.adb_server_port, local_path, device_path) + try: + self._adb_device.push(local_path, device_path) + finally: + self._adb_lock.release() + + # Lock could not be acquired + _LOGGER.warning("ADB command not sent to %s via ADB server %s:%s: push(%s, %s)", self.host, self.adb_server_ip, self.adb_server_port, local_path, device_path) + return + def shell(self, cmd): """Send an ADB command using an ADB server. diff --git a/androidtv/basetv.py b/androidtv/basetv.py index 6bc90331..d4f48abe 100644 --- a/androidtv/basetv.py +++ b/androidtv/basetv.py @@ -120,6 +120,38 @@ def adb_shell(self, cmd): """ return self._adb.shell(cmd) + def adb_pull(self, local_path, device_path): + """Pull a file from the device. + + This calls :py:meth:`androidtv.adb_manager.ADBPython.pull` or :py:meth:`androidtv.adb_manager.ADBServer.pull`, + depending on whether the Python ADB implementation or an ADB server is used for communicating with the device. + + Parameters + ---------- + local_path : str + The path where the file will be saved + device_path : str + The file on the device that will be pulled + + """ + return self._adb.pull(local_path, device_path) + + def adb_push(self, local_path, device_path): + """Push a file to the device. + + This calls :py:meth:`androidtv.adb_manager.ADBPython.push` or :py:meth:`androidtv.adb_manager.ADBServer.push`, + depending on whether the Python ADB implementation or an ADB server is used for communicating with the device. + + Parameters + ---------- + local_path : str + The file that will be pushed to the device + device_path : str + The path where the file will be saved on the device + + """ + return self._adb.push(local_path, device_path) + def adb_connect(self, always_log_errors=True, auth_timeout_s=constants.DEFAULT_AUTH_TIMEOUT_S): """Connect to an Android TV / Fire TV device. diff --git a/setup.py b/setup.py index fd54d2f1..9260fe5b 100644 --- a/setup.py +++ b/setup.py @@ -9,7 +9,7 @@ author='Jeff Irion', author_email='jefflirion@users.noreply.github.com', packages=['androidtv'], - install_requires=['adb-shell>=0.0.7', 'pure-python-adb>=0.2.2.dev0'], + install_requires=['adb-shell>=0.0.9', 'pure-python-adb>=0.2.2.dev0'], classifiers=[ 'License :: OSI Approved :: MIT License', 'Operating System :: OS Independent', diff --git a/tests/patchers.py b/tests/patchers.py index abfde3ce..7c12464f 100644 --- a/tests/patchers.py +++ b/tests/patchers.py @@ -23,6 +23,12 @@ def connect(self, *args, **kwargs): """Try to connect to a device.""" raise NotImplementedError + def push(self, *args, **kwargs): + """Push a file to the device.""" + + def pull(self, *args, **kwargs): + """Pull a file from the device.""" + def shell(self, cmd): """Send an ADB shell command.""" return None @@ -73,6 +79,12 @@ def get_serial_no(self): """Get the serial number for the device (IP:PORT).""" return self.host + def push(self, *args, **kwargs): + """Push a file to the device.""" + + def pull(self, *args, **kwargs): + """Pull a file from the device.""" + def shell(self, cmd): """Send an ADB shell command.""" raise NotImplementedError @@ -117,6 +129,10 @@ def shell_fail_server(self, cmd): return {"python": patch("{}.AdbDeviceFake.shell".format(__name__), shell_fail_python), "server": patch("{}.DeviceFake.shell".format(__name__), shell_fail_server)} +PATCH_PUSH = {"python": patch("{}.AdbDeviceFake.push".format(__name__)), "server": patch("{}.DeviceFake.push".format(__name__))} + +PATCH_PULL = {"python": patch("{}.AdbDeviceFake.pull".format(__name__)), "server": patch("{}.DeviceFake.pull".format(__name__))} + PATCH_ADB_DEVICE = patch("androidtv.adb_manager.AdbDevice", AdbDeviceFake) diff --git a/tests/test_adb_manager.py b/tests/test_adb_manager.py index 8eaeadc1..5ab89516 100644 --- a/tests/test_adb_manager.py +++ b/tests/test_adb_manager.py @@ -110,7 +110,7 @@ def test_connect_fail_lock(self): self.assertFalse(self.adb._available) def test_adb_shell_fail(self): - """Test when an ADB command is not sent because the device is unavailable. + """Test when an ADB shell command is not sent because the device is unavailable. """ self.assertFalse(self.adb.available) @@ -123,13 +123,67 @@ def test_adb_shell_fail(self): self.assertIsNone(self.adb.shell("TEST")) def test_adb_shell_success(self): - """Test when an ADB command is successfully sent. + """Test when an ADB shell command is successfully sent. """ with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("TEST")[self.PATCH_KEY]: self.assertTrue(self.adb.connect()) self.assertEqual(self.adb.shell("TEST"), "TEST") - + + def test_adb_push_fail(self): + """Test when an ADB push command is not executed because the device is unavailable. + + """ + self.assertFalse(self.adb.available) + with patchers.patch_connect(True)[self.PATCH_KEY]: + with patchers.PATCH_PUSH[self.PATCH_KEY] as patch_push: + self.adb.push("TEST_LOCAL_PATCH", "TEST_DEVICE_PATH") + patch_push.assert_not_called() + + with patchers.patch_connect(True)[self.PATCH_KEY]: + with patchers.PATCH_PUSH[self.PATCH_KEY] as patch_push: + self.assertTrue(self.adb.connect()) + with patch.object(self.adb, '_adb_lock', LockedLock): + self.adb.push("TEST_LOCAL_PATH", "TEST_DEVICE_PATH") + patch_push.assert_not_called() + + def test_adb_push_success(self): + """Test when an ADB push command is successfully executed. + + """ + with patchers.patch_connect(True)[self.PATCH_KEY]: + with patchers.PATCH_PUSH[self.PATCH_KEY] as patch_push: + self.assertTrue(self.adb.connect()) + self.adb.push("TEST_LOCAL_PATH", "TEST_DEVICE_PATH") + self.assertEqual(patch_push.call_count, 1) + + def test_adb_pull_fail(self): + """Test when an ADB pull command is not executed because the device is unavailable. + + """ + self.assertFalse(self.adb.available) + with patchers.patch_connect(True)[self.PATCH_KEY]: + with patchers.PATCH_PULL[self.PATCH_KEY] as patch_pull: + self.adb.pull("TEST_LOCAL_PATCH", "TEST_DEVICE_PATH") + patch_pull.assert_not_called() + + with patchers.patch_connect(True)[self.PATCH_KEY]: + with patchers.PATCH_PULL[self.PATCH_KEY] as patch_pull: + self.assertTrue(self.adb.connect()) + with patch.object(self.adb, '_adb_lock', LockedLock): + self.adb.pull("TEST_LOCAL_PATH", "TEST_DEVICE_PATH") + patch_pull.assert_not_called() + + def test_adb_pull_success(self): + """Test when an ADB pull command is successfully executed. + + """ + with patchers.patch_connect(True)[self.PATCH_KEY]: + with patchers.PATCH_PULL[self.PATCH_KEY] as patch_pull: + self.assertTrue(self.adb.connect()) + self.adb.pull("TEST_LOCAL_PATH", "TEST_DEVICE_PATH") + self.assertEqual(patch_pull.call_count, 1) + class TestADBServer(TestADBPython): """Test the `ADBServer` class.""" diff --git a/tests/test_basetv.py b/tests/test_basetv.py index a1a74b41..2b7b372d 100644 --- a/tests/test_basetv.py +++ b/tests/test_basetv.py @@ -106,6 +106,22 @@ def test_adb_close(self): else: self.assertTrue(self.btv.available) + def test_adb_pull(self): + """Test that the ``adb_pull`` method works correctly. + + """ + with patchers.PATCH_PULL[self.PATCH_KEY] as patch_pull: + self.btv.adb_pull("TEST_LOCAL_PATCH", "TEST_DEVICE_PATH") + self.assertEqual(patch_pull.call_count, 1) + + def test_adb_push(self): + """Test that the ``adb_push`` method works correctly. + + """ + with patchers.PATCH_PUSH[self.PATCH_KEY] as patch_push: + self.btv.adb_push("TEST_LOCAL_PATCH", "TEST_DEVICE_PATH") + self.assertEqual(patch_push.call_count, 1) + def test_keys(self): """Test that the key methods send the correct commands.