#
# Copyright (C) 2024 The Android Open Source Project
#
# 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 unittest
import os
import subprocess
from unittest import mock
from command import ProfilerCommand
from device import AdbDevice

TEST_DEVICE_SERIAL = "test-device-serial"
TEST_DEVICE_SERIAL2 = "test-device-serial2"
TEST_FILE_PATH = "test-file-path"
TEST_STRING_FILE = "test-string-file"
TEST_FAILURE_MSG = "test-failure"
TEST_EXCEPTION = Exception(TEST_FAILURE_MSG)
TEST_USER_ID_1 = 0
TEST_USER_ID_2 = 1
TEST_USER_ID_3 = 2
TEST_PACKAGE_1 = "test-package-1"
TEST_PACKAGE_2 = "test-package-2"
TEST_PROP = "test-prop"
TEST_PROP_VALUE = "test-prop-value"
TEST_PID_OUTPUT = b"8241\n"
BOOT_COMPLETE_OUTPUT = b"1\n"
ANDROID_SDK_VERSION_T = 33

class DeviceUnitTest(unittest.TestCase):

  @staticmethod
  def generate_adb_devices_result(devices, adb_started=True):
    devices = [device.encode('utf-8') for device in devices]
    stdout_string = b'List of devices attached\n'
    if not adb_started:
      stdout_string = (b'* daemon not running; starting now at tcp:1234\n'
                       b'* daemon started successfully\n') + stdout_string
    if len(devices) > 0:
      stdout_string += b'\tdevice\n'.join(devices) + b'\tdevice\n'
      stdout_string += b'\n'
    return subprocess.CompletedProcess(args=['adb', 'devices'], returncode=0,
                                       stdout=stdout_string)

  @staticmethod
  def generate_mock_completed_process(stdout_string=b'\n', stderr_string=b'\n'):
    return mock.create_autospec(subprocess.CompletedProcess, instance=True,
                                stdout=stdout_string, stderr=stderr_string)

  @staticmethod
  def subprocess_output(first_return_value, polling_return_value):
    # Mocking the return value of a call to adb root and the return values of
    # many followup calls to adb devices
    yield first_return_value
    while True:
      yield polling_return_value

  @staticmethod
  def mock_users():
    return mock.create_autospec(subprocess.CompletedProcess, instance=True,
                                stdout=(b'Users:\n\tUserInfo{%d:Driver:813}'
                                        b' running\n\tUserInfo{%d:Driver:412}\n'
                                        % (TEST_USER_ID_1, TEST_USER_ID_2)))

  @staticmethod
  def mock_packages():
    return mock.create_autospec(subprocess.CompletedProcess, instance=True,
                                stdout=(b'package:%b\npackage:%b\n'
                                        % (TEST_PACKAGE_1.encode("utf-8"),
                                           TEST_PACKAGE_2.encode("utf-8"))))

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_adb_devices_returns_devices(self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL,
                                          TEST_DEVICE_SERIAL2]))
    adbDevice = AdbDevice(None)

    devices = adbDevice.get_adb_devices()

    self.assertEqual(len(devices), 2)
    self.assertEqual(devices[0], TEST_DEVICE_SERIAL)
    self.assertEqual(devices[1], TEST_DEVICE_SERIAL2)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_adb_devices_returns_devices_and_adb_not_started(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL,
                                          TEST_DEVICE_SERIAL2], False))
    adbDevice = AdbDevice(None)

    devices = adbDevice.get_adb_devices()

    self.assertEqual(len(devices), 2)
    self.assertEqual(devices[0], TEST_DEVICE_SERIAL)
    self.assertEqual(devices[1], TEST_DEVICE_SERIAL2)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_adb_devices_returns_no_device(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_adb_devices_result([])
    adbDevice = AdbDevice(None)

    devices = adbDevice.get_adb_devices()

    self.assertEqual(devices, [])

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_adb_devices_returns_no_device_and_adb_not_started(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([], False))
    adbDevice = AdbDevice(None)

    devices = adbDevice.get_adb_devices()

    self.assertEqual(devices, [])

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_adb_devices_command_failure_error(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(None)

    with self.assertRaises(Exception) as e:
      adbDevice.get_adb_devices()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_serial_arg_in_devices(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    error = adbDevice.check_device_connection()

    self.assertEqual(error, None)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_serial_arg_not_in_devices_error(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
    invalid_device_serial = "invalid-device-serial"
    adbDevice = AdbDevice(invalid_device_serial)

    error = adbDevice.check_device_connection()

    self.assertNotEqual(error, None)
    self.assertEqual(error.message, ("Device with serial %s is not connected."
                                     % invalid_device_serial))
    self.assertEqual(error.suggestion, None)

  @mock.patch.dict(os.environ, {"ANDROID_SERIAL": TEST_DEVICE_SERIAL},
                   clear=True)
  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_env_variable_in_devices(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
    adbDevice = AdbDevice(None)

    error = adbDevice.check_device_connection()

    self.assertEqual(error, None)
    self.assertEqual(adbDevice.serial, TEST_DEVICE_SERIAL)

  @mock.patch.dict(os.environ, {"ANDROID_SERIAL": "invalid-device-serial"},
                   clear=True)
  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_env_variable_not_in_devices_error(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
    adbDevice = AdbDevice(None)

    error = adbDevice.check_device_connection()

    self.assertNotEqual(error, None)
    self.assertEqual(error.message, ("Device with serial invalid-device-serial"
                                     " is set as environment variable,"
                                     " ANDROID_SERIAL, but is not connected."))
    self.assertEqual(error.suggestion, None)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_adb_devices_command_fails_error(self,
      mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(None)

    with self.assertRaises(Exception) as e:
      adbDevice.check_device_connection()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_no_devices_connected_error(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (self.generate_adb_devices_result([]))
    adbDevice = AdbDevice(None)

    error = adbDevice.check_device_connection()

    self.assertNotEqual(error, None)
    self.assertEqual(error.message, "There are currently no devices connected.")
    self.assertEqual(error.suggestion, None)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_no_devices_connected_adb_not_started_error(
      self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([], False))
    adbDevice = AdbDevice(None)

    error = adbDevice.check_device_connection()

    self.assertNotEqual(error, None)
    self.assertEqual(error.message, "There are currently no devices connected.")
    self.assertEqual(error.suggestion, None)

  @mock.patch.dict(os.environ, {}, clear=True)
  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_only_one_device(self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL]))
    adbDevice = AdbDevice(None)

    error = adbDevice.check_device_connection()

    self.assertEqual(error, None)
    self.assertEqual(adbDevice.serial, TEST_DEVICE_SERIAL)

  @mock.patch.dict(os.environ, {}, clear=True)
  @mock.patch.object(subprocess, "run", autospec=True)
  def test_check_device_connection_multiple_devices_error(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL,
                                          TEST_DEVICE_SERIAL2]))
    adbDevice = AdbDevice(None)

    error = adbDevice.check_device_connection()

    self.assertNotEqual(error, None)
    self.assertEqual(error.message, ("There is more than one device currently"
                                     " connected."))
    self.assertEqual(error.suggestion, ("Run one of the following commands to"
                                        " choose one of the connected devices:"
                                        "\n\t torq --serial %s"
                                        "\n\t torq --serial %s"
                                        % (TEST_DEVICE_SERIAL,
                                           TEST_DEVICE_SERIAL2)))

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_root_device_success(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = [
        self.generate_mock_completed_process(),
        self.generate_adb_devices_result([TEST_DEVICE_SERIAL])]
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.root_device()

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_root_device_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.root_device()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_root_device_times_out_error(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = lambda args, capture_output=True: (
        next(self.subprocess_output(self.generate_adb_devices_result([]),
                                    self.generate_mock_completed_process())))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.root_device()

    self.assertEqual(str(e.exception), ("Device with serial %s took too long to"
                                        " reconnect after being rooted."
                                        % TEST_DEVICE_SERIAL))

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_root_device_and_adb_devices_fails_error(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = [self.generate_mock_completed_process(),
                                       TEST_EXCEPTION]
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.root_device()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_remove_file_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.remove_file(TEST_FILE_PATH)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_remove_file_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.remove_file(TEST_FILE_PATH)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "Popen", autospec=True)
  def test_start_perfetto_trace_success(self, mock_subprocess_popen):
    # Mocking the return value of subprocess.Popen to ensure it's
    # not modified and returned by AdbDevice.start_perfetto_trace
    mock_subprocess_popen.return_value = mock.Mock()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    mock_process = adbDevice.start_perfetto_trace(None)

    # No exception is expected to be thrown
    self.assertEqual(mock_process, mock_subprocess_popen.return_value)

  @mock.patch.object(subprocess, "Popen", autospec=True)
  def test_start_perfetto_trace_failure(self, mock_subprocess_popen):
    mock_subprocess_popen.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.start_perfetto_trace(None)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "Popen", autospec=True)
  def test_start_simpleperf_trace_success(self, mock_subprocess_popen):
    # Mocking the return value of subprocess.Popen to ensure it's
    # not modified and returned by AdbDevice.start_simpleperf_trace
    mock_subprocess_popen.return_value = mock.Mock()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)
    command = ProfilerCommand("profiler", "custom", None, None,
                              10000, None, None, ["cpu-cycles"], None, None,
                              None, None, None, None, None)
    mock_process = adbDevice.start_simpleperf_trace(command)

    # No exception is expected to be thrown
    self.assertEqual(mock_process, mock_subprocess_popen.return_value)

  @mock.patch.object(subprocess, "Popen", autospec=True)
  def test_start_simpleperf_trace_failure(self, mock_subprocess_popen):
    mock_subprocess_popen.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    command = ProfilerCommand("profiler", "custom", None, None,
                              10000, None, None, ["cpu-cycles"], None, None,
                              None, None, None, None, None)
    with self.assertRaises(Exception) as e:
      adbDevice.start_simpleperf_trace(command)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)


  @mock.patch.object(subprocess, "run", autospec=True)
  def test_pull_file_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.pull_file(TEST_FILE_PATH, TEST_FILE_PATH)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_pull_file_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.pull_file(TEST_FILE_PATH, TEST_FILE_PATH)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_all_users_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.mock_users()

    users = AdbDevice(TEST_DEVICE_SERIAL).get_all_users()

    self.assertEqual(users, [TEST_USER_ID_1, TEST_USER_ID_2])

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_all_users_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.get_all_users()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_user_exists_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.mock_users()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    error = adbDevice.user_exists(TEST_USER_ID_1)

    self.assertEqual(error, None)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_user_exists_and_user_does_not_exist_failure(self,
      mock_subprocess_run):
    mock_subprocess_run.return_value = self.mock_users()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    error = adbDevice.user_exists(TEST_USER_ID_3)

    self.assertNotEqual(error, None)
    self.assertEqual(error.message, ("User ID %s does not exist on device with"
                                     " serial %s." % (TEST_USER_ID_3,
                                                      TEST_DEVICE_SERIAL)))
    self.assertEqual(error.suggestion,
                     ("Select from one of the following user IDs on device with"
                      " serial %s: %s, %s"
                      % (TEST_DEVICE_SERIAL, TEST_USER_ID_1, TEST_USER_ID_2)))

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_user_exists_and_get_all_users_fails_error(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.user_exists(TEST_USER_ID_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_current_user_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        mock.create_autospec(subprocess.CompletedProcess, instance=True,
                             stdout=b'%d\n' % TEST_USER_ID_1))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    user = adbDevice.get_current_user()

    self.assertEqual(user, TEST_USER_ID_1)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_current_user_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.get_current_user()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_perform_user_switch_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.perform_user_switch(TEST_USER_ID_1)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_perform_user_switch_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.perform_user_switch(TEST_USER_ID_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_write_to_file_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.write_to_file(TEST_FILE_PATH, TEST_STRING_FILE)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_write_to_file_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.write_to_file(TEST_FILE_PATH, TEST_STRING_FILE)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_set_prop_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.set_prop(TEST_PROP, TEST_PROP_VALUE)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_set_prop_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.set_prop(TEST_PROP, TEST_PROP_VALUE)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_reboot_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.reboot()

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_reboot_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.reboot()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_wait_for_device_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.wait_for_device()

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_wait_for_device_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.wait_for_device()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_is_boot_completed_and_is_completed(self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_mock_completed_process(BOOT_COMPLETE_OUTPUT))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    is_completed = adbDevice.is_boot_completed()

    self.assertEqual(is_completed, True)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_is_boot_completed_and_is_not_completed(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    is_completed = adbDevice.is_boot_completed()

    self.assertEqual(is_completed, False)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_is_boot_completed_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.is_boot_completed()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_wait_for_boot_to_complete_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_mock_completed_process(BOOT_COMPLETE_OUTPUT))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.wait_for_boot_to_complete()

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_wait_for_boot_to_complete_and_is_boot_completed_fails_error(self,
      mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.wait_for_boot_to_complete()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_wait_for_boot_to_complete_times_out_error(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.wait_for_boot_to_complete()

    self.assertEqual(str(e.exception), ("Device with serial %s took too long to"
                                        " finish rebooting."
                                        % adbDevice.serial))

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_packages_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.mock_packages()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    packages = adbDevice.get_packages()

    self.assertEqual(packages, [TEST_PACKAGE_1, TEST_PACKAGE_2])

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_packages_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.get_packages()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_pid_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process(
        TEST_PID_OUTPUT)
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    process_id = adbDevice.get_pid(TEST_PACKAGE_1)

    self.assertEqual(process_id, "8241")

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_pid_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.get_pid(TEST_PACKAGE_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_package_running(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process(
        TEST_PID_OUTPUT)
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    is_running = adbDevice.is_package_running(TEST_PACKAGE_1)

    self.assertEqual(is_running, True)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_package_not_running(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    is_running = adbDevice.is_package_running(TEST_PACKAGE_1)

    self.assertEqual(is_running, False)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_package_running_and_get_pid_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.is_package_running(TEST_PACKAGE_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_start_package_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process()
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    error = adbDevice.start_package(TEST_PACKAGE_1)

    self.assertEqual(error, None)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_start_package_fails_with_service_app(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process(
        stderr_string=b'%s\n' % TEST_FAILURE_MSG.encode("utf-8"))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    error = adbDevice.start_package(TEST_PACKAGE_1)

    self.assertNotEqual(error, None)
    self.assertEqual(error.message, ("Cannot start package %s on device with"
                                     " serial %s because %s is a service"
                                     " package, which doesn't implement a MAIN"
                                     " activity." % (TEST_PACKAGE_1,
                                                     TEST_DEVICE_SERIAL,
                                                     TEST_PACKAGE_1)))
    self.assertEqual(error.suggestion, None)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_start_package_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.start_package(TEST_PACKAGE_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_kill_pid_success(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = [
        self.generate_mock_completed_process(TEST_PID_OUTPUT), None]
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.kill_pid(TEST_PACKAGE_1)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_kill_pid_and_get_pid_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.kill_pid(TEST_PACKAGE_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_kill_pid_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = [
        self.generate_mock_completed_process(TEST_PID_OUTPUT), TEST_EXCEPTION]
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.kill_pid(TEST_PACKAGE_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_force_stop_package_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = None
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    # No exception is expected to be thrown
    adbDevice.force_stop_package(TEST_PACKAGE_1)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_force_stop_package_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.force_stop_package(TEST_PACKAGE_1)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_prop_success(self, mock_subprocess_run):
    test_prop_value = ANDROID_SDK_VERSION_T
    mock_subprocess_run.return_value = self.generate_mock_completed_process(
        stdout_string=b'%d\n' % test_prop_value)
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    prop_value = int(adbDevice.get_prop(TEST_PROP))

    self.assertEqual(prop_value, test_prop_value)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_prop_package_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.get_prop(TEST_PROP)

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_android_sdk_version_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = self.generate_mock_completed_process(
        stdout_string=b'%d\n' % ANDROID_SDK_VERSION_T)
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    prop_value = adbDevice.get_android_sdk_version()

    self.assertEqual(prop_value, ANDROID_SDK_VERSION_T)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_get_android_sdk_version_failure(self, mock_subprocess_run):
    mock_subprocess_run.side_effect = TEST_EXCEPTION
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    with self.assertRaises(Exception) as e:
      adbDevice.get_android_sdk_version()

    self.assertEqual(str(e.exception), TEST_FAILURE_MSG)

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_simpleperf_event_exists_success(self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_mock_completed_process(b'List of software events:\n  '
                                             b'alignment-faults\n  '
                                             b'context-switches\n  '
                                             b'cpu-clock\n  '
                                             b'cpu-migrations\n  '
                                             b'emulation-faults\n  '
                                             b'major-faults\n  '
                                             b'minor-faults\n  page-faults\n  '
                                             b'task-clock'))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    events = ["cpu-clock", "minor-faults"]
    # No exception is expected to be thrown
    error = adbDevice.simpleperf_event_exists(events)

    self.assertEqual(error, None)
    # Check that the list passed to the function is unchanged
    self.assertEqual(events, ["cpu-clock", "minor-faults"])

  @mock.patch.object(subprocess, "run", autospec=True)
  def test_simpleperf_event_exists_failure(self, mock_subprocess_run):
    mock_subprocess_run.return_value = (
        self.generate_mock_completed_process(b'List of software events:\n  '
                                             b'alignment-faults\n  '
                                             b'context-switches\n  '
                                             b'cpu-clock\n  '
                                             b'cpu-migrations\n  '
                                             b'emulation-faults\n  '
                                             b'major-faults\n  '
                                             b'minor-faults\n  page-faults\n  '
                                             b'task-clock'))
    adbDevice = AdbDevice(TEST_DEVICE_SERIAL)

    error = adbDevice.simpleperf_event_exists(["cpu-clock", "minor-faults",
                                               "List"])

    self.assertEqual(error.message, "The following simpleperf event(s) are "
                                    "invalid: ['List'].")
    self.assertEqual(error.suggestion, "Run adb shell simpleperf list to"
                                       " see valid simpleperf events.")


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