Merge ef3b604 into e7b7382

aio_adb_shell/adb_device.py

@@ -437,6 +437,25 @@ async def _streaming_service(self, service, command, timeout_s=None, total_timeo
             async for line in stream:
                 yield line
 
+    async def root(self, timeout_s=None, total_timeout_s=constants.DEFAULT_TOTAL_TIMEOUT_S):
+        """Gain root access.
+
+        The device must be rooted in order for this to work.
+
+        Parameters
+        ----------
+        timeout_s : float, None
+            Timeout in seconds for sending and receiving packets, or ``None``; see :meth:`BaseHandle.bulk_read() <adb_shell.handle.base_handle.BaseHandle.bulk_read>`
+            and :meth:`BaseHandle.bulk_write() <adb_shell.handle.base_handle.BaseHandle.bulk_write>`
+        total_timeout_s : float
+            The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`AdbDevice._read`
+
+        """
+        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()`?)")
+
+        await self._service(b'root', b'', timeout_s, total_timeout_s, False)
+
     async def shell(self, command, timeout_s=None, total_timeout_s=constants.DEFAULT_TOTAL_TIMEOUT_S, decode=True):
         """Send an ADB shell command to the device.
 

tests/test_adb_device.py

@@ -16,6 +16,15 @@
 from .filesync_helpers import FileSyncMessage, FileSyncListMessage, FileSyncStatMessage
 from .keygen_stub import open_priv_pub
 
+try:
+    from unittest.mock import AsyncMock
+except ImportError:
+    from unittest.mock import MagicMock
+
+    class AsyncMock(MagicMock):
+        async def __call__(self, *args, **kwargs):
+            return super(AsyncMock, self).__call__(*args, **kwargs)
+
 
 # https://stackoverflow.com/a/7483862
 _LOGGER = logging.getLogger('aio_adb_shell.adb_device')
@@ -56,6 +65,9 @@ async def test_adb_connection_error(self):
             async_generator = self.device.streaming_shell('FAIL')
             await async_generator.__anext__()
 
+        with self.assertRaises(exceptions.AdbConnectionError):
+            await self.device.root()
+
         with self.assertRaises(exceptions.AdbConnectionError):
             await self.device.list('FAIL')
 
@@ -483,6 +495,20 @@ async def test_streaming_shell_dont_decode(self):
         self.assertEqual(await async_generator.__anext__(), b'123')
 
 
+    # ======================================================================= #
+    #                                                                         #
+    #                               `root` test                               #
+    #                                                                         #
+    # ======================================================================= #
+    @awaiter
+    async def test_root(self):
+        self.assertTrue(await self.device.connect())
+
+        with patch('aio_adb_shell.adb_device.AdbDevice._service', new_callable=AsyncMock) as patch_service:
+            await self.device.root()
+            patch_service.assert_called_once()
+
+
     # ======================================================================= #
     #                                                                         #
     #                         `filesync` tests                                #