# 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 io
import mock
import subprocess

from collections import OrderedDict
from future.tests.base import unittest
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 = 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_process(self, mock_psutil_process, mock_popen):
    # the created proc object in adb._exec_cmd()
    mock_proc = mock.Mock()
    mock_popen.return_value = mock_proc

    # the created process object in adb._exec_cmd()
    mock_psutil_process.return_value = mock.Mock()

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

  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_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 = adb.psutil.TimeoutExpired('Timed out')
    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 = adb.psutil.TimeoutExpired('Timed out')
    with self.assertRaisesRegex(
        adb.AdbTimeoutError, 'Timed out executing command "adb '
        'fake-cmd" after 0.01s.') as context:
      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'))

  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__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, 2)
    mock_sleep.assert_called_with(10)

  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()