unit_tests/controllers/android_lib/logcat_test.py

#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# 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 unittest

import mock
from antlion.controllers.android_lib import logcat
from antlion.controllers.android_lib.logcat import TimestampTracker

BASE_TIMESTAMP = '2000-01-01 12:34:56.789   123 75348 '


class LogcatTest(unittest.TestCase):
    """Tests antlion.controllers.android_lib.logcat"""

    @staticmethod
    def patch(patched):
        return mock.patch('antlion.controllers.android_lib.logcat.%s' % patched)

    def setUp(self):
        self._get_log_level = logcat._get_log_level

    def tearDown(self):
        logcat._get_log_level = self._get_log_level

    # TimestampTracker

    def test_read_output_sets_last_timestamp_if_found(self):
        tracker = TimestampTracker()
        tracker.read_output(BASE_TIMESTAMP + 'D message')

        self.assertEqual(tracker.last_timestamp, '2000-01-01 12:34:56.789')

    def test_read_output_keeps_last_timestamp_if_no_new_stamp_is_found(self):
        tracker = TimestampTracker()
        tracker.read_output(BASE_TIMESTAMP + 'D message')
        tracker.read_output('--------- beginning of main')

        self.assertEqual(tracker.last_timestamp, '2000-01-01 12:34:56.789')

    def test_read_output_updates_timestamp_to_first_in_results(self):
        tracker = TimestampTracker()
        tracker.read_output(BASE_TIMESTAMP + 'D 9999-99-99 12:34:56.789')

        self.assertEqual(tracker.last_timestamp, '2000-01-01 12:34:56.789')

    # _get_log_level

    def test_get_log_level_verbose(self):
        """Tests that Logcat's verbose logs make it to the debug level."""
        level = logcat._get_log_level(BASE_TIMESTAMP + 'V')

        self.assertEqual(level, logging.DEBUG)

    def test_get_log_level_debug(self):
        """Tests that Logcat's debug logs make it to the debug level."""
        level = logcat._get_log_level(BASE_TIMESTAMP + 'D')

        self.assertEqual(level, logging.DEBUG)

    def test_get_log_level_info(self):
        """Tests that Logcat's info logs make it to the info level."""
        level = logcat._get_log_level(BASE_TIMESTAMP + 'I')

        self.assertEqual(level, logging.INFO)

    def test_get_log_level_warning(self):
        """Tests that Logcat's warning logs make it to the warning level."""
        level = logcat._get_log_level(BASE_TIMESTAMP + 'W')

        self.assertEqual(level, logging.WARNING)

    def test_get_log_level_error(self):
        """Tests that Logcat's error logs make it to the error level."""
        level = logcat._get_log_level(BASE_TIMESTAMP + 'E')

        self.assertEqual(level, logging.ERROR)

    def test_get_log_level_markers(self):
        """Tests that Logcat's marker logs make it to the error level."""
        level = logcat._get_log_level('--------- beginning of main')

        self.assertEqual(level, logging.ERROR)

    # _log_line_func

    def test_log_line_func_returns_func_that_logs_to_given_logger(self):
        logcat._get_log_level = lambda message: logging.INFO
        tracker = mock.Mock()
        log = mock.Mock()
        message = 'MESSAGE'

        logcat._log_line_func(log, tracker)(message)

        self.assertEqual(log.log.called, True)
        log.log.assert_called_once_with(logging.INFO, message)

    def test_log_line_func_returns_func_that_updates_the_timestamp(self):
        logcat._get_log_level = lambda message: logging.INFO
        tracker = mock.Mock()
        log = mock.Mock()
        message = 'MESSAGE'

        logcat._log_line_func(log, tracker)(message)

        self.assertEqual(tracker.read_output.called, True)
        tracker.read_output.assert_called_once_with(message)

    # _on_retry

    def test_on_retry_returns_func_that_formats_with_last_timestamp(self):
        tracker = TimestampTracker()
        tracker.read_output(BASE_TIMESTAMP)
        new_command = logcat._on_retry('S3R14L', 'extra_params', tracker)(None)

        self.assertIn('-T "%s"' % tracker.last_timestamp, new_command)

    def test_on_retry_func_returns_string_that_contains_the_given_serial(self):
        tracker = TimestampTracker()
        tracker.read_output(BASE_TIMESTAMP)
        new_command = logcat._on_retry('S3R14L', 'extra_params', tracker)(None)

        self.assertTrue('-s S3R14L' in new_command)

    def test_on_retry_func_returns_string_that_contains_any_extra_params(self):
        tracker = TimestampTracker()
        tracker.read_output(BASE_TIMESTAMP)
        new_command = logcat._on_retry('S3R14L', 'extra_params', tracker)(None)

        self.assertTrue('extra_params' in new_command)

    # create_logcat_keepalive_process

    def test_create_logcat_keepalive_process_creates_a_new_logger(self):
        with self.patch('log_stream') as log_stream, self.patch('Process'):
            logcat.create_logcat_keepalive_process('S3R14L', 'dir')
        self.assertEqual(log_stream.create_logger.call_args[0][0],
                         'adblog_S3R14L')
        self.assertEqual(log_stream.create_logger.call_args[1]['subcontext'],
                         'dir')

    def test_create_logcat_keepalive_process_creates_a_new_process(self):
        with self.patch('log_stream'), self.patch('Process') as process:
            logcat.create_logcat_keepalive_process('S3R14L', 'dir')

        self.assertIn('S3R14L', process.call_args[0][0])

    def test_create_logcat_keepalive_process_sets_output_callback(self):
        with self.patch('log_stream'), self.patch('Process'):
            process = logcat.create_logcat_keepalive_process('S3R14L', 'dir')

        self.assertEqual(process.set_on_output_callback.called, True)

    def test_create_logcat_keepalive_process_sets_on_terminate_callback(self):
        with self.patch('log_stream'), self.patch('Process'):
            process = logcat.create_logcat_keepalive_process('S3R14L', 'dir')

        self.assertEqual(process.set_on_terminate_callback.called, True)


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