path: root/tests/mobly/controllers/android_device_lib/adb_test.py

# Copyright 2017 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import collections
import io
import subprocess
import unittest
from unittest import mock

from mobly.controllers.android_device_lib import adb

# Mock parameters for instrumentation.
MOCK_INSTRUMENTATION_PACKAGE = 'com.my.instrumentation.tests'
MOCK_INSTRUMENTATION_RUNNER = 'com.my.instrumentation.runner'
MOCK_INSTRUMENTATION_OPTIONS = collections.OrderedDict([
    ('option1', 'value1'),
    ('option2', 'value2'),
])
# Mock android instrumentation commands.
MOCK_BASIC_INSTRUMENTATION_COMMAND = ('am instrument -r -w  com.my'
                                      '.instrumentation.tests/com.android'
                                      '.common.support.test.runner'
                                      '.AndroidJUnitRunner')
MOCK_RUNNER_INSTRUMENTATION_COMMAND = ('am instrument -r -w  com.my'
                                       '.instrumentation.tests/com.my'
                                       '.instrumentation.runner')
MOCK_OPTIONS_INSTRUMENTATION_COMMAND = ('am instrument -r -w -e option1 value1'
                                        ' -e option2 value2 com.my'
                                        '.instrumentation.tests/com.android'
                                        '.common.support.test.runner'
                                        '.AndroidJUnitRunner')

# Mock root command outputs.
MOCK_ROOT_SUCCESS_OUTPUT = 'adbd is already running as root'
MOCK_ROOT_ERROR_OUTPUT = (
    'adb: unable to connect for root: closed'.encode('utf-8'))

# Mock Shell Command
MOCK_SHELL_COMMAND = 'ls'
MOCK_COMMAND_OUTPUT = '/system/bin/ls'.encode('utf-8')
MOCK_DEFAULT_STDOUT = 'out'
MOCK_DEFAULT_STDERR = 'err'
MOCK_DEFAULT_COMMAND_OUTPUT = MOCK_DEFAULT_STDOUT.encode('utf-8')
MOCK_ADB_SHELL_COMMAND_CHECK = 'adb shell command -v ls'


class AdbTest(unittest.TestCase):
  """Unit tests for mobly.controllers.android_device_lib.adb."""

  def _mock_execute_and_process_stdout_process(self, mock_popen):
    # the created proc object in adb._execute_and_process_stdout()
    mock_proc = mock.Mock()
    mock_popen.return_value = mock_proc

    mock_popen.return_value.stdout.readline.side_effect = ['']

    mock_proc.communicate = mock.Mock(
        return_value=('', MOCK_DEFAULT_STDERR.encode('utf-8')))
    mock_proc.returncode = 0
    return mock_popen

  @mock.patch('mobly.utils.run_command')
  def test_is_adb_available(self, mock_run_command):
    mock_run_command.return_value = (0, '/usr/local/bin/adb\n'.encode('utf-8'),
                                     ''.encode('utf-8'))
    self.assertTrue(adb.is_adb_available())

  @mock.patch('mobly.utils.run_command')
  def test_is_adb_available_negative(self, mock_run_command):
    mock_run_command.return_value = (0, ''.encode('utf-8'), ''.encode('utf-8'))
    self.assertFalse(adb.is_adb_available())

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_no_timeout_success(self, mock_run_command):
    mock_run_command.return_value = (0, MOCK_DEFAULT_STDOUT.encode('utf-8'),
                                     MOCK_DEFAULT_STDERR.encode('utf-8'))
    out = adb.AdbProxy()._exec_cmd(['fake_cmd'],
                                   shell=False,
                                   timeout=None,
                                   stderr=None)
    self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8'))
    mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=None)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_error_with_serial(self, mock_run_command):
    # Return 1 for retcode for error.
    mock_run_command.return_value = (1, MOCK_DEFAULT_STDOUT.encode('utf-8'),
                                     MOCK_DEFAULT_STDERR.encode('utf-8'))
    mock_serial = 'ABCD1234'
    with self.assertRaisesRegex(adb.AdbError,
                                'Error executing adb cmd .*') as context:
      adb.AdbProxy(mock_serial).fake_cmd()
    self.assertEqual(context.exception.serial, mock_serial)
    self.assertIn(mock_serial, context.exception.cmd)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_error_without_serial(self, mock_run_command):
    # Return 1 for retcode for error.
    mock_run_command.return_value = (1, MOCK_DEFAULT_STDOUT.encode('utf-8'),
                                     MOCK_DEFAULT_STDERR.encode('utf-8'))
    with self.assertRaisesRegex(adb.AdbError,
                                'Error executing adb cmd .*') as context:
      adb.AdbProxy()._exec_cmd(['fake_cmd'],
                               shell=False,
                               timeout=None,
                               stderr=None)
    self.assertFalse(context.exception.serial)
    mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=None)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_with_timeout_success(self, mock_run_command):
    mock_run_command.return_value = (0, MOCK_DEFAULT_STDOUT.encode('utf-8'),
                                     MOCK_DEFAULT_STDERR.encode('utf-8'))

    out = adb.AdbProxy()._exec_cmd(['fake_cmd'],
                                   shell=False,
                                   timeout=1,
                                   stderr=None)
    self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8'))
    mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=1)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_timed_out(self, mock_run_command):
    mock_run_command.side_effect = subprocess.TimeoutExpired(cmd='mock_command',
                                                             timeout=0.01)
    mock_serial = '1234Abcd'

    with self.assertRaisesRegex(
        adb.AdbTimeoutError, 'Timed out executing command "adb -s '
        '1234Abcd fake-cmd" after 0.01s.') as context:
      adb.AdbProxy(mock_serial).fake_cmd(timeout=0.01)

    self.assertEqual(context.exception.serial, mock_serial)
    self.assertIn(mock_serial, context.exception.cmd)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_timed_out_without_serial(self, mock_run_command):
    mock_run_command.side_effect = subprocess.TimeoutExpired(cmd='mock_command',
                                                             timeout=0.01)

    with self.assertRaisesRegex(
        adb.AdbTimeoutError,
        'Timed out executing command "adb fake-cmd" after 0.01s.'):
      adb.AdbProxy().fake_cmd(timeout=0.01)

  def test_exec_cmd_with_negative_timeout_value(self):
    with self.assertRaisesRegex(ValueError,
                                'Timeout is not a positive value: -1'):
      adb.AdbProxy()._exec_cmd(['fake_cmd'],
                               shell=False,
                               timeout=-1,
                               stderr=None)

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_reads_stdout(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '']
    mock_handler = mock.MagicMock()

    err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                     shell=False,
                                                     handler=mock_handler)
    self.assertEqual(mock_handler.call_count, 2)
    mock_handler.assert_any_call('1')
    mock_handler.assert_any_call('2')

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_reads_unexpected_stdout(self, mock_popen):
    unexpected_stdout = MOCK_DEFAULT_STDOUT.encode('utf-8')

    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_handler = mock.MagicMock()
    mock_popen.return_value.communicate = mock.Mock(
        return_value=(unexpected_stdout, MOCK_DEFAULT_STDERR.encode('utf-8')))

    err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                     shell=False,
                                                     handler=mock_handler)
    self.assertEqual(mock_handler.call_count, 1)
    mock_handler.assert_called_with(unexpected_stdout)

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  @mock.patch('logging.debug')
  def test_execute_and_process_stdout_logs_cmd(self, mock_debug_logger,
                                               mock_popen):
    raw_expected_stdout = ''
    expected_stdout = '[elided, processed via handler]'
    expected_stderr = MOCK_DEFAULT_STDERR.encode('utf-8')
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.communicate = mock.Mock(
        return_value=(raw_expected_stdout, expected_stderr))

    err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                     shell=False,
                                                     handler=mock.MagicMock())
    mock_debug_logger.assert_called_with(
        'cmd: %s, stdout: %s, stderr: %s, ret: %s', 'fake_cmd', expected_stdout,
        expected_stderr, 0)

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  @mock.patch('logging.debug')
  def test_execute_and_process_stdout_logs_cmd_with_unexpected_stdout(
      self, mock_debug_logger, mock_popen):
    raw_expected_stdout = MOCK_DEFAULT_STDOUT.encode('utf-8')
    expected_stdout = '[unexpected stdout] %s' % raw_expected_stdout
    expected_stderr = MOCK_DEFAULT_STDERR.encode('utf-8')

    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.communicate = mock.Mock(
        return_value=(raw_expected_stdout, expected_stderr))

    err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                     shell=False,
                                                     handler=mock.MagicMock())
    mock_debug_logger.assert_called_with(
        'cmd: %s, stdout: %s, stderr: %s, ret: %s', 'fake_cmd', expected_stdout,
        expected_stderr, 0)

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_despite_cmd_exits(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.poll.side_effect = [None, 0]
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', '']
    mock_handler = mock.MagicMock()

    err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                     shell=False,
                                                     handler=mock_handler)

    self.assertEqual(mock_handler.call_count, 3)
    mock_handler.assert_any_call('1')
    mock_handler.assert_any_call('2')
    mock_handler.assert_any_call('3')

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_when_cmd_eof(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', '']
    mock_handler = mock.MagicMock()

    err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                     shell=False,
                                                     handler=mock_handler)

    self.assertEqual(mock_handler.call_count, 3)
    mock_handler.assert_any_call('1')
    mock_handler.assert_any_call('2')
    mock_handler.assert_any_call('3')

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_returns_stderr(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)

    err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                     shell=False,
                                                     handler=mock.MagicMock())
    self.assertEqual(MOCK_DEFAULT_STDERR, err.decode('utf-8'))

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_raises_adb_error(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.returncode = 1

    with self.assertRaisesRegex(adb.AdbError, 'Error executing adb cmd .*'):
      err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                       shell=False,
                                                       handler=mock.MagicMock())

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_when_handler_crash(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', '']
    mock_handler = mock.MagicMock()
    mock_handler.side_effect = ['', TypeError('fake crash'), '', '']

    with self.assertRaisesRegex(TypeError, 'fake crash'):
      err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'],
                                                       shell=False,
                                                       handler=mock_handler)

    mock_popen.return_value.communicate.assert_called_once_with()

  def test_construct_adb_cmd(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', 'arg1', shell=False)
    self.assertEqual(adb_cmd, ['adb', 'shell', 'arg1'])

  def test_construct_adb_cmd_with_one_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell ls /asdafsfd/asdf-asfd/asa', [], shell=False)
    self.assertEqual(adb_cmd, ['adb', 'shell ls /asdafsfd/asdf-asfd/asa'])

  def test_construct_adb_cmd_with_one_arg_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell',
                                                'ls /asdafsfd/asdf-asfd/asa',
                                                shell=False)
    self.assertEqual(adb_cmd, ['adb', 'shell', 'ls /asdafsfd/asdf-asfd/asa'])

  def test_construct_adb_cmd_with_one_arg_command_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell',
                                                ['ls /asdafsfd/asdf-asfd/asa'],
                                                shell=False)
    self.assertEqual(adb_cmd, ['adb', 'shell', 'ls /asdafsfd/asdf-asfd/asa'])

  def test_construct_adb_cmd_with_special_characters(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell',
                                                ['a b', '"blah"', r'\/\/'],
                                                shell=False)
    self.assertEqual(adb_cmd, ['adb', 'shell', 'a b', '"blah"', r"\/\/"])

  def test_construct_adb_cmd_with_serial(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell',
                                                       'arg1',
                                                       shell=False)
    self.assertEqual(adb_cmd, ['adb', '-s', '12345', 'shell', 'arg1'])

  def test_construct_adb_cmd_with_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', ['arg1', 'arg2'],
                                                shell=False)
    self.assertEqual(adb_cmd, ['adb', 'shell', 'arg1', 'arg2'])

  def test_construct_adb_cmd_with_serial_with_list(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell',
                                                       ['arg1', 'arg2'],
                                                       shell=False)
    self.assertEqual(adb_cmd, ['adb', '-s', '12345', 'shell', 'arg1', 'arg2'])

  def test_construct_adb_cmd_with_shell_true(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell',
                                                'arg1 arg2',
                                                shell=True)
    self.assertEqual(adb_cmd, '"adb" shell arg1 arg2')

  def test_construct_adb_cmd_with_shell_true_with_one_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell ls /asdafsfd/asdf-asfd/asa', [], shell=True)
    self.assertEqual(adb_cmd, '"adb" shell ls /asdafsfd/asdf-asfd/asa ')

  def test_construct_adb_cmd_with_shell_true_with_one_arg_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell',
                                                'ls /asdafsfd/asdf-asfd/asa',
                                                shell=True)
    self.assertEqual(adb_cmd, '"adb" shell ls /asdafsfd/asdf-asfd/asa')

  def test_construct_adb_cmd_with_shell_true_with_one_arg_command_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell',
                                                ['ls /asdafsfd/asdf-asfd/asa'],
                                                shell=True)
    self.assertEqual(adb_cmd, '"adb" shell \'ls /asdafsfd/asdf-asfd/asa\'')

  def test_construct_adb_cmd_with_shell_true_with_auto_quotes(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell',
                                                ['a b', '"blah"', r'\/\/'],
                                                shell=True)
    self.assertEqual(adb_cmd, '"adb" shell \'a b\' \'"blah"\' \'\\/\\/\'')

  def test_construct_adb_cmd_with_shell_true_with_serial(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell',
                                                       'arg1 arg2',
                                                       shell=True)
    self.assertEqual(adb_cmd, '"adb" -s "12345" shell arg1 arg2')

  def test_construct_adb_cmd_with_shell_true_with_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', ['arg1', 'arg2'],
                                                shell=True)
    self.assertEqual(adb_cmd, '"adb" shell arg1 arg2')

  def test_construct_adb_cmd_with_shell_true_with_serial_with_list(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell',
                                                       ['arg1', 'arg2'],
                                                       shell=True)
    self.assertEqual(adb_cmd, '"adb" -s "12345" shell arg1 arg2')

  def test_exec_adb_cmd(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      adb.AdbProxy().shell(['arg1', 'arg2'])
      mock_exec_cmd.assert_called_once_with(['adb', 'shell', 'arg1', 'arg2'],
                                            shell=False,
                                            timeout=None,
                                            stderr=None)

  def test_exec_adb_cmd_with_shell_true(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      adb.AdbProxy().shell('arg1 arg2', shell=True)
      mock_exec_cmd.assert_called_once_with('"adb" shell arg1 arg2',
                                            shell=True,
                                            timeout=None,
                                            stderr=None)

  def test_exec_adb_cmd_formats_command(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      with mock.patch.object(adb.AdbProxy,
                             '_construct_adb_cmd') as mock_construct_adb_cmd:
        mock_adb_cmd = mock.MagicMock()
        mock_adb_args = mock.MagicMock()
        mock_construct_adb_cmd.return_value = mock_adb_cmd
        mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT

        adb.AdbProxy().shell(mock_adb_args)
        mock_construct_adb_cmd.assert_called_once_with('shell',
                                                       mock_adb_args,
                                                       shell=False)
        mock_exec_cmd.assert_called_once_with(mock_adb_cmd,
                                              shell=False,
                                              timeout=None,
                                              stderr=None)

  def test_exec_adb_cmd_formats_command_with_shell_true(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      with mock.patch.object(adb.AdbProxy,
                             '_construct_adb_cmd') as mock_construct_adb_cmd:
        mock_adb_cmd = mock.MagicMock()
        mock_adb_args = mock.MagicMock()
        mock_construct_adb_cmd.return_value = mock_adb_cmd

        adb.AdbProxy().shell(mock_adb_args, shell=True)
        mock_construct_adb_cmd.assert_called_once_with('shell',
                                                       mock_adb_args,
                                                       shell=True)
        mock_exec_cmd.assert_called_once_with(mock_adb_cmd,
                                              shell=True,
                                              timeout=None,
                                              stderr=None)

  def test_execute_adb_and_process_stdout_formats_command(self):
    with mock.patch.object(
        adb.AdbProxy,
        '_execute_and_process_stdout') as mock_execute_and_process_stdout:
      with mock.patch.object(adb.AdbProxy,
                             '_construct_adb_cmd') as mock_construct_adb_cmd:
        mock_adb_cmd = mock.MagicMock()
        mock_adb_args = mock.MagicMock()
        mock_handler = mock.MagicMock()
        mock_construct_adb_cmd.return_value = mock_adb_cmd

        adb.AdbProxy()._execute_adb_and_process_stdout('shell',
                                                       mock_adb_args,
                                                       shell=False,
                                                       handler=mock_handler)
        mock_construct_adb_cmd.assert_called_once_with('shell',
                                                       mock_adb_args,
                                                       shell=False)
        mock_execute_and_process_stdout.assert_called_once_with(
            mock_adb_cmd, shell=False, handler=mock_handler)

  @mock.patch('mobly.utils.run_command')
  def test_exec_adb_cmd_with_stderr_pipe(self, mock_run_command):
    mock_run_command.return_value = (0, MOCK_DEFAULT_STDOUT.encode('utf-8'),
                                     MOCK_DEFAULT_STDERR.encode('utf-8'))
    stderr_redirect = io.BytesIO()
    out = adb.AdbProxy().shell('arg1 arg2', shell=True, stderr=stderr_redirect)
    self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8'))
    self.assertEqual(MOCK_DEFAULT_STDERR,
                     stderr_redirect.getvalue().decode('utf-8'))

  @mock.patch('mobly.utils.run_command')
  def test_connect_success(self, mock_run_command):
    mock_address = 'localhost:1234'
    mock_run_command.return_value = (
        0, f'connected to {mock_address}'.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'))

    out = adb.AdbProxy().connect(mock_address)
    self.assertEqual('connected to localhost:1234', out.decode('utf-8'))

  @mock.patch('mobly.utils.run_command')
  def test_connect_already_connected(self, mock_run_command):
    mock_address = 'localhost:1234'
    mock_run_command.return_value = (
        0, f'already connected to {mock_address}'.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'))

    out = adb.AdbProxy().connect(mock_address)
    self.assertEqual('already connected to localhost:1234', out.decode('utf-8'))

  @mock.patch('mobly.utils.run_command')
  def test_connect_fail(self, mock_run_command):
    mock_address = 'localhost:1234'
    mock_run_command.return_value = (0, 'Connection refused\n'.encode('utf-8'),
                                     MOCK_DEFAULT_STDERR.encode('utf-8'))

    with self.assertRaisesRegex(
        adb.AdbError, 'Error executing adb cmd "connect localhost:1234".'):
      out = adb.AdbProxy().connect(mock_address)

  def test_getprop(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = b'blah'
      self.assertEqual(adb.AdbProxy().getprop('haha'), 'blah')
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'getprop', 'haha'],
          shell=False,
          stderr=None,
          timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC)

  def test_getprop_custom_timeout(self):
    timeout_s = adb.DEFAULT_GETPROP_TIMEOUT_SEC * 2
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = b'blah'
      self.assertEqual(adb.AdbProxy().getprop('haha', timeout=timeout_s),
                       'blah')
      mock_exec_cmd.assert_called_once_with(['adb', 'shell', 'getprop', 'haha'],
                                            shell=False,
                                            stderr=None,
                                            timeout=timeout_s)

  def test__parse_getprop_output_special_values(self):
    mock_adb_output = (
        b'[selinux.restorecon_recursive]: [/data/misc_ce/10]\n'
        b'[selinux.abc]: [key: value]\n'  # "key: value" as value
        b'[persist.sys.boot.reason.history]: [reboot,adb,1558549857\n'
        b'reboot,factory_reset,1558483886\n'  # multi-line value
        b'reboot,1558483823]\n'
        b'[persist.something]: [haha\n'
        b']\n'
        b'[[wrapped.key]]: [[wrapped value]]\n'
        b'[persist.byte]: [J\xaa\x8bb\xab\x9dP\x0f]\n'  # non-decodable
    )
    parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output)
    expected_output = {
        'persist.sys.boot.reason.history':
            ('reboot,adb,1558549857\nreboot,factory_reset,1558483886\n'
             'reboot,1558483823'),
        'selinux.abc': 'key: value',
        'persist.something': 'haha\n',
        'selinux.restorecon_recursive': '/data/misc_ce/10',
        '[wrapped.key]': '[wrapped value]',
        'persist.byte': 'JbP\x0f',
    }
    self.assertEqual(parsed_props, expected_output)

  def test__parse_getprop_output_malformat_output(self):
    mock_adb_output = (
        b'[selinux.restorecon_recursive][/data/misc_ce/10]\n'  # Malformat
        b'[persist.sys.boot.reason]: [reboot,adb,1558549857]\n'
        b'[persist.something]: [haha]\n')
    parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output)
    expected_output = {
        'persist.sys.boot.reason': 'reboot,adb,1558549857',
        'persist.something': 'haha'
    }
    self.assertEqual(parsed_props, expected_output)

  def test__parse_getprop_output_special_line_separator(self):
    mock_adb_output = (
        b'[selinux.restorecon_recursive][/data/misc_ce/10]\r\n'  # Malformat
        b'[persist.sys.boot.reason]: [reboot,adb,1558549857]\r\n'
        b'[persist.something]: [haha]\r\n')
    parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output)
    expected_output = {
        'persist.sys.boot.reason': 'reboot,adb,1558549857',
        'persist.something': 'haha'
    }
    self.assertEqual(parsed_props, expected_output)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  def test_getprops(self, mock_sleep):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = (
          b'\n[sendbug.preferred.domain]: [google.com]\n'
          b'[sys.uidcpupower]: []\n'
          b'[sys.wifitracing.started]: [1]\n'
          b'[telephony.lteOnCdmaDevice]: [1]\n\n')
      actual_output = adb.AdbProxy().getprops([
          'sys.wifitracing.started',  # "numeric" value
          'sys.uidcpupower',  # empty value
          'sendbug.preferred.domain',  # string value
          'nonExistentProp'
      ])
      self.assertEqual(
          actual_output, {
              'sys.wifitracing.started': '1',
              'sys.uidcpupower': '',
              'sendbug.preferred.domain': 'google.com'
          })
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'getprop'],
          shell=False,
          stderr=None,
          timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC)
      mock_sleep.assert_not_called()

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  def test_getprops_when_empty_string_randomly_returned(self, mock_sleep):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.side_effect = [
          b'', (b'\n[ro.build.id]: [AB42]\n'
                b'[ro.build.type]: [userdebug]\n\n')
      ]
      actual_output = adb.AdbProxy().getprops(['ro.build.id'])
      self.assertEqual(actual_output, {
          'ro.build.id': 'AB42',
      })
      self.assertEqual(mock_exec_cmd.call_count, 2)
      mock_exec_cmd.assert_called_with(['adb', 'shell', 'getprop'],
                                       shell=False,
                                       stderr=None,
                                       timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC)
      self.assertEqual(mock_sleep.call_count, 1)
      mock_sleep.assert_called_with(1)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  def test_getprops_when_empty_string_always_returned(self, mock_sleep):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = b''
      actual_output = adb.AdbProxy().getprops(['ro.build.id'])
      self.assertEqual(actual_output, {})
      self.assertEqual(mock_exec_cmd.call_count, 3)
      mock_exec_cmd.assert_called_with(['adb', 'shell', 'getprop'],
                                       shell=False,
                                       stderr=None,
                                       timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC)
      self.assertEqual(mock_sleep.call_count, 2)
      mock_sleep.assert_called_with(1)

  def test_forward(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      adb.AdbProxy().forward(MOCK_SHELL_COMMAND)

  def test_instrument_without_parameters(self):
    """Verifies the AndroidDevice object's instrument command is correct in
    the basic case.
    """
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      output = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE)
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND],
          shell=False,
          timeout=None,
          stderr=None)
      self.assertEqual(output, mock_exec_cmd.return_value)

  def test_instrument_with_runner(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a runner specified.
    """
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      stdout = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE,
                                         runner=MOCK_INSTRUMENTATION_RUNNER)
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND],
          shell=False,
          timeout=None,
          stderr=None)
      self.assertEqual(stdout, mock_exec_cmd.return_value)

  def test_instrument_with_options(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with options.
    """
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      stdout = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE,
                                         options=MOCK_INSTRUMENTATION_OPTIONS)
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND],
          shell=False,
          timeout=None,
          stderr=None)
      self.assertEqual(stdout, mock_exec_cmd.return_value)

  def test_instrument_with_handler(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a handler passed in.
    """

    def mock_handler(raw_line):
      pass

    with mock.patch.object(
        adb.AdbProxy,
        '_execute_and_process_stdout') as mock_execute_and_process_stdout:
      stderr = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE,
                                         handler=mock_handler)
      mock_execute_and_process_stdout.assert_called_once_with(
          ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND],
          shell=False,
          handler=mock_handler)
      self.assertEqual(stderr, mock_execute_and_process_stdout.return_value)

  def test_instrument_with_handler_with_runner(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a handler passed in and a runner specified.
    """

    def mock_handler(raw_line):
      pass

    with mock.patch.object(
        adb.AdbProxy,
        '_execute_and_process_stdout') as mock_execute_and_process_stdout:
      stderr = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE,
                                         runner=MOCK_INSTRUMENTATION_RUNNER,
                                         handler=mock_handler)
      mock_execute_and_process_stdout.assert_called_once_with(
          ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND],
          shell=False,
          handler=mock_handler)
      self.assertEqual(stderr, mock_execute_and_process_stdout.return_value)

  def test_instrument_with_handler_with_options(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a handler passed in and options.
    """

    def mock_handler(raw_line):
      pass

    with mock.patch.object(
        adb.AdbProxy,
        '_execute_and_process_stdout') as mock_execute_and_process_stdout:
      stderr = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE,
                                         options=MOCK_INSTRUMENTATION_OPTIONS,
                                         handler=mock_handler)
      mock_execute_and_process_stdout.assert_called_once_with(
          ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND],
          shell=False,
          handler=mock_handler)
      self.assertEqual(stderr, mock_execute_and_process_stdout.return_value)

  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_root_success(self, mock_exec_cmd):
    mock_exec_cmd.return_value = MOCK_ROOT_SUCCESS_OUTPUT
    output = adb.AdbProxy().root()
    mock_exec_cmd.assert_called_once_with(['adb', 'root'],
                                          shell=False,
                                          timeout=None,
                                          stderr=None)
    self.assertEqual(output, MOCK_ROOT_SUCCESS_OUTPUT)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_root_success_with_retry(self, mock_exec_cmd, mock_sleep):
    mock_exec_cmd.side_effect = [
        adb.AdbError('adb root', '', MOCK_ROOT_ERROR_OUTPUT, 1),
        MOCK_ROOT_SUCCESS_OUTPUT
    ]
    output = adb.AdbProxy().root()
    mock_exec_cmd.assert_called_with(['adb', 'root'],
                                     shell=False,
                                     timeout=None,
                                     stderr=None)
    self.assertEqual(output, MOCK_ROOT_SUCCESS_OUTPUT)
    self.assertEqual(mock_sleep.call_count, 1)
    mock_sleep.assert_called_with(10)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_root_raises_adb_error_when_all_retries_failed(
      self, mock_exec_cmd, mock_sleep):
    mock_exec_cmd.side_effect = adb.AdbError('adb root', '',
                                             MOCK_ROOT_ERROR_OUTPUT, 1)
    expected_msg = ('Error executing adb cmd "adb root". '
                    'ret: 1, stdout: , stderr: %s' % MOCK_ROOT_ERROR_OUTPUT)
    with self.assertRaisesRegex(adb.AdbError, expected_msg):
      adb.AdbProxy().root()
      mock_exec_cmd.assert_called_with(['adb', 'root'],
                                       shell=False,
                                       timeout=None,
                                       stderr=None)
    self.assertEqual(mock_sleep.call_count, adb.ADB_ROOT_RETRY_ATTEMPTS - 1)
    mock_sleep.assert_has_calls([mock.call(10), mock.call(20)])

  def test_has_shell_command_called_correctly(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND)
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'command', '-v', MOCK_SHELL_COMMAND],
          shell=False,
          timeout=None,
          stderr=None)

  def test_has_shell_command_with_existing_command(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_COMMAND_OUTPUT
      self.assertTrue(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND))

  def test_has_shell_command_with_missing_command_on_older_devices(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      mock_exec_cmd.side_effect = adb.AdbError(MOCK_ADB_SHELL_COMMAND_CHECK, '',
                                               '', 0)
      self.assertFalse(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND))

  def test_has_shell_command_with_missing_command_on_newer_devices(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      mock_exec_cmd.side_effect = adb.AdbError(MOCK_ADB_SHELL_COMMAND_CHECK, '',
                                               '', 1)
      self.assertFalse(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND))

  @mock.patch.object(adb.AdbProxy, 'getprop')
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_current_user_id_25_and_above(self, mock_exec_cmd, mock_getprop):
    mock_getprop.return_value = b'25'
    mock_exec_cmd.return_value = b'123'
    user_id = adb.AdbProxy().current_user_id
    mock_exec_cmd.assert_called_once_with(
        ['adb', 'shell', 'am', 'get-current-user'],
        shell=False,
        stderr=None,
        timeout=None)
    self.assertEqual(user_id, 123)

  @mock.patch.object(adb.AdbProxy, 'getprop')
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_current_user_id_between_21_and_24(self, mock_exec_cmd, mock_getprop):
    mock_getprop.return_value = b'23'
    mock_exec_cmd.return_value = (b'Users:\n'
                                  b'UserInfo{123:Owner:13} serialNo=0\n'
                                  b'Created: <unknown>\n'
                                  b'Last logged in: +1h22m12s497ms ago\n'
                                  b'UserInfo{456:Owner:14} serialNo=0\n'
                                  b'Created: <unknown>\n'
                                  b'Last logged in: +1h01m12s497ms ago\n')
    user_id = adb.AdbProxy().current_user_id
    mock_exec_cmd.assert_called_once_with(['adb', 'shell', 'dumpsys', 'user'],
                                          shell=False,
                                          stderr=None,
                                          timeout=None)
    self.assertEqual(user_id, 123)


if __name__ == '__main__':
  unittest.main()