#!/usr/bin/env python3.4
#
#   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 base_test
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/"
# The expected result of the cat adb operation.
MOCK_ADB_LOGCAT_CAT_RESULT = [
    "02-29 14:02:21.456  4454  Something\n",
    "02-29 14:02:21.789  4454  Something again\n"]
# A mockd piece of adb logcat output.
MOCK_ADB_LOGCAT = (
    "02-29 14:02:19.123  4454  Nothing\n"
    "%s"
    "02-29 14:02:22.123  4454  Something again and again\n"
    ) % ''.join(MOCK_ADB_LOGCAT_CAT_RESULT)
# Mock start and end time of the adb cat.
MOCK_ADB_LOGCAT_BEGIN_TIME = "02-29 14:02:20.123"
MOCK_ADB_LOGCAT_END_TIME = "02-29 14:02:22.000"

def get_mock_ads(num, logger=None):
    """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",
                            logger=logger,
                            serial=i,
                            h_port=None)
        ads.append(ad)
    return ads

def get_mock_logger():
    return mock.MagicMock(name="Logger", log_path=MOCK_LOG_PATH)

def mock_get_all_instances(logger=None):
    return get_mock_ads(5, logger=logger)

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

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

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

    def shell(self, params):
        if params == "id -u":
            return b"root"
        if (params == "getprop | grep ro.build.product" or
            params == "getprop | grep ro.product.name"):
            return b"[ro.build.product]: [FakeModel]"

    def bugreport(self, params):
        expected = os.path.join(MOCK_LOG_PATH,
                                "AndroidDevice%s" % self.serial,
                                "BugReports",
                                "test_something,sometime,%s.txt" % (
                                    self.serial))
        expected = " > %s" % expected
        assert params == expected, "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):
            clean_name = name.replace('_', '-')
            arg_str = ' '.join(str(elem) for elem in args)
            return arg_str
        return adb_call

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

    def setUp(self):
        """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.assertRaisesRegexp(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.assertRaisesRegexp(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.assertRaisesRegexp(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.assertRaisesRegexp(android_device.AndroidDeviceError,
                                     expected_msg):
            ad = android_device.get_device(ads, serial=target_serial)

    # 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(1))
    def test_AndroidDevice_instantiation(self, MockAdbProxy):
        """Verifies the AndroidDevice object's basic attributes are correctly
        set after instantiation.
        """
        mock_serial = 1
        ml = get_mock_logger()
        ad = android_device.AndroidDevice(serial=mock_serial, logger=ml)
        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(ml.log_path,
                                   "AndroidDevice%s" % mock_serial)
        self.assertEqual(ad.log_path, expected_lp)

    @mock.patch('acts.controllers.adb.AdbProxy', return_value=MockAdbProxy(1))
    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.exe_cmd')
    def test_AndroidDevice_take_bug_report(self,
                                           exe_mock,
                                           create_dir_mock,
                                           MockAdbProxy):
        """Verifies AndroidDevice.take_bug_report calls the correct adb command
        and writes the bugreport file to the correct path.
        """
        mock_serial = 1
        ml = get_mock_logger()
        ad = android_device.AndroidDevice(serial=mock_serial, logger=ml)
        ad.take_bug_report("test_something", "sometime")
        expected_path = os.path.join(MOCK_LOG_PATH,
                                     "AndroidDevice%s" % ad.serial,
                                     "BugReports")
        create_dir_mock.assert_called_with(expected_path)

    @mock.patch('acts.controllers.adb.AdbProxy', return_value=MockAdbProxy(1))
    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.start_standing_subprocess', return_value="process")
    @mock.patch('acts.utils.stop_standing_subprocess')
    def test_AndroidDevice_take_logcat(self,
                                       stop_proc_mock,
                                       start_proc_mock,
                                       creat_dir_mock,
                                       MockAdbProxy):
        """Verifies the steps of collecting adb logcat on an AndroidDevice
        object, including various function calls and the expected behaviors of
        the calls.
        """
        mock_serial = 1
        ml = get_mock_logger()
        ad = android_device.AndroidDevice(serial=mock_serial, logger=ml)
        expected_msg = ("Android device .* does not have an ongoing adb logcat"
                        " collection.")
        # Expect error if stop is called before start.
        with self.assertRaisesRegexp(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(
                                    MOCK_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 -v threadtime -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.assertRaisesRegexp(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(1))
    @mock.patch('acts.utils.create_dir')
    @mock.patch('acts.utils.start_standing_subprocess', return_value="process")
    @mock.patch('acts.utils.stop_standing_subprocess')
    def test_AndroidDevice_take_logcat_with_user_param(self,
                                                       stop_proc_mock,
                                                       start_proc_mock,
                                                       creat_dir_mock,
                                                       MockAdbProxy):
        """Verifies the steps of collecting adb logcat on an AndroidDevice
        object, including various function calls and the expected behaviors of
        the calls.
        """
        mock_serial = 1
        ml = get_mock_logger()
        ad = android_device.AndroidDevice(serial=mock_serial, logger=ml)
        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.assertRaisesRegexp(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(
                                    MOCK_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 -v threadtime -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(1))
    @mock.patch('acts.utils.start_standing_subprocess', return_value="process")
    @mock.patch('acts.utils.stop_standing_subprocess')
    @mock.patch('acts.logger.get_log_line_timestamp',
                return_value=MOCK_ADB_LOGCAT_END_TIME)
    def test_AndroidDevice_cat_adb_log(self,
                                       mock_timestamp_getter,
                                       stop_proc_mock,
                                       start_proc_mock,
                                       MockAdbProxy):
        """Verifies that AndroidDevice.cat_adb_log loads the correct adb log
        file, locates the correct adb log lines within the given time range,
        and writes the lines to the correct output file.
        """
        mock_serial = 1
        ml = get_mock_logger()
        ad = android_device.AndroidDevice(serial=mock_serial, logger=ml)
        # Expect error if attempted to cat adb log before starting adb logcat.
        expected_msg = ("Attempting to cat adb log when none has been "
                        "collected on Android device .*")
        with self.assertRaisesRegexp(android_device.AndroidDeviceError,
                                     expected_msg):
            ad.cat_adb_log("some_test", MOCK_ADB_LOGCAT_BEGIN_TIME)
        ad.start_adb_logcat()
        # Direct the log path of the ad to a temp dir to avoid racing.
        ad.log_path = os.path.join(self.tmp_dir, ad.log_path)
        mock_adb_log_path = os.path.join(ad.log_path, "adblog,%s,%s.txt" %
                                         (ad.model, ad.serial))
        with open(mock_adb_log_path, 'w') as f:
            f.write(MOCK_ADB_LOGCAT)
        ad.cat_adb_log("some_test", MOCK_ADB_LOGCAT_BEGIN_TIME)
        cat_file_path = os.path.join(ad.log_path,
                                     "AdbLogExcerpts",
                                     ("some_test,02-29 14:02:20.123,%s,%s.txt"
                                     ) % (ad.model, ad.serial))
        with open(cat_file_path, 'r') as f:
            actual_cat = f.read()
        self.assertEqual(actual_cat, ''.join(MOCK_ADB_LOGCAT_CAT_RESULT))
        # Stops adb logcat.
        ad.stop_adb_logcat()

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