#!/usr/bin/env python3
#
#   Copyright 2016 - 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 logging
import mock
import os
import shutil
import tempfile
import unittest

from acts import logger
from acts.controllers import android_device

# Mock log path for a test run.
MOCK_LOG_PATH = "/tmp/logs/MockTest/xx-xx-xx_xx-xx-xx/"

# Mock start and end time of the adb cat.
MOCK_ADB_LOGCAT_BEGIN_TIME = "1970-01-02 21:03:20.123"
MOCK_ADB_LOGCAT_END_TIME = "1970-01-02 21:22:02.000"
MOCK_ADB_EPOCH_BEGIN_TIME = 191000123

MOCK_SERIAL = 1
MOCK_RELEASE_BUILD_ID = "ABC1.123456.007"
MOCK_DEV_BUILD_ID = "ABC-MR1"
MOCK_NYC_BUILD_ID = "N4F27P"


def get_mock_ads(num):
    """Generates a list of mock AndroidDevice objects.

    The serial number of each device will be integer 0 through num - 1.

    Args:
        num: An integer that is the number of mock AndroidDevice objects to
            create.
    """
    ads = []
    for i in range(num):
        ad = mock.MagicMock(name="AndroidDevice", serial=i, h_port=None)
        ad.ensure_screen_on = mock.MagicMock(return_value=True)
        ads.append(ad)
    return ads


def mock_get_all_instances():
    return get_mock_ads(5)


def mock_list_adb_devices():
    return [ad.serial for ad in get_mock_ads(5)]


class MockAdbProxy(object):
    """Mock class that swaps out calls to adb with mock calls."""

    def __init__(self,
                 serial,
                 fail_br=False,
                 fail_br_before_N=False,
                 build_id=MOCK_RELEASE_BUILD_ID):
        self.serial = serial
        self.fail_br = fail_br
        self.fail_br_before_N = fail_br_before_N
        self.return_value = None
        self.return_multiple = False
        self.build_id = build_id

    def shell(self, params, ignore_status=False, timeout=60):
        if params == "id -u":
            return "root"
        elif params == "bugreportz":
            if self.fail_br:
                return "OMG I died!\n"
            return "OK:/path/bugreport.zip\n"
        elif params == "bugreportz -v":
            if self.fail_br_before_N:
                return "/system/bin/sh: bugreportz: not found"
            return "1.1"
        else:
            if self.return_multiple:
                return self.return_value.pop(0)
            else:
                return self.return_value

    def getprop(self, params):
        if params == "ro.build.id":
            return self.build_id
        elif params == "ro.build.version.incremental":
            return "123456789"
        elif params == "ro.build.type":
            return "userdebug"
        elif params == "ro.build.product" or params == "ro.product.name":
            return "FakeModel"
        elif params == "sys.boot_completed":
            return "1"

    def devices(self):
        return "\t".join([str(self.serial), "device"])

    def bugreport(self, params, timeout=android_device.BUG_REPORT_TIMEOUT):
        expected = os.path.join(
            logging.log_path, "AndroidDevice%s" % self.serial,
            "test_something", "AndroidDevice%s_%s" %
            (self.serial,
             logger.normalize_log_line_timestamp(MOCK_ADB_LOGCAT_BEGIN_TIME)))
        assert expected in params, "Expected '%s', got '%s'." % (expected,
                                                                 params)

    def __getattr__(self, name):
        """All calls to the none-existent functions in adb proxy would
        simply return the adb command string.
        """

        def adb_call(*args, **kwargs):
            arg_str = ' '.join(str(elem) for elem in args)
            return arg_str

        return adb_call


class MockFastbootProxy():
    """Mock class that swaps out calls to adb with mock calls."""

    def __init__(self, serial):
        self.serial = serial

    def devices(self):
        return "xxxx\tdevice\nyyyy\tdevice"

    def __getattr__(self, name):
        def fastboot_call(*args):
            arg_str = ' '.join(str(elem) for elem in args)
            return arg_str

        return fastboot_call


class ActsAndroidDeviceTest(unittest.TestCase):
    """This test class has unit tests for the implementation of everything
    under acts.controllers.android_device.
    """

    def setUp(self):
        # Set log_path to logging since acts logger setup is not called.
        if not hasattr(logging, "log_path"):
            setattr(logging, "log_path", "/tmp/logs")
        # Creates a temp dir to be used by tests in this test class.
        self.tmp_dir = tempfile.mkdtemp()

    def tearDown(self):
        """Removes the temp dir.
        """
        shutil.rmtree(self.tmp_dir)

    # Tests for android_device module functions.
    # These tests use mock AndroidDevice instances.

    @mock.patch.object(
        android_device, "get_all_instances", new=mock_get_all_instances)
    @mock.patch.object(
        android_device, "list_adb_devices", new=mock_list_adb_devices)
    def test_create_with_pickup_all(self):
        pick_all_token = android_device.ANDROID_DEVICE_PICK_ALL_TOKEN
        actual_ads = android_device.create(pick_all_token)
        for actual, expected in zip(actual_ads, get_mock_ads(5)):
            self.assertEqual(actual.serial, expected.serial)

    def test_create_with_empty_config(self):
        expected_msg = android_device.ANDROID_DEVICE_EMPTY_CONFIG_MSG
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            android_device.create([])

    def test_create_with_not_list_config(self):
        expected_msg = android_device.ANDROID_DEVICE_NOT_LIST_CONFIG_MSG
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            android_device.create("HAHA")

    def test_get_device_success_with_serial(self):
        ads = get_mock_ads(5)
        expected_serial = 0
        ad = android_device.get_device(ads, serial=expected_serial)
        self.assertEqual(ad.serial, expected_serial)

    def test_get_device_success_with_serial_and_extra_field(self):
        ads = get_mock_ads(5)
        expected_serial = 1
        expected_h_port = 5555
        ads[1].h_port = expected_h_port
        ad = android_device.get_device(
            ads, serial=expected_serial, h_port=expected_h_port)
        self.assertEqual(ad.serial, expected_serial)
        self.assertEqual(ad.h_port, expected_h_port)

    def test_get_device_no_match(self):
        ads = get_mock_ads(5)
        expected_msg = ("Could not find a target device that matches condition"
                        ": {'serial': 5}.")
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            ad = android_device.get_device(ads, serial=len(ads))

    def test_get_device_too_many_matches(self):
        ads = get_mock_ads(5)
        target_serial = ads[1].serial = ads[0].serial
        expected_msg = "More than one device matched: \[0, 0\]"
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            ad = android_device.get_device(ads, serial=target_serial)

    def test_start_services_on_ads(self):
        """Makes sure when an AndroidDevice fails to start some services, all
        AndroidDevice objects get cleaned up.
        """
        msg = "Some error happened."
        ads = get_mock_ads(3)
        ads[0].start_services = mock.MagicMock()
        ads[0].clean_up = mock.MagicMock()
        ads[1].start_services = mock.MagicMock()
        ads[1].clean_up = mock.MagicMock()
        ads[2].start_services = mock.MagicMock(
            side_effect=android_device.AndroidDeviceError(msg))
        ads[2].clean_up = mock.MagicMock()
        with self.assertRaisesRegex(android_device.AndroidDeviceError, msg):
            android_device._start_services_on_ads(ads)
        ads[0].clean_up.assert_called_once_with()
        ads[1].clean_up.assert_called_once_with()
        ads[2].clean_up.assert_called_once_with()

    # Tests for android_device.AndroidDevice class.
    # These tests mock out any interaction with the OS and real android device
    # in AndroidDeivce.

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    def test_AndroidDevice_instantiation(self, MockFastboot, MockAdbProxy):
        """Verifies the AndroidDevice object's basic attributes are correctly
        set after instantiation.
        """
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        self.assertEqual(ad.serial, 1)
        self.assertEqual(ad.model, "fakemodel")
        self.assertIsNone(ad.adb_logcat_process)
        self.assertIsNone(ad.adb_logcat_file_path)
        expected_lp = os.path.join(logging.log_path,
                                   "AndroidDevice%s" % MOCK_SERIAL)
        self.assertEqual(ad.log_path, expected_lp)

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    def test_AndroidDevice_build_info_release(self, MockFastboot,
                                              MockAdbProxy):
        """Verifies the AndroidDevice object's basic attributes are correctly
        set after instantiation.
        """
        ad = android_device.AndroidDevice(serial=1)
        build_info = ad.build_info
        self.assertEqual(build_info["build_id"], "ABC1.123456.007")
        self.assertEqual(build_info["build_type"], "userdebug")

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL, build_id=MOCK_DEV_BUILD_ID))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    def test_AndroidDevice_build_info_release(self, MockFastboot,
                                              MockAdbProxy):
        """Verifies the AndroidDevice object's basic attributes are correctly
        set after instantiation.
        """
        global MOCK_BUILD_ID
        ad = android_device.AndroidDevice(serial=1)
        old_mock_build_id = MOCK_BUILD_ID
        MOCK_BUILD_ID = "ABC-MR1"
        build_info = ad.build_info
        self.assertEqual(build_info["build_id"], "123456789")
        self.assertEqual(build_info["build_type"], "userdebug")
        MOCK_BUILD_ID = old_mock_build_id

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    def test_AndroidDevice_build_info_dev(self, MockFastboot, MockAdbProxy):
        """Verifies the AndroidDevice object's basic attributes are correctly
        set after instantiation.
        """
        ad = android_device.AndroidDevice(serial=1)
        build_info = ad.build_info
        self.assertEqual(build_info["build_id"], "123456789")
        self.assertEqual(build_info["build_type"], "userdebug")

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL, build_id=MOCK_NYC_BUILD_ID))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    def test_AndroidDevice_build_info_nyc(self, MockFastboot, MockAdbProxy):
        """Verifies the AndroidDevice object's build id is set correctly for
        NYC releases.
        """
        ad = android_device.AndroidDevice(serial=1)
        build_info = ad.build_info
        self.assertEqual(build_info["build_id"], MOCK_NYC_BUILD_ID)

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))

    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.exe_cmd')
    def test_AndroidDevice_take_bug_report(self, exe_mock, create_dir_mock,
                                           FastbootProxy, MockAdbProxy):
        """Verifies AndroidDevice.take_bug_report calls the correct adb command
        and writes the bugreport file to the correct path.
        """
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        ad.take_bug_report("test_something", 234325.32)
        expected_path = os.path.join(
            logging.log_path, "AndroidDevice%s" % ad.serial, "test_something")
        create_dir_mock.assert_called_with(expected_path)

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL, fail_br=True))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.exe_cmd')
    def test_AndroidDevice_take_bug_report_fail(
            self, exe_mock, create_dir_mock, FastbootProxy, MockAdbProxy):
        """Verifies AndroidDevice.take_bug_report writes out the correct message
        when taking bugreport fails.
        """
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        expected_msg = "Failed to take bugreport on 1: OMG I died!"
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            ad.take_bug_report("test_something", 4346343.23)

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL, fail_br_before_N=True))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.exe_cmd')
    def test_AndroidDevice_take_bug_report_fallback(
            self, exe_mock, create_dir_mock, FastbootProxy, MockAdbProxy):
        """Verifies AndroidDevice.take_bug_report falls back to traditional
        bugreport on builds that do not have bugreportz.
        """
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        ad.take_bug_report("test_something", MOCK_ADB_EPOCH_BEGIN_TIME)
        expected_path = os.path.join(
            logging.log_path, "AndroidDevice%s" % ad.serial, "test_something")
        create_dir_mock.assert_called_with(expected_path)

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.start_standing_subprocess', return_value="process")
    @mock.patch('acts.utils.stop_standing_subprocess')
    @mock.patch('acts.utils._assert_subprocess_running')
    def test_AndroidDevice_take_logcat(self, check_proc_mock, stop_proc_mock,
                                       start_proc_mock, creat_dir_mock,
                                       FastbootProxy, MockAdbProxy):
        """Verifies the steps of collecting adb logcat on an AndroidDevice
        object, including various function calls and the expected behaviors of
        the calls.
        """
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        expected_msg = ("Android device .* does not have an ongoing adb logcat"
                        " collection.")
        # Expect error if stop is called before start.
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            ad.stop_adb_logcat()
        ad.start_adb_logcat()
        # Verify start did the correct operations.
        self.assertTrue(ad.adb_logcat_process)
        expected_log_path = os.path.join(logging.log_path,
                                         "AndroidDevice%s" % ad.serial,
                                         "adblog,fakemodel,%s.txt" % ad.serial)
        creat_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
        adb_cmd = 'adb -s %s logcat -T 1 -v year -b all >> %s'
        start_proc_mock.assert_called_with(adb_cmd % (ad.serial,
                                                      expected_log_path))
        self.assertEqual(ad.adb_logcat_file_path, expected_log_path)
        expected_msg = ("Android device .* already has an adb logcat thread "
                        "going on. Cannot start another one.")
        # Expect error if start is called back to back.
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            ad.start_adb_logcat()
        # Verify stop did the correct operations.
        ad.stop_adb_logcat()
        stop_proc_mock.assert_called_with("process")
        self.assertIsNone(ad.adb_logcat_process)
        self.assertEqual(ad.adb_logcat_file_path, expected_log_path)

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    @mock.patch(
        'acts.controllers.fastboot.FastbootProxy',
        return_value=MockFastbootProxy(MOCK_SERIAL))
    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.start_standing_subprocess', return_value="process")
    @mock.patch('acts.utils.stop_standing_subprocess')
    @mock.patch('acts.utils._assert_subprocess_running')
    def test_AndroidDevice_take_logcat_with_user_param(
            self, check_proc_mock, stop_proc_mock, start_proc_mock,
            creat_dir_mock, FastbootProxy, MockAdbProxy):
        """Verifies the steps of collecting adb logcat on an AndroidDevice
        object, including various function calls and the expected behaviors of
        the calls.
        """
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        ad.adb_logcat_param = "-b radio"
        expected_msg = ("Android device .* does not have an ongoing adb logcat"
                        " collection.")
        # Expect error if stop is called before start.
        with self.assertRaisesRegex(android_device.AndroidDeviceError,
                                    expected_msg):
            ad.stop_adb_logcat()
        ad.start_adb_logcat()
        # Verify start did the correct operations.
        self.assertTrue(ad.adb_logcat_process)
        expected_log_path = os.path.join(logging.log_path,
                                         "AndroidDevice%s" % ad.serial,
                                         "adblog,fakemodel,%s.txt" % ad.serial)
        creat_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
        adb_cmd = 'adb -s %s logcat -T 1 -v year -b radio >> %s'
        start_proc_mock.assert_called_with(adb_cmd % (ad.serial,
                                                      expected_log_path))
        self.assertEqual(ad.adb_logcat_file_path, expected_log_path)

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    def test_get_apk_process_id_process_cannot_find(self, adb_proxy):
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        ad.adb.return_value = "does_not_contain_value"
        self.assertEqual(None, ad.get_package_pid("some_package"))

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    def test_get_apk_process_id_process_exists_second_try(self, adb_proxy):
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        ad.adb.return_multiple = True
        ad.adb.return_value = ["", "system 1 2 3 4  S com.some_package"]
        self.assertEqual(1, ad.get_package_pid("some_package"))

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    def test_get_apk_process_id_bad_return(self, adb_proxy):
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        ad.adb.return_value = "bad_return_index_error"
        self.assertEqual(None, ad.get_package_pid("some_package"))

    @mock.patch(
        'acts.controllers.adb.AdbProxy',
        return_value=MockAdbProxy(MOCK_SERIAL))
    def test_get_apk_process_id_bad_return(self, adb_proxy):
        ad = android_device.AndroidDevice(serial=MOCK_SERIAL)
        ad.adb.return_value = "bad return value error"
        self.assertEqual(None, ad.get_package_pid("some_package"))


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