Merge 2c4935e into 7165dfd

JeffLIrion · Dec 22, 2020 · 0287e4c · 0287e4c
2 parents 7165dfd + 2c4935e
commit 0287e4c
Show file tree

Hide file tree

Showing 4 changed files with 113 additions and 0 deletions.
diff --git a/adb_shell/adb_device.py b/adb_shell/adb_device.py
@@ -365,6 +365,36 @@ def _streaming_service(self, service, command, transport_timeout_s=None, read_ti
             for line in stream:
                 yield line
 
+    def exec_out(self, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None, decode=True):
+        """Send an ADB ``exec-out`` command to the device.
+
+        https://www.linux-magazine.com/Issues/2017/195/Ask-Klaus
+
+        Parameters
+        ----------
+        command : str
+            The exec-out command that will be sent
+        transport_timeout_s : float, None
+            Timeout in seconds for sending and receiving packets, or ``None``; see :meth:`BaseTransport.bulk_read() <adb_shell.transport.base_transport.BaseTransport.bulk_read>`
+            and :meth:`BaseTransport.bulk_write() <adb_shell.transport.base_transport.BaseTransport.bulk_write>`
+        read_timeout_s : float
+            The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`AdbDevice._read`
+        timeout_s : float, None
+            The total time in seconds to wait for the ADB command to finish
+        decode : bool
+            Whether to decode the output to utf8 before returning
+
+        Returns
+        -------
+        bytes, str
+            The output of the ADB exec-out command as a string if ``decode`` is True, otherwise as bytes.
+
+        """
+        if not self.available:
+            raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established.  (Did you call `AdbDevice.connect()`?)")
+
+        return self._service(b'exec', command.encode('utf8'), transport_timeout_s, read_timeout_s, timeout_s, decode)
+
     def root(self, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None):
         """Gain root access.
 

diff --git a/adb_shell/adb_device_async.py b/adb_shell/adb_device_async.py
@@ -360,6 +360,36 @@ async def _streaming_service(self, service, command, transport_timeout_s=None, r
             async for line in stream:
                 yield line
 
+    async def exec_out(self, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None, decode=True):
+        """Send an ADB ``exec-out`` command to the device.
+
+        https://www.linux-magazine.com/Issues/2017/195/Ask-Klaus
+
+        Parameters
+        ----------
+        command : str
+            The exec-out command that will be sent
+        transport_timeout_s : float, None
+            Timeout in seconds for sending and receiving packets, or ``None``; see :meth:`BaseTransport.bulk_read() <adb_shell.transport.base_transport.BaseTransport.bulk_read>`
+            and :meth:`BaseTransport.bulk_write() <adb_shell.transport.base_transport.BaseTransport.bulk_write>`
+        read_timeout_s : float
+            The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`AdbDevice._read`
+        timeout_s : float, None
+            The total time in seconds to wait for the ADB command to finish
+        decode : bool
+            Whether to decode the output to utf8 before returning
+
+        Returns
+        -------
+        bytes, str
+            The output of the ADB exec-out command as a string if ``decode`` is True, otherwise as bytes.
+
+        """
+        if not self.available:
+            raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established.  (Did you call `AdbDevice.connect()`?)")
+
+        return await self._service(b'exec', command.encode('utf8'), transport_timeout_s, read_timeout_s, timeout_s, decode)
+
     async def root(self, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None):
         """Gain root access.
 

diff --git a/tests/test_adb_device.py b/tests/test_adb_device.py
@@ -48,6 +48,12 @@ def tearDown(self):
         self.assertFalse(self.device._transport._bulk_read)
 
     def test_adb_connection_error(self):
+        with self.assertRaises(exceptions.AdbConnectionError):
+            self.device.exec_out('FAIL')
+
+        with self.assertRaises(exceptions.AdbConnectionError):
+            self.device.root()
+
         with self.assertRaises(exceptions.AdbConnectionError):
             self.device.shell('FAIL')
 
@@ -490,6 +496,28 @@ def test_root(self):
             self.device.root()
             patch_service.assert_called_once()
 
+
+    # ======================================================================= #
+    #                                                                         #
+    #                          `exec_out` test                                #
+    #                                                                         #
+    # ======================================================================= #
+    def test_exec_out(self):
+        self.assertTrue(self.device.connect())
+
+        # Provide the `bulk_read` return values
+        self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
+                                                          AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=b'TEST'),
+                                                          AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))
+
+        self.device._transport._bulk_read = b''.join([b'OKAY\x14\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb0\xb4\xbe\xa6',
+                                                      b'WRTE\x14\x00\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00J\x01\x00\x00\xa8\xad\xab\xba',
+                                                      b'TEST\n',
+                                                      b'',
+                                                      b'CLSE\x14\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xbc\xb3\xac\xba'])
+
+        self.assertEqual(self.device.exec_out("echo 'TEST'"), "TEST\n")
+
     # ======================================================================= #
     #                                                                         #
     #                         `filesync` tests                                #

diff --git a/tests/test_adb_device_async.py b/tests/test_adb_device_async.py
@@ -51,6 +51,12 @@ def tearDown(self):
 
     @awaiter
     async def test_adb_connection_error(self):
+        with self.assertRaises(exceptions.AdbConnectionError):
+            await self.device.exec_out('FAIL')
+
+        with self.assertRaises(exceptions.AdbConnectionError):
+            await self.device.root()
+
         with self.assertRaises(exceptions.AdbConnectionError):
             await self.device.shell('FAIL')
 
@@ -525,6 +531,25 @@ async def test_root(self):
             patch_service.assert_called_once()
 
 
+    # ======================================================================= #
+    #                                                                         #
+    #                         `exec_out` test                                 #
+    #                                                                         #
+    # ======================================================================= #
+    @awaiter
+    async def test_exec_out(self):
+        self.assertTrue(await self.device.connect())
+
+        # Provide the `bulk_read` return values
+        self.device._transport._bulk_read = b''.join([b'OKAY\x14\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb0\xb4\xbe\xa6',
+                                                      b'WRTE\x14\x00\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00J\x01\x00\x00\xa8\xad\xab\xba',
+                                                      b'TEST\n',
+                                                      b'',
+                                                      b'CLSE\x14\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xbc\xb3\xac\xba'])
+
+        self.assertEqual(await self.device.exec_out("echo 'TEST'"), "TEST\n")
+
+
     # ======================================================================= #
     #                                                                         #
     #                         `filesync` tests                                #