# Copyright 2018 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 logging
import os
import shutil
import tempfile
import unittest
from unittest import mock

from mobly import records
from mobly import runtime_test_info
from mobly.controllers import android_device
from mobly.controllers.android_device_lib import adb
from mobly.controllers.android_device_lib.services import logcat
from tests.lib import mock_android_device

# 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 mocked 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)
# The expected result of the cat adb operation.
MOCK_ADB_UNICODE_LOGCAT_CAT_RESULT = [
    '02-29 14:02:21.456  4454  Something \u901a\n',
    '02-29 14:02:21.789  4454  Something again\n',
]
# A mocked piece of adb logcat output.
MOCK_ADB_UNICODE_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_UNICODE_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'

# Mock AdbError for missing logpersist scripts
MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR = adb.AdbError(
    'logpersist.stop --clear',
    b'',
    '/system/bin/sh: logpersist.stop: not found',
    0,
)
MOCK_LOGPERSIST_START_MISSING_ADB_ERROR = adb.AdbError(
    'logpersist.start --clear',
    b'',
    b'/system/bin/sh: logpersist.stop: not found',
    0,
)


class LogcatTest(unittest.TestCase):
  """Tests for Logcat service and its integration with AndroidDevice."""

  def setUp(self):
    # Set log_path to logging since mobly 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)

  def AssertFileContains(self, content, file_path):
    with open(file_path, 'r') as f:
      output = f.read()
    self.assertIn(content, output)

  def AssertFileDoesNotContain(self, content, file_path):
    with open(file_path, 'r') as f:
      output = f.read()
    self.assertNotIn(content, output)

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  @mock.patch('mobly.utils.create_dir')
  @mock.patch('mobly.utils.start_standing_subprocess', return_value='process')
  @mock.patch('mobly.utils.stop_standing_subprocess')
  @mock.patch.object(logcat.Logcat, '_open_logcat_file')
  @mock.patch('mobly.logger.get_log_file_timestamp')
  def test_start_and_stop(
      self,
      get_timestamp_mock,
      open_logcat_mock,
      stop_proc_mock,
      start_proc_mock,
      create_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.
    """
    mock_serial = '1'
    get_timestamp_mock.return_value = '123'
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service.start()
    # Verify start did the correct operations.
    self.assertTrue(logcat_service._adb_logcat_process)
    expected_log_path = os.path.join(
        logging.log_path,
        'AndroidDevice%s' % ad.serial,
        'logcat,%s,fakemodel,123.txt' % ad.serial,
    )
    create_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
    adb_cmd = ' "adb" -s %s logcat -v threadtime -T 1  >> %s'
    start_proc_mock.assert_called_with(
        adb_cmd % (ad.serial, '"%s" ' % expected_log_path), shell=True
    )
    self.assertEqual(logcat_service.adb_logcat_file_path, expected_log_path)
    expected_msg = 'Logcat thread is already running, cannot start another one.'
    # Expect error if start is called back to back.
    with self.assertRaisesRegex(logcat.Error, expected_msg):
      logcat_service.start()
    # Verify stop did the correct operations.
    logcat_service.stop()
    stop_proc_mock.assert_called_with('process')
    self.assertIsNone(logcat_service._adb_logcat_process)
    self.assertEqual(logcat_service.adb_logcat_file_path, expected_log_path)

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  @mock.patch('mobly.utils.start_standing_subprocess')
  @mock.patch(
      'mobly.controllers.android_device.list_fastboot_devices', return_value='1'
  )
  def test_start_in_fastboot_mode(
      self, _, start_proc_mock, FastbootProxy, MockAdbProxy
  ):
    mock_serial = '1'
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service.start()
    # Verify start is not performed
    self.assertFalse(logcat_service._adb_logcat_process)
    start_proc_mock.assert_not_called()

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  @mock.patch('mobly.utils.create_dir')
  @mock.patch('mobly.utils.start_standing_subprocess', return_value='process')
  @mock.patch('mobly.utils.stop_standing_subprocess')
  @mock.patch.object(logcat.Logcat, '_open_logcat_file')
  def test_update_config(
      self,
      open_logcat_mock,
      stop_proc_mock,
      start_proc_mock,
      create_dir_mock,
      FastbootProxy,
      MockAdbProxy,
  ):
    mock_serial = '1'
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service.start()
    logcat_service.stop()
    new_log_params = '-a -b -c'
    new_file_path = 'some/path/log.txt'
    new_config = logcat.Config(
        logcat_params=new_log_params, output_file_path=new_file_path
    )
    logcat_service.update_config(new_config)
    logcat_service.start()
    self.assertTrue(logcat_service._adb_logcat_process)
    create_dir_mock.assert_has_calls([mock.call('some/path')])
    expected_adb_cmd = (
        ' "adb" -s 1 logcat -v threadtime -T 1 -a -b -c >> "some/path/log.txt" '
    )
    start_proc_mock.assert_called_with(expected_adb_cmd, shell=True)
    self.assertEqual(logcat_service.adb_logcat_file_path, 'some/path/log.txt')
    logcat_service.stop()

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  @mock.patch('mobly.utils.create_dir')
  @mock.patch('mobly.utils.start_standing_subprocess', return_value='process')
  @mock.patch('mobly.utils.stop_standing_subprocess')
  @mock.patch.object(logcat.Logcat, '_open_logcat_file')
  def test_update_config_while_running(
      self,
      open_logcat_mock,
      stop_proc_mock,
      start_proc_mock,
      create_dir_mock,
      FastbootProxy,
      MockAdbProxy,
  ):
    mock_serial = '1'
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service.start()
    new_config = logcat.Config(
        logcat_params='-blah', output_file_path='some/path/file.txt'
    )
    with self.assertRaisesRegex(
        logcat.Error,
        'Logcat thread is already running, cannot start another one',
    ):
      logcat_service.update_config(new_config)
    self.assertTrue(logcat_service.is_alive)
    logcat_service.stop()

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  @mock.patch('mobly.utils.create_dir')
  @mock.patch('mobly.utils.start_standing_subprocess', return_value='process')
  @mock.patch('mobly.utils.stop_standing_subprocess')
  @mock.patch.object(logcat.Logcat, '_open_logcat_file')
  @mock.patch(
      'mobly.controllers.android_device_lib.services.logcat.Logcat.clear_adb_log',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  def test_pause_and_resume(
      self,
      clear_adb_mock,
      open_logcat_mock,
      stop_proc_mock,
      start_proc_mock,
      create_dir_mock,
      FastbootProxy,
      MockAdbProxy,
  ):
    mock_serial = '1'
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad, logcat.Config(clear_log=True))
    logcat_service.start()
    clear_adb_mock.assert_called_once_with()
    self.assertTrue(logcat_service.is_alive)
    logcat_service.pause()
    self.assertFalse(logcat_service.is_alive)
    stop_proc_mock.assert_called_with('process')
    self.assertIsNone(logcat_service._adb_logcat_process)
    clear_adb_mock.reset_mock()
    logcat_service.resume()
    self.assertTrue(logcat_service.is_alive)
    clear_adb_mock.assert_not_called()
    logcat_service.stop()

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  @mock.patch('mobly.utils.start_standing_subprocess', return_value='process')
  @mock.patch('mobly.utils.stop_standing_subprocess')
  @mock.patch(
      'mobly.controllers.android_device_lib.services.logcat.Logcat.clear_adb_log',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  def test_logcat_service_create_output_excerpts(
      self,
      clear_adb_mock,
      stop_proc_mock,
      start_proc_mock,
      FastbootProxy,
      MockAdbProxy,
  ):
    mock_serial = '1'
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service._start()

    def _write_logcat_file_and_assert_excerpts_exists(
        logcat_file_content, test_begin_time, test_name
    ):
      with open(logcat_service.adb_logcat_file_path, 'a') as f:
        f.write(logcat_file_content)
      test_output_dir = os.path.join(self.tmp_dir, test_name)
      mock_record = records.TestResultRecord(test_name)
      mock_record.begin_time = test_begin_time
      mock_record.signature = f'{test_name}-{test_begin_time}'
      test_run_info = runtime_test_info.RuntimeTestInfo(
          test_name, test_output_dir, mock_record
      )
      actual_path = logcat_service.create_output_excerpts(test_run_info)[0]
      expected_path = os.path.join(
          test_output_dir,
          '{test_name}-{test_begin_time}'.format(
              test_name=test_name, test_begin_time=test_begin_time
          ),
          'logcat,{mock_serial},fakemodel,{test_name}-{test_begin_time}.txt'.format(
              mock_serial=mock_serial,
              test_name=test_name,
              test_begin_time=test_begin_time,
          ),
      )
      self.assertEqual(actual_path, expected_path)
      self.assertTrue(os.path.exists(expected_path))
      return expected_path

    # Generate logs before the file pointer is created.
    # This message will not be captured in the excerpt.
    NOT_IN_EXCERPT = 'Not in excerpt.\n'
    with open(logcat_service.adb_logcat_file_path, 'a') as f:
      f.write(NOT_IN_EXCERPT)
    # With the file pointer created, generate logs and make an excerpt.
    logcat_service._open_logcat_file()
    FILE_CONTENT = 'Some log.\n'
    expected_path1 = _write_logcat_file_and_assert_excerpts_exists(
        logcat_file_content=FILE_CONTENT,
        test_begin_time=123,
        test_name='test_foo',
    )
    self.AssertFileContains(FILE_CONTENT, expected_path1)
    self.AssertFileDoesNotContain(NOT_IN_EXCERPT, expected_path1)
    # Generate some new logs and do another excerpt.
    FILE_CONTENT = 'Some more logs!!!\n'
    expected_path2 = _write_logcat_file_and_assert_excerpts_exists(
        logcat_file_content=FILE_CONTENT,
        test_begin_time=456,
        test_name='test_bar',
    )
    self.AssertFileContains(FILE_CONTENT, expected_path2)
    self.AssertFileDoesNotContain(FILE_CONTENT, expected_path1)
    # Simulate devices accidentally go offline, logcat service stopped.
    logcat_service.stop()
    FILE_CONTENT = 'Whatever logs\n'
    expected_path3 = _write_logcat_file_and_assert_excerpts_exists(
        logcat_file_content=FILE_CONTENT,
        test_begin_time=789,
        test_name='test_offline',
    )
    self.assertEqual(os.stat(expected_path3).st_size, 0)

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  @mock.patch('mobly.utils.create_dir')
  @mock.patch('mobly.utils.start_standing_subprocess', return_value='process')
  @mock.patch('mobly.utils.stop_standing_subprocess')
  @mock.patch.object(logcat.Logcat, '_open_logcat_file')
  @mock.patch('mobly.logger.get_log_file_timestamp')
  def test_take_logcat_with_extra_params(
      self,
      get_timestamp_mock,
      open_logcat_mock,
      stop_proc_mock,
      start_proc_mock,
      create_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.
    """
    mock_serial = '1'
    get_timestamp_mock.return_value = '123'
    ad = android_device.AndroidDevice(serial=mock_serial)
    configs = logcat.Config()
    configs.logcat_params = '-b radio'
    logcat_service = logcat.Logcat(ad, configs)
    logcat_service.start()
    # Verify start did the correct operations.
    self.assertTrue(logcat_service._adb_logcat_process)
    expected_log_path = os.path.join(
        logging.log_path,
        'AndroidDevice%s' % ad.serial,
        'logcat,%s,fakemodel,123.txt' % ad.serial,
    )
    create_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
    adb_cmd = ' "adb" -s %s logcat -v threadtime -T 1 -b radio >> %s'
    start_proc_mock.assert_called_with(
        adb_cmd % (ad.serial, '"%s" ' % expected_log_path), shell=True
    )
    self.assertEqual(logcat_service.adb_logcat_file_path, expected_log_path)
    logcat_service.stop()

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock_android_device.MockAdbProxy('1'),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  def test_instantiation(self, MockFastboot, MockAdbProxy):
    """Verifies the AndroidDevice object's basic attributes are correctly
    set after instantiation.
    """
    mock_serial = 1
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    self.assertIsNone(logcat_service._adb_logcat_process)
    self.assertIsNone(logcat_service.adb_logcat_file_path)

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock.MagicMock(),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  def test__enable_logpersist_with_logpersist(self, MockFastboot, MockAdbProxy):
    mock_serial = '1'
    mock_adb_proxy = MockAdbProxy.return_value
    mock_adb_proxy.getprops.return_value = {
        'ro.build.id': 'AB42',
        'ro.build.type': 'userdebug',
        'ro.debuggable': '1',
    }
    mock_adb_proxy.has_shell_command.side_effect = lambda command: {
        'logpersist.start': True,
        'logpersist.stop': True,
    }[command]
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service._enable_logpersist()
    mock_adb_proxy.shell.assert_has_calls(
        [
            mock.call('logpersist.stop --clear'),
            mock.call('logpersist.start'),
        ]
    )

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock.MagicMock(),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  def test__enable_logpersist_with_user_build_device(
      self, MockFastboot, MockAdbProxy
  ):
    mock_serial = '1'
    mock_adb_proxy = MockAdbProxy.return_value
    mock_adb_proxy.getprops.return_value = {
        'ro.build.id': 'AB42',
        'ro.build.type': 'user',
        'ro.debuggable': '0',
    }
    mock_adb_proxy.has_shell_command.side_effect = lambda command: {
        'logpersist.start': True,
        'logpersist.stop': True,
    }[command]
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service._enable_logpersist()
    mock_adb_proxy.shell.assert_not_called()

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock.MagicMock(),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  def test__enable_logpersist_with_missing_all_logpersist(
      self, MockFastboot, MockAdbProxy
  ):
    def adb_shell_helper(command):
      if command == 'logpersist.start':
        raise MOCK_LOGPERSIST_START_MISSING_ADB_ERROR
      elif command == 'logpersist.stop --clear':
        raise MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR
      else:
        return b''

    mock_serial = '1'
    mock_adb_proxy = MockAdbProxy.return_value
    mock_adb_proxy.getprops.return_value = {
        'ro.build.id': 'AB42',
        'ro.build.type': 'userdebug',
        'ro.debuggable': '1',
    }
    mock_adb_proxy.has_shell_command.side_effect = lambda command: {
        'logpersist.start': False,
        'logpersist.stop': False,
    }[command]
    mock_adb_proxy.shell.side_effect = adb_shell_helper
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service._enable_logpersist()
    mock_adb_proxy.shell.assert_not_called()

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock.MagicMock(),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  def test__enable_logpersist_with_missing_logpersist_stop(
      self, MockFastboot, MockAdbProxy
  ):
    def adb_shell_helper(command):
      if command == 'logpersist.stop --clear':
        raise MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR
      else:
        return b''

    mock_serial = '1'
    mock_adb_proxy = MockAdbProxy.return_value
    mock_adb_proxy.getprops.return_value = {
        'ro.build.id': 'AB42',
        'ro.build.type': 'userdebug',
        'ro.debuggable': '1',
    }
    mock_adb_proxy.has_shell_command.side_effect = lambda command: {
        'logpersist.start': True,
        'logpersist.stop': False,
    }[command]
    mock_adb_proxy.shell.side_effect = adb_shell_helper
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service._enable_logpersist()
    mock_adb_proxy.shell.assert_has_calls(
        [
            mock.call('logpersist.stop --clear'),
        ]
    )

  @mock.patch(
      'mobly.controllers.android_device_lib.adb.AdbProxy',
      return_value=mock.MagicMock(),
  )
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  def test__enable_logpersist_with_missing_logpersist_start(
      self, MockFastboot, MockAdbProxy
  ):
    def adb_shell_helper(command):
      if command == 'logpersist.start':
        raise MOCK_LOGPERSIST_START_MISSING_ADB_ERROR
      else:
        return b''

    mock_serial = '1'
    mock_adb_proxy = MockAdbProxy.return_value
    mock_adb_proxy.getprops.return_value = {
        'ro.build.id': 'AB42',
        'ro.build.type': 'userdebug',
        'ro.debuggable': '1',
    }
    mock_adb_proxy.has_shell_command.side_effect = lambda command: {
        'logpersist.start': False,
        'logpersist.stop': True,
    }[command]
    mock_adb_proxy.shell.side_effect = adb_shell_helper
    ad = android_device.AndroidDevice(serial=mock_serial)
    logcat_service = logcat.Logcat(ad)
    logcat_service._enable_logpersist()
    mock_adb_proxy.shell.assert_not_called()

  @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy')
  @mock.patch(
      'mobly.controllers.android_device_lib.fastboot.FastbootProxy',
      return_value=mock_android_device.MockFastbootProxy('1'),
  )
  def test_clear_adb_log(self, MockFastboot, MockAdbProxy):
    mock_serial = '1'
    ad = android_device.AndroidDevice(serial=mock_serial)
    ad.adb.logcat = mock.MagicMock()
    ad.adb.logcat.side_effect = adb.AdbError(
        cmd='cmd', stdout=b'', stderr=b'failed to clear "main" log', ret_code=1
    )
    logcat_service = logcat.Logcat(ad)
    logcat_service.clear_adb_log()


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