Add ADB FileSync push and pull functionality (#127)

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.
 

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.
 

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',

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)
 
 

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."""

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.