ADB push fails with Android emulator

adb_shell/adb_device.py

@@ -757,8 +757,17 @@ def _read(self, expected_cmds, adb_info):
         start = time.time()
 
         while True:
-            msg = self._transport.bulk_read(constants.MESSAGE_SIZE, adb_info.transport_timeout_s)
-            _LOGGER.debug("bulk_read(%d): %s", constants.MESSAGE_SIZE, repr(msg))
+            # Read until `constants.MESSAGE_SIZE` bytes are received
+            msg = bytearray()
+            msg_length = constants.MESSAGE_SIZE
+            while msg_length > 0:
+                msg += self._transport.bulk_read(msg_length, adb_info.transport_timeout_s)
+                _LOGGER.debug("bulk_read(%d): %s", msg_length, repr(msg))
+                msg_length -= len(msg)
+
+                if time.time() - start > adb_info.read_timeout_s:
+                    raise exceptions.AdbTimeoutError("Took longer than %f seconds to read %d bytes" % (adb_info.read_timeout_s, constants.MESSAGE_SIZE))
+
             cmd, arg0, arg1, data_length, data_checksum = unpack(msg)
             command = constants.WIRE_TO_ID.get(cmd)
 

adb_shell/adb_device_async.py

@@ -752,8 +752,17 @@ async def _read(self, expected_cmds, adb_info):
         start = time.time()
 
         while True:
-            msg = await self._transport.bulk_read(constants.MESSAGE_SIZE, adb_info.transport_timeout_s)
-            _LOGGER.debug("bulk_read(%d): %s", constants.MESSAGE_SIZE, repr(msg))
+            # Read until `constants.MESSAGE_SIZE` bytes are received
+            msg = bytearray()
+            msg_length = constants.MESSAGE_SIZE
+            while msg_length > 0:
+                msg += await self._transport.bulk_read(msg_length, adb_info.transport_timeout_s)
+                _LOGGER.debug("bulk_read(%d): %s", msg_length, repr(msg))
+                msg_length -= len(msg)
+
+                if time.time() - start > adb_info.read_timeout_s:
+                    raise exceptions.AdbTimeoutError("Took longer than %f seconds to read %d bytes" % (adb_info.read_timeout_s, constants.MESSAGE_SIZE))
+
             cmd, arg0, arg1, data_length, data_checksum = unpack(msg)
             command = constants.WIRE_TO_ID.get(cmd)
 

adb_shell/adb_message.py

@@ -159,6 +159,17 @@ def __init__(self, command, arg0, arg1, data=b''):
         self.arg1 = arg1
         self.data = data
 
+    def __str__(self):
+        """A string containing info about this object.
+
+        Returns
+        -------
+        str
+            A description of this object
+
+        """
+        return "AdbMessage(command = {} = {}, arg0 = {}, arg1 = {}, data = {}, magic = {})".format(self.command, constants.WIRE_TO_ID[self.command], self.arg0, self.arg1, self.data, self.magic)
+
     def pack(self):
         """Returns this message in an over-the-wire format.
 

adb_shell/hidden_helpers.py

@@ -138,6 +138,17 @@ def __init__(self, local_id, remote_id, transport_timeout_s=None, read_timeout_s
         self.read_timeout_s = read_timeout_s if self.timeout_s is None else min(read_timeout_s, self.timeout_s)
         self.transport_timeout_s = self.read_timeout_s if transport_timeout_s is None else min(transport_timeout_s, self.read_timeout_s)
 
+    def __str__(self):
+        """A string containing info about this object.
+
+        Returns
+        -------
+        str
+            A description of this object
+
+        """
+        return "_AdbTransactionInfo(local_id = {}, remote_id = {}, timeout_s = {}, read_timeout_s = {}, transport_timeout_s = {})".format(self.local_id, self.remote_id, self.timeout_s, self.read_timeout_s, self.transport_timeout_s)
+
 
 class _FileSyncTransactionInfo(object):  # pylint: disable=too-few-public-methods
     """A class for storing info used during a single FileSync "transaction."
@@ -175,6 +186,17 @@ def __init__(self, recv_message_format, maxdata=constants.MAX_ADB_DATA):
 
         self._maxdata = maxdata
 
+    def __str__(self):
+        """A string containing info about this object.
+
+        Returns
+        -------
+        str
+            A description of this object
+
+        """
+        return "_FileSyncTransactionInfo(send_idx = {}, recv_message_format = {}, recv_message_size = {}, _maxdata = {})".format(self.send_idx, self.recv_message_format, self.recv_message_size, self._maxdata)
+
     def can_add_to_send_buffer(self, data_len):
         """Determine whether ``data_len`` bytes of data can be added to the send buffer without exceeding :const:`constants.MAX_ADB_DATA`.
 

tests/patchers.py

@@ -29,6 +29,9 @@ def __init__(self, read_data):
             self.read_data = read_data
             _mock_open.written = read_data[:0]
 
+        def fileno(self):
+            return 123
+
         def read(self, size=-1):
             if size == -1:
                 ret = self.read_data

tests/test_adb_device.py

@@ -22,6 +22,145 @@
 _LOGGER.setLevel(logging.DEBUG)
 _LOGGER.addHandler(logging.StreamHandler(sys.stdout))
 
+_LOGGER2 = logging.getLogger(__name__)
+_LOGGER2.setLevel(logging.DEBUG)
+_LOGGER2.addHandler(logging.StreamHandler(sys.stdout))
+
+
+def parse_module(infile):
+    """Code for converting `AdbDevice` methods into `AdbDeviceTest` methods (see below)."""
+    with open(infile) as f:
+         for line in f.readlines():
+             if line.strip().startswith("def "):
+                 parse_function(line)
+                 print()
+
+def parse_function(line):
+    """Code for converting an `AdbDevice` method into an `AdbDeviceTest` method (see below)."""
+    name = line.split("(")[0].split()[-1]
+    args = line.split("(")[1].split(")")[0]
+    args_list = args.split(",")
+    arg_names = [arg.split("=")[0].strip() for arg in args_list]
+    args_format = ", ".join(["{}" for _ in arg_names[1:]])
+    args_str = ", ".join(arg_names[1:])
+    print("    {}".format(line.strip()))
+    print("        _LOGGER2.info(\"AdbDevice.{}({})\".format({}))".format(name, args_format, args_str))
+    print("        return AdbDevice.{}(self, {})".format(name, args_str))
+
+
+class AdbDeviceTest(AdbDevice):
+    def __init__(self, transport, banner=None):
+        _LOGGER2.info("AdbDevice.__init__({}, {})".format(transport, banner))
+        return AdbDevice.__init__(self, transport, banner)
+
+    def close(self):
+        _LOGGER2.info("AdbDevice.close()".format())
+        return AdbDevice.close(self)
+
+    def connect(self, rsa_keys=None, transport_timeout_s=None, auth_timeout_s=constants.DEFAULT_AUTH_TIMEOUT_S, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, auth_callback=None):
+        _LOGGER2.info("AdbDevice.connect({}, {}, {}, {}, {})".format(rsa_keys, transport_timeout_s, auth_timeout_s, read_timeout_s, auth_callback))
+        return AdbDevice.connect(self, rsa_keys, transport_timeout_s, auth_timeout_s, read_timeout_s, auth_callback)
+
+    def _service(self, service, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None, decode=True):
+        _LOGGER2.info("AdbDevice._service({}, {}, {}, {}, {}, {})".format(service, command, transport_timeout_s, read_timeout_s, timeout_s, decode))
+        return AdbDevice._service(self, service, command, transport_timeout_s, read_timeout_s, timeout_s, decode)
+
+    def _streaming_service(self, service, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, decode=True):
+        _LOGGER2.info("AdbDevice._streaming_service({}, {}, {}, {}, {})".format(service, command, transport_timeout_s, read_timeout_s, decode))
+        return AdbDevice._streaming_service(self, service, command, transport_timeout_s, read_timeout_s, decode)
+
+    def root(self, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None):
+        _LOGGER2.info("AdbDevice.root({}, {}, {})".format(transport_timeout_s, read_timeout_s, timeout_s))
+        return AdbDevice.root(self, transport_timeout_s, read_timeout_s, timeout_s)
+
+    def shell(self, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None, decode=True):
+        _LOGGER2.info("AdbDevice.shell({}, {}, {}, {}, {})".format(command, transport_timeout_s, read_timeout_s, timeout_s, decode))
+        return AdbDevice.shell(self, command, transport_timeout_s, read_timeout_s, timeout_s, decode)
+
+    def streaming_shell(self, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, decode=True):
+        _LOGGER2.info("AdbDevice.streaming_shell({}, {}, {}, {})".format(command, transport_timeout_s, read_timeout_s, decode))
+        return AdbDevice.streaming_shell(self, command, transport_timeout_s, read_timeout_s, decode)
+
+    def list(self, device_path, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
+        _LOGGER2.info("AdbDevice.list('{}', {}, {})".format(device_path, transport_timeout_s, read_timeout_s))
+        return AdbDevice.list(self, device_path, transport_timeout_s, read_timeout_s)
+
+    def pull(self, device_path, local_path, progress_callback=None, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
+        _LOGGER2.info("AdbDevice.pull('{}', '{}', {}, {}, {})".format(device_path, local_path, progress_callback, transport_timeout_s, read_timeout_s))
+        return AdbDevice.pull(self, device_path, local_path, progress_callback, transport_timeout_s, read_timeout_s)
+
+    def _pull(self, device_path, stream, progress_callback, adb_info, filesync_info):
+        _LOGGER2.info("AdbDevice._pull('{}', {}, {}, {}, {})".format(device_path, stream, progress_callback, adb_info, filesync_info))
+        return AdbDevice._pull(self, device_path, stream, progress_callback, adb_info, filesync_info)
+
+    def push(self, local_path, device_path, st_mode=constants.DEFAULT_PUSH_MODE, mtime=0, progress_callback=None, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
+        _LOGGER2.info("AdbDevice.push('{}', '{}', {}, {}, {}, {}, {})".format(local_path, device_path, st_mode, mtime, progress_callback, transport_timeout_s, read_timeout_s))
+        return AdbDevice.push(self, local_path, device_path, st_mode, mtime, progress_callback, transport_timeout_s, read_timeout_s)
+
+    def _push(self, stream, device_path, st_mode, mtime, progress_callback, adb_info, filesync_info):
+        _LOGGER2.info("AdbDevice._push({}, '{}', {}, {}, {}, {}, {})".format(stream, device_path, st_mode, mtime, progress_callback, adb_info, filesync_info))
+        return AdbDevice._push(self, stream, device_path, st_mode, mtime, progress_callback, adb_info, filesync_info)
+
+    def stat(self, device_path, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
+        _LOGGER2.info("AdbDevice.stat('{}', {}, {})".format(device_path, transport_timeout_s, read_timeout_s))
+        return AdbDevice.stat(self, device_path, transport_timeout_s, read_timeout_s)
+
+    def _close(self, adb_info):
+        _LOGGER2.info("AdbDevice._close({})".format(adb_info))
+        return AdbDevice._close(self, adb_info)
+
+    def _okay(self, adb_info):
+        _LOGGER2.info("AdbDevice._okay({})".format(adb_info))
+        return AdbDevice._okay(self, adb_info)
+
+    def _open(self, destination, adb_info):
+        _LOGGER2.info("AdbDevice._open({}, {})".format(destination, adb_info))
+        return AdbDevice._open(self, destination, adb_info)
+
+    def _read(self, expected_cmds, adb_info):
+        _LOGGER2.info("AdbDevice._read({}, {})".format(expected_cmds, adb_info))
+        return AdbDevice._read(self, expected_cmds, adb_info)
+
+    def _read_until(self, expected_cmds, adb_info):
+        _LOGGER2.info("AdbDevice._read_until({}, {})".format(expected_cmds, adb_info))
+        return AdbDevice._read_until(self, expected_cmds, adb_info)
+
+    def _read_until_close(self, adb_info):
+        _LOGGER2.info("AdbDevice._read_until_close({})".format(adb_info))
+        return AdbDevice._read_until_close(self, adb_info)
+
+    def _send(self, msg, adb_info):
+        _LOGGER2.info("AdbDevice._send({}, {})".format(msg, adb_info))
+        return AdbDevice._send(self, msg, adb_info)
+
+    def _streaming_command(self, service, command, adb_info):
+        _LOGGER2.info("AdbDevice._streaming_command({}, {}, {})".format(service, command, adb_info))
+        return AdbDevice._streaming_command(self, service, command, adb_info)
+
+    def _write(self, data, adb_info):
+        _LOGGER2.info("AdbDevice._write({}, {})".format(data, adb_info))
+        return AdbDevice._write(self, data, adb_info)
+
+    def _filesync_flush(self, adb_info, filesync_info):
+        _LOGGER2.info("AdbDevice._filesync_flush({}, {})".format(adb_info, filesync_info))
+        return AdbDevice._filesync_flush(self, adb_info, filesync_info)
+
+    def _filesync_read(self, expected_ids, adb_info, filesync_info, read_data=True):
+        _LOGGER2.info("AdbDevice._filesync_read({}, {}, {}, {})".format(expected_ids, adb_info, filesync_info, read_data))
+        return AdbDevice._filesync_read(self, expected_ids, adb_info, filesync_info, read_data)
+
+    def _filesync_read_buffered(self, size, adb_info, filesync_info):
+        _LOGGER2.info("AdbDevice._filesync_read_buffered({}, {}, {})".format(size, adb_info, filesync_info))
+        return AdbDevice._filesync_read_buffered(self, size, adb_info, filesync_info)
+
+    def _filesync_read_until(self, expected_ids, finish_ids, adb_info, filesync_info):
+        _LOGGER2.info("AdbDevice._filesync_read_until({}, {}, {}, {})".format(expected_ids, finish_ids, adb_info, filesync_info))
+        return AdbDevice._filesync_read_until(self, expected_ids, finish_ids, adb_info, filesync_info)
+
+    def _filesync_send(self, command_id, adb_info, filesync_info, data=b'', size=None):
+        _LOGGER2.info("AdbDevice._filesync_send({}, {}, {}, {}, {})".format(command_id, adb_info, filesync_info, data, size))
+        return AdbDevice._filesync_send(self, command_id, adb_info, filesync_info, data, size)
+
 
 def to_int(cmd):
     return sum(c << (i * 8) for i, c in enumerate(bytearray(cmd)))
@@ -571,6 +710,53 @@ def test_push_file(self):
             self.device.push('TEST_FILE', '/data', mtime=mtime)
             self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)
 
+    def _test_push_issue113(self):
+        # pytest tests/test_adb_device.py::TestAdbDevice::test_push_issue113 --log-cli-level=INFO
+        def push_progress_callback(device_path, bytes_written, total_bytes):
+            _LOGGER2.warning(f"ADB Push-Progress: {device_path} bytes_written:{bytes_written} total_bytes:{total_bytes}")
+
+        self.device = AdbDeviceTest(transport=patchers.FakeTcpTransport('host', 5555))
+        self.device._transport._bulk_read = b''.join(patchers.BULK_READ_LIST)
+        self.device._transport._bulk_read = b''.join([b'CNXN\x01\x00\x00\x01\x00\x00\x10\x00\x10\x01\x00\x00\xa1f\x00\x00\xbc\xb1\xa7\xb1',
+                                                      b'device::ro.product.name=sdk_gphone_x86_64_arm64;ro.product.model=sdk_gphone_x86_64_arm64;ro.product.device=generic_x86_64_arm64;features=sendrecv_v2_brotli,remount_shell,sendrecv_v2,abb_exec,fixed_push_mkdir,fixed_push_symlink_timestamp,abb,shell_v2,cmd,ls_v2,apex,stat_v2',
+                                                      b'OKAYj\x06\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb0\xb4\xbe\xa6',
+                                                      b'WRTEj\x06\x00\x00\x01\x00\x00\x00 \x00\x00\x00\x8c\x0b\x00\x00\xa8\xad\xab\xba',
+                                                      b'adbd is already running as root\n',
+                                                      b'CLSEj\x06\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xbc\xb3\xac\xba'])
+
+        self.assertTrue(self.device.connect())
+        self.device.root()
+        self.assertEqual(self.device._transport._bulk_read, b'')
+
+        self.device._transport._bulk_write = b''
+
+        mtime = 100
+        filedata = b'Ohayou sekai.\nGood morning world!'
+
+        # Provide the `bulk_read` return values
+        self.device._transport._bulk_read = b''.join([b'CLSEj\x06\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xbc\xb3\xac\xba',
+                                                      b'OKAYk\x06\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb0\xb4\xbe\xa6'])
+        #self.device._transport._bulk_read = join_messages(AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
+        #                                                  AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b''),
+        #                                                  AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=FileSyncMessage(constants.OKAY).pack()),
+        #                                                  AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))
+
+        # Expected `bulk_write` values
+        expected_bulk_write = join_messages(AdbMessage(command=constants.OPEN, arg0=1, arg1=0, data=b'sync:\x00'),
+                                            AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=join_messages(FileSyncMessage(command=constants.SEND, data=b'/data,33272'),
+                                                                                                                  FileSyncMessage(command=constants.DATA, data=filedata),
+                                                                                                                  FileSyncMessage(command=constants.DONE, arg0=mtime, data=b''))),
+                                            AdbMessage(command=constants.OKAY, arg0=1, arg1=1),
+                                            AdbMessage(command=constants.CLSE, arg0=1, arg1=1, data=b''))
+
+        class StSize:
+            def __init__(self):
+                self.st_size = -1
+
+        with patch('adb_shell.adb_device.open', patchers.mock_open(read_data=filedata)), patch('os.fstat', return_value=StSize()):
+            self.device.push('TEST_FILE', '/data', mtime=mtime, progress_callback=push_progress_callback)
+            self.assertEqual(expected_bulk_write, self.device._transport._bulk_write)
+
     def test_push_file_mtime0(self):
         self.assertTrue(self.device.connect())
         self.device._transport._bulk_write = b''