# Copyright 2017 The Abseil 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.

"""Unit tests for absl.logging."""

import contextlib
import functools
import getpass
import io
import logging as std_logging
import os
import re
import socket
import sys
import tempfile
import threading
import time
import traceback
import unittest
from unittest import mock

from absl import flags
from absl import logging
from absl.testing import absltest
from absl.testing import flagsaver
from absl.testing import parameterized

FLAGS = flags.FLAGS


class ConfigurationTest(absltest.TestCase):
  """Tests the initial logging configuration."""

  def test_logger_and_handler(self):
    absl_logger = std_logging.getLogger('absl')
    self.assertIs(absl_logger, logging.get_absl_logger())
    self.assertIsInstance(absl_logger, logging.ABSLLogger)
    self.assertIsInstance(
        logging.get_absl_handler().python_handler.formatter,
        logging.PythonFormatter)


class LoggerLevelsTest(parameterized.TestCase):

  def setUp(self):
    super(LoggerLevelsTest, self).setUp()
    # Since these tests muck with the flag, always save/restore in case the
    # tests forget to clean up properly.
    # enter_context() is py3-only, but manually enter/exit should suffice.
    cm = self.set_logger_levels({})
    cm.__enter__()
    self.addCleanup(lambda: cm.__exit__(None, None, None))

  @contextlib.contextmanager
  def set_logger_levels(self, levels):
    original_levels = {
        name: std_logging.getLogger(name).level for name in levels
    }

    try:
      with flagsaver.flagsaver(logger_levels=levels):
        yield
    finally:
      for name, level in original_levels.items():
        std_logging.getLogger(name).setLevel(level)

  def assert_logger_level(self, name, expected_level):
    logger = std_logging.getLogger(name)
    self.assertEqual(logger.level, expected_level)

  def assert_logged(self, logger_name, expected_msgs):
    logger = std_logging.getLogger(logger_name)
    # NOTE: assertLogs() sets the logger to INFO if not specified.
    with self.assertLogs(logger, logger.level) as cm:
      logger.debug('debug')
      logger.info('info')
      logger.warning('warning')
      logger.error('error')
      logger.critical('critical')

    actual = {r.getMessage() for r in cm.records}
    self.assertEqual(set(expected_msgs), actual)

  def test_setting_levels(self):
    # Other tests change the root logging level, so we can't
    # assume it's the default.
    orig_root_level = std_logging.root.getEffectiveLevel()
    with self.set_logger_levels({'foo': 'ERROR', 'bar': 'DEBUG'}):

      self.assert_logger_level('foo', std_logging.ERROR)
      self.assert_logger_level('bar', std_logging.DEBUG)
      self.assert_logger_level('', orig_root_level)

      self.assert_logged('foo', {'error', 'critical'})
      self.assert_logged('bar',
                         {'debug', 'info', 'warning', 'error', 'critical'})

  @parameterized.named_parameters(
      ('empty', ''),
      ('one_value', 'one:INFO'),
      ('two_values', 'one.a:INFO,two.b:ERROR'),
      ('whitespace_ignored', ' one : DEBUG , two : INFO'),
  )
  def test_serialize_parse(self, levels_str):
    fl = FLAGS['logger_levels']
    fl.parse(levels_str)
    expected = levels_str.replace(' ', '')
    actual = fl.serialize()
    self.assertEqual('--logger_levels={}'.format(expected), actual)

  def test_invalid_value(self):
    with self.assertRaisesRegex(ValueError, 'Unknown level.*10'):
      FLAGS['logger_levels'].parse('foo:10')


class PythonHandlerTest(absltest.TestCase):
  """Tests the PythonHandler class."""

  def setUp(self):
    super().setUp()
    (year, month, day, hour, minute, sec,
     dunno, dayofyear, dst_flag) = (1979, 10, 21, 18, 17, 16, 3, 15, 0)
    self.now_tuple = (year, month, day, hour, minute, sec,
                      dunno, dayofyear, dst_flag)
    self.python_handler = logging.PythonHandler()

  def tearDown(self):
    mock.patch.stopall()
    super().tearDown()

  @flagsaver.flagsaver(logtostderr=False)
  def test_set_google_log_file_no_log_to_stderr(self):
    with mock.patch.object(self.python_handler, 'start_logging_to_file'):
      self.python_handler.use_absl_log_file()
      self.python_handler.start_logging_to_file.assert_called_once_with(
          program_name=None, log_dir=None)

  @flagsaver.flagsaver(logtostderr=True)
  def test_set_google_log_file_with_log_to_stderr(self):
    self.python_handler.stream = None
    self.python_handler.use_absl_log_file()
    self.assertEqual(sys.stderr, self.python_handler.stream)

  @mock.patch.object(logging, 'find_log_dir_and_names')
  @mock.patch.object(logging.time, 'localtime')
  @mock.patch.object(logging.time, 'time')
  @mock.patch.object(os.path, 'islink')
  @mock.patch.object(os, 'unlink')
  @mock.patch.object(os, 'getpid')
  def test_start_logging_to_file(
      self, mock_getpid, mock_unlink, mock_islink, mock_time,
      mock_localtime, mock_find_log_dir_and_names):
    mock_find_log_dir_and_names.return_value = ('here', 'prog1', 'prog1')
    mock_time.return_value = '12345'
    mock_localtime.return_value = self.now_tuple
    mock_getpid.return_value = 4321
    symlink = os.path.join('here', 'prog1.INFO')
    mock_islink.return_value = True
    with mock.patch.object(
        logging, 'open', return_value=sys.stdout, create=True):
      if getattr(os, 'symlink', None):
        with mock.patch.object(os, 'symlink'):
          self.python_handler.start_logging_to_file()
          mock_unlink.assert_called_once_with(symlink)
          os.symlink.assert_called_once_with(
              'prog1.INFO.19791021-181716.4321', symlink)
      else:
        self.python_handler.start_logging_to_file()

  def test_log_file(self):
    handler = logging.PythonHandler()
    self.assertEqual(sys.stderr, handler.stream)

    stream = mock.Mock()
    handler = logging.PythonHandler(stream)
    self.assertEqual(stream, handler.stream)

  def test_flush(self):
    stream = mock.Mock()
    handler = logging.PythonHandler(stream)
    handler.flush()
    stream.flush.assert_called_once()

  def test_flush_with_value_error(self):
    stream = mock.Mock()
    stream.flush.side_effect = ValueError
    handler = logging.PythonHandler(stream)
    handler.flush()
    stream.flush.assert_called_once()

  def test_flush_with_environment_error(self):
    stream = mock.Mock()
    stream.flush.side_effect = EnvironmentError
    handler = logging.PythonHandler(stream)
    handler.flush()
    stream.flush.assert_called_once()

  def test_flush_with_assertion_error(self):
    stream = mock.Mock()
    stream.flush.side_effect = AssertionError
    handler = logging.PythonHandler(stream)
    with self.assertRaises(AssertionError):
      handler.flush()

  def test_ignore_flush_if_stream_is_none(self):
    # Happens if creating a Windows executable without console.
    with mock.patch.object(sys, 'stderr', new=None):
      handler = logging.PythonHandler(None)
      # Test that this does not fail.
      handler.flush()

  def test_ignore_flush_if_stream_does_not_support_flushing(self):
    class BadStream:
      pass

    handler = logging.PythonHandler(BadStream())
    # Test that this does not fail.
    handler.flush()

  def test_log_to_std_err(self):
    record = std_logging.LogRecord(
        'name', std_logging.INFO, 'path', 12, 'logging_msg', [], False)
    with mock.patch.object(std_logging.StreamHandler, 'emit'):
      self.python_handler._log_to_stderr(record)
      std_logging.StreamHandler.emit.assert_called_once_with(record)

  @flagsaver.flagsaver(logtostderr=True)
  def test_emit_log_to_stderr(self):
    record = std_logging.LogRecord(
        'name', std_logging.INFO, 'path', 12, 'logging_msg', [], False)
    with mock.patch.object(self.python_handler, '_log_to_stderr'):
      self.python_handler.emit(record)
      self.python_handler._log_to_stderr.assert_called_once_with(record)

  def test_emit(self):
    stream = io.StringIO()
    handler = logging.PythonHandler(stream)
    handler.stderr_threshold = std_logging.FATAL
    record = std_logging.LogRecord(
        'name', std_logging.INFO, 'path', 12, 'logging_msg', [], False)
    handler.emit(record)
    self.assertEqual(1, stream.getvalue().count('logging_msg'))

  @flagsaver.flagsaver(stderrthreshold='debug')
  def test_emit_and_stderr_threshold(self):
    mock_stderr = io.StringIO()
    stream = io.StringIO()
    handler = logging.PythonHandler(stream)
    record = std_logging.LogRecord(
        'name', std_logging.INFO, 'path', 12, 'logging_msg', [], False)
    with mock.patch.object(sys, 'stderr', new=mock_stderr) as mock_stderr:
      handler.emit(record)
      self.assertEqual(1, stream.getvalue().count('logging_msg'))
      self.assertEqual(1, mock_stderr.getvalue().count('logging_msg'))

  @flagsaver.flagsaver(alsologtostderr=True)
  def test_emit_also_log_to_stderr(self):
    mock_stderr = io.StringIO()
    stream = io.StringIO()
    handler = logging.PythonHandler(stream)
    handler.stderr_threshold = std_logging.FATAL
    record = std_logging.LogRecord(
        'name', std_logging.INFO, 'path', 12, 'logging_msg', [], False)
    with mock.patch.object(sys, 'stderr', new=mock_stderr) as mock_stderr:
      handler.emit(record)
      self.assertEqual(1, stream.getvalue().count('logging_msg'))
      self.assertEqual(1, mock_stderr.getvalue().count('logging_msg'))

  def test_emit_on_stderr(self):
    mock_stderr = io.StringIO()
    with mock.patch.object(sys, 'stderr', new=mock_stderr) as mock_stderr:
      handler = logging.PythonHandler()
      handler.stderr_threshold = std_logging.INFO
      record = std_logging.LogRecord(
          'name', std_logging.INFO, 'path', 12, 'logging_msg', [], False)
      handler.emit(record)
      self.assertEqual(1, mock_stderr.getvalue().count('logging_msg'))

  def test_emit_fatal_absl(self):
    stream = io.StringIO()
    handler = logging.PythonHandler(stream)
    record = std_logging.LogRecord(
        'name', std_logging.FATAL, 'path', 12, 'logging_msg', [], False)
    record.__dict__[logging._ABSL_LOG_FATAL] = True
    with mock.patch.object(handler, 'flush') as mock_flush:
      with mock.patch.object(os, 'abort') as mock_abort:
        handler.emit(record)
        mock_abort.assert_called_once()
        mock_flush.assert_called()  # flush is also called by super class.

  def test_emit_fatal_non_absl(self):
    stream = io.StringIO()
    handler = logging.PythonHandler(stream)
    record = std_logging.LogRecord(
        'name', std_logging.FATAL, 'path', 12, 'logging_msg', [], False)
    with mock.patch.object(os, 'abort') as mock_abort:
      handler.emit(record)
      mock_abort.assert_not_called()

  def test_close(self):
    stream = mock.Mock()
    stream.isatty.return_value = True
    handler = logging.PythonHandler(stream)
    with mock.patch.object(handler, 'flush') as mock_flush:
      with mock.patch.object(std_logging.StreamHandler, 'close') as super_close:
        handler.close()
        mock_flush.assert_called_once()
        super_close.assert_called_once()
        stream.close.assert_not_called()

  def test_close_afile(self):
    stream = mock.Mock()
    stream.isatty.return_value = False
    stream.close.side_effect = ValueError
    handler = logging.PythonHandler(stream)
    with mock.patch.object(handler, 'flush') as mock_flush:
      with mock.patch.object(std_logging.StreamHandler, 'close') as super_close:
        handler.close()
        mock_flush.assert_called_once()
        super_close.assert_called_once()

  def test_close_stderr(self):
    with mock.patch.object(sys, 'stderr') as mock_stderr:
      mock_stderr.isatty.return_value = False
      handler = logging.PythonHandler(sys.stderr)
      handler.close()
      mock_stderr.close.assert_not_called()

  def test_close_stdout(self):
    with mock.patch.object(sys, 'stdout') as mock_stdout:
      mock_stdout.isatty.return_value = False
      handler = logging.PythonHandler(sys.stdout)
      handler.close()
      mock_stdout.close.assert_not_called()

  def test_close_original_stderr(self):
    with mock.patch.object(sys, '__stderr__') as mock_original_stderr:
      mock_original_stderr.isatty.return_value = False
      handler = logging.PythonHandler(sys.__stderr__)
      handler.close()
      mock_original_stderr.close.assert_not_called()

  def test_close_original_stdout(self):
    with mock.patch.object(sys, '__stdout__') as mock_original_stdout:
      mock_original_stdout.isatty.return_value = False
      handler = logging.PythonHandler(sys.__stdout__)
      handler.close()
      mock_original_stdout.close.assert_not_called()

  def test_close_fake_file(self):

    class FakeFile(object):
      """A file-like object that does not implement "isatty"."""

      def __init__(self):
        self.closed = False

      def close(self):
        self.closed = True

      def flush(self):
        pass

    fake_file = FakeFile()
    handler = logging.PythonHandler(fake_file)
    handler.close()
    self.assertTrue(fake_file.closed)


class ABSLHandlerTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    formatter = logging.PythonFormatter()
    self.absl_handler = logging.ABSLHandler(formatter)

  def test_activate_python_handler(self):
    self.absl_handler.activate_python_handler()
    self.assertEqual(
        self.absl_handler._current_handler, self.absl_handler.python_handler)


class ABSLLoggerTest(absltest.TestCase):
  """Tests the ABSLLogger class."""

  def set_up_mock_frames(self):
    """Sets up mock frames for use with the testFindCaller methods."""
    logging_file = os.path.join('absl', 'logging', '__init__.py')

    # Set up mock frame 0
    mock_frame_0 = mock.Mock()
    mock_code_0 = mock.Mock()
    mock_code_0.co_filename = logging_file
    mock_code_0.co_name = 'LoggingLog'
    mock_code_0.co_firstlineno = 124
    mock_frame_0.f_code = mock_code_0
    mock_frame_0.f_lineno = 125

    # Set up mock frame 1
    mock_frame_1 = mock.Mock()
    mock_code_1 = mock.Mock()
    mock_code_1.co_filename = 'myfile.py'
    mock_code_1.co_name = 'Method1'
    mock_code_1.co_firstlineno = 124
    mock_frame_1.f_code = mock_code_1
    mock_frame_1.f_lineno = 125

    # Set up mock frame 2
    mock_frame_2 = mock.Mock()
    mock_code_2 = mock.Mock()
    mock_code_2.co_filename = 'myfile.py'
    mock_code_2.co_name = 'Method2'
    mock_code_2.co_firstlineno = 124
    mock_frame_2.f_code = mock_code_2
    mock_frame_2.f_lineno = 125

    # Set up mock frame 3
    mock_frame_3 = mock.Mock()
    mock_code_3 = mock.Mock()
    mock_code_3.co_filename = 'myfile.py'
    mock_code_3.co_name = 'Method3'
    mock_code_3.co_firstlineno = 124
    mock_frame_3.f_code = mock_code_3
    mock_frame_3.f_lineno = 125

    # Set up mock frame 4 that has the same function name as frame 2.
    mock_frame_4 = mock.Mock()
    mock_code_4 = mock.Mock()
    mock_code_4.co_filename = 'myfile.py'
    mock_code_4.co_name = 'Method2'
    mock_code_4.co_firstlineno = 248
    mock_frame_4.f_code = mock_code_4
    mock_frame_4.f_lineno = 249

    # Tie them together.
    mock_frame_4.f_back = None
    mock_frame_3.f_back = mock_frame_4
    mock_frame_2.f_back = mock_frame_3
    mock_frame_1.f_back = mock_frame_2
    mock_frame_0.f_back = mock_frame_1

    mock.patch.object(sys, '_getframe').start()
    sys._getframe.return_value = mock_frame_0

  def setUp(self):
    super().setUp()
    self.message = 'Hello Nurse'
    self.logger = logging.ABSLLogger('')

  def tearDown(self):
    mock.patch.stopall()
    self.logger._frames_to_skip.clear()
    super().tearDown()

  def test_constructor_without_level(self):
    self.logger = logging.ABSLLogger('')
    self.assertEqual(std_logging.NOTSET, self.logger.getEffectiveLevel())

  def test_constructor_with_level(self):
    self.logger = logging.ABSLLogger('', std_logging.DEBUG)
    self.assertEqual(std_logging.DEBUG, self.logger.getEffectiveLevel())

  def test_find_caller_normal(self):
    self.set_up_mock_frames()
    expected_name = 'Method1'
    self.assertEqual(expected_name, self.logger.findCaller()[2])

  def test_find_caller_skip_method1(self):
    self.set_up_mock_frames()
    self.logger.register_frame_to_skip('myfile.py', 'Method1')
    expected_name = 'Method2'
    self.assertEqual(expected_name, self.logger.findCaller()[2])

  def test_find_caller_skip_method1_and_method2(self):
    self.set_up_mock_frames()
    self.logger.register_frame_to_skip('myfile.py', 'Method1')
    self.logger.register_frame_to_skip('myfile.py', 'Method2')
    expected_name = 'Method3'
    self.assertEqual(expected_name, self.logger.findCaller()[2])

  def test_find_caller_skip_method1_and_method3(self):
    self.set_up_mock_frames()
    self.logger.register_frame_to_skip('myfile.py', 'Method1')
    # Skipping Method3 should change nothing since Method2 should be hit.
    self.logger.register_frame_to_skip('myfile.py', 'Method3')
    expected_name = 'Method2'
    self.assertEqual(expected_name, self.logger.findCaller()[2])

  def test_find_caller_skip_method1_and_method4(self):
    self.set_up_mock_frames()
    self.logger.register_frame_to_skip('myfile.py', 'Method1')
    # Skipping frame 4's Method2 should change nothing for frame 2's Method2.
    self.logger.register_frame_to_skip('myfile.py', 'Method2', 248)
    expected_name = 'Method2'
    expected_frame_lineno = 125
    self.assertEqual(expected_name, self.logger.findCaller()[2])
    self.assertEqual(expected_frame_lineno, self.logger.findCaller()[1])

  def test_find_caller_skip_method1_method2_and_method3(self):
    self.set_up_mock_frames()
    self.logger.register_frame_to_skip('myfile.py', 'Method1')
    self.logger.register_frame_to_skip('myfile.py', 'Method2', 124)
    self.logger.register_frame_to_skip('myfile.py', 'Method3')
    expected_name = 'Method2'
    expected_frame_lineno = 249
    self.assertEqual(expected_name, self.logger.findCaller()[2])
    self.assertEqual(expected_frame_lineno, self.logger.findCaller()[1])

  def test_find_caller_stack_info(self):
    self.set_up_mock_frames()
    self.logger.register_frame_to_skip('myfile.py', 'Method1')
    with mock.patch.object(traceback, 'print_stack') as print_stack:
      self.assertEqual(
          ('myfile.py', 125, 'Method2', 'Stack (most recent call last):'),
          self.logger.findCaller(stack_info=True))
    print_stack.assert_called_once()

  def test_critical(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.critical(self.message)
      self.logger.log.assert_called_once_with(
          std_logging.CRITICAL, self.message)

  def test_fatal(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.fatal(self.message)
      self.logger.log.assert_called_once_with(std_logging.FATAL, self.message)

  def test_error(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.error(self.message)
      self.logger.log.assert_called_once_with(std_logging.ERROR, self.message)

  def test_warn(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.warn(self.message)
      self.logger.log.assert_called_once_with(std_logging.WARN, self.message)

  def test_warning(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.warning(self.message)
      self.logger.log.assert_called_once_with(std_logging.WARNING, self.message)

  def test_info(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.info(self.message)
      self.logger.log.assert_called_once_with(std_logging.INFO, self.message)

  def test_debug(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.debug(self.message)
      self.logger.log.assert_called_once_with(std_logging.DEBUG, self.message)

  def test_log_debug_with_python(self):
    with mock.patch.object(self.logger, 'log'):
      FLAGS.verbosity = 1
      self.logger.debug(self.message)
      self.logger.log.assert_called_once_with(std_logging.DEBUG, self.message)

  def test_log_fatal_with_python(self):
    with mock.patch.object(self.logger, 'log'):
      self.logger.fatal(self.message)
      self.logger.log.assert_called_once_with(std_logging.FATAL, self.message)

  def test_register_frame_to_skip(self):
    # This is basically just making sure that if I put something in a
    # list, it actually appears in that list.
    frame_tuple = ('file', 'method')
    self.logger.register_frame_to_skip(*frame_tuple)
    self.assertIn(frame_tuple, self.logger._frames_to_skip)

  def test_register_frame_to_skip_with_lineno(self):
    frame_tuple = ('file', 'method', 123)
    self.logger.register_frame_to_skip(*frame_tuple)
    self.assertIn(frame_tuple, self.logger._frames_to_skip)

  def test_logger_cannot_be_disabled(self):
    self.logger.disabled = True
    record = self.logger.makeRecord(
        'name', std_logging.INFO, 'fn', 20, 'msg', [], False)
    with mock.patch.object(self.logger, 'callHandlers') as mock_call_handlers:
      self.logger.handle(record)
    mock_call_handlers.assert_called_once()


class ABSLLogPrefixTest(parameterized.TestCase):

  def setUp(self):
    super().setUp()
    self.record = std_logging.LogRecord(
        'name', std_logging.INFO, 'path/to/source.py', 13, 'log message',
        None, None)

  @parameterized.named_parameters(
      ('debug', std_logging.DEBUG, 'I'),
      ('info', std_logging.INFO, 'I'),
      ('warning', std_logging.WARNING, 'W'),
      ('error', std_logging.ERROR, 'E'),
  )
  def test_default_prefixes(self, levelno, level_prefix):
    self.record.levelno = levelno
    self.record.created = 1494293880.378885
    thread_id = '{: >5}'.format(logging._get_thread_id())
    # Use UTC so the test passes regardless of the local time zone.
    with mock.patch.object(time, 'localtime', side_effect=time.gmtime):
      self.assertEqual(
          '{}0509 01:38:00.378885 {} source.py:13] '.format(
              level_prefix, thread_id),
          logging.get_absl_log_prefix(self.record))
      time.localtime.assert_called_once_with(self.record.created)

  def test_absl_prefix_regex(self):
    self.record.created = 1226888258.0521369
    # Use UTC so the test passes regardless of the local time zone.
    with mock.patch.object(time, 'localtime', side_effect=time.gmtime):
      prefix = logging.get_absl_log_prefix(self.record)

    match = re.search(logging.ABSL_LOGGING_PREFIX_REGEX, prefix)
    self.assertTrue(match)

    expect = {'severity': 'I',
              'month': '11',
              'day': '17',
              'hour': '02',
              'minute': '17',
              'second': '38',
              'microsecond': '052136',
              'thread_id': str(logging._get_thread_id()),
              'filename': 'source.py',
              'line': '13',
             }
    actual = {name: match.group(name) for name in expect}
    self.assertEqual(expect, actual)

  def test_critical_absl(self):
    self.record.levelno = std_logging.CRITICAL
    self.record.created = 1494293880.378885
    self.record._absl_log_fatal = True
    thread_id = '{: >5}'.format(logging._get_thread_id())
    # Use UTC so the test passes regardless of the local time zone.
    with mock.patch.object(time, 'localtime', side_effect=time.gmtime):
      self.assertEqual(
          'F0509 01:38:00.378885 {} source.py:13] '.format(thread_id),
          logging.get_absl_log_prefix(self.record))
      time.localtime.assert_called_once_with(self.record.created)

  def test_critical_non_absl(self):
    self.record.levelno = std_logging.CRITICAL
    self.record.created = 1494293880.378885
    thread_id = '{: >5}'.format(logging._get_thread_id())
    # Use UTC so the test passes regardless of the local time zone.
    with mock.patch.object(time, 'localtime', side_effect=time.gmtime):
      self.assertEqual(
          'E0509 01:38:00.378885 {} source.py:13] CRITICAL - '.format(
              thread_id),
          logging.get_absl_log_prefix(self.record))
      time.localtime.assert_called_once_with(self.record.created)


class LogCountTest(absltest.TestCase):

  def test_counter_threadsafe(self):
    threads_start = threading.Event()
    counts = set()
    k = object()

    def t():
      threads_start.wait()
      counts.add(logging._get_next_log_count_per_token(k))

    threads = [threading.Thread(target=t) for _ in range(100)]
    for thread in threads:
      thread.start()
    threads_start.set()
    for thread in threads:
      thread.join()
    self.assertEqual(counts, {i for i in range(100)})


class LoggingTest(absltest.TestCase):

  def test_fatal(self):
    with mock.patch.object(os, 'abort') as mock_abort:
      logging.fatal('Die!')
      mock_abort.assert_called_once()

  def test_find_log_dir_with_arg(self):
    with mock.patch.object(os, 'access'), \
        mock.patch.object(os.path, 'isdir'):
      os.path.isdir.return_value = True
      os.access.return_value = True
      log_dir = logging.find_log_dir(log_dir='./')
      self.assertEqual('./', log_dir)

  @flagsaver.flagsaver(log_dir='./')
  def test_find_log_dir_with_flag(self):
    with mock.patch.object(os, 'access'), \
        mock.patch.object(os.path, 'isdir'):
      os.path.isdir.return_value = True
      os.access.return_value = True
      log_dir = logging.find_log_dir()
      self.assertEqual('./', log_dir)

  @flagsaver.flagsaver(log_dir='')
  def test_find_log_dir_with_hda_tmp(self):
    with mock.patch.object(os, 'access'), \
        mock.patch.object(os.path, 'exists'), \
        mock.patch.object(os.path, 'isdir'):
      os.path.exists.return_value = True
      os.path.isdir.return_value = True
      os.access.return_value = True
      log_dir = logging.find_log_dir()
      self.assertEqual(tempfile.gettempdir(), log_dir)

  @flagsaver.flagsaver(log_dir='')
  def test_find_log_dir_with_tmp(self):
    with mock.patch.object(os, 'access'), \
        mock.patch.object(os.path, 'exists'), \
        mock.patch.object(os.path, 'isdir'):
      os.path.exists.return_value = False
      os.path.isdir.side_effect = lambda path: path == tempfile.gettempdir()
      os.access.return_value = True
      log_dir = logging.find_log_dir()
      self.assertEqual(tempfile.gettempdir(), log_dir)

  def test_find_log_dir_with_nothing(self):
    with mock.patch.object(os.path, 'exists'), \
        mock.patch.object(os.path, 'isdir'):
      os.path.exists.return_value = False
      os.path.isdir.return_value = False
      with self.assertRaises(FileNotFoundError):
        logging.find_log_dir()

  def test_find_log_dir_and_names_with_args(self):
    user = 'test_user'
    host = 'test_host'
    log_dir = 'here'
    program_name = 'prog1'
    with mock.patch.object(getpass, 'getuser'), \
        mock.patch.object(logging, 'find_log_dir') as mock_find_log_dir, \
        mock.patch.object(socket, 'gethostname') as mock_gethostname:
      getpass.getuser.return_value = user
      mock_gethostname.return_value = host
      mock_find_log_dir.return_value = log_dir

      prefix = '%s.%s.%s.log' % (program_name, host, user)
      self.assertEqual((log_dir, prefix, program_name),
                       logging.find_log_dir_and_names(
                           program_name=program_name, log_dir=log_dir))

  def test_find_log_dir_and_names_without_args(self):
    user = 'test_user'
    host = 'test_host'
    log_dir = 'here'
    py_program_name = 'py_prog1'
    sys.argv[0] = 'path/to/prog1'
    with mock.patch.object(getpass, 'getuser'), \
        mock.patch.object(logging, 'find_log_dir') as mock_find_log_dir, \
        mock.patch.object(socket, 'gethostname'):
      getpass.getuser.return_value = user
      socket.gethostname.return_value = host
      mock_find_log_dir.return_value = log_dir
      prefix = '%s.%s.%s.log' % (py_program_name, host, user)
      self.assertEqual((log_dir, prefix, py_program_name),
                       logging.find_log_dir_and_names())

  def test_find_log_dir_and_names_wo_username(self):
    # Windows doesn't have os.getuid at all
    if hasattr(os, 'getuid'):
      mock_getuid = mock.patch.object(os, 'getuid')
      uid = 100
      logged_uid = '100'
    else:
      # The function doesn't exist, but our test code still tries to mock
      # it, so just use a fake thing.
      mock_getuid = _mock_windows_os_getuid()
      uid = -1
      logged_uid = 'unknown'

    host = 'test_host'
    log_dir = 'here'
    program_name = 'prog1'
    with mock.patch.object(getpass, 'getuser'), \
        mock_getuid as getuid, \
        mock.patch.object(logging, 'find_log_dir') as mock_find_log_dir, \
        mock.patch.object(socket, 'gethostname') as mock_gethostname:
      getpass.getuser.side_effect = KeyError()
      getuid.return_value = uid
      mock_gethostname.return_value = host
      mock_find_log_dir.return_value = log_dir

      prefix = '%s.%s.%s.log' % (program_name, host, logged_uid)
      self.assertEqual((log_dir, prefix, program_name),
                       logging.find_log_dir_and_names(
                           program_name=program_name, log_dir=log_dir))

  def test_errors_in_logging(self):
    with mock.patch.object(sys, 'stderr', new=io.StringIO()) as stderr:
      logging.info('not enough args: %s %s', 'foo')  # pylint: disable=logging-too-few-args
      self.assertIn('Traceback (most recent call last):', stderr.getvalue())
      self.assertIn('TypeError', stderr.getvalue())

  def test_dict_arg(self):
    # Tests that passing a dictionary as a single argument does not crash.
    logging.info('%(test)s', {'test': 'Hello world!'})

  def test_exception_dict_format(self):
    # Just verify that this doesn't raise a TypeError.
    logging.exception('%(test)s', {'test': 'Hello world!'})

  def test_exception_with_exc_info(self):
    # Just verify that this doesn't raise a KeyeError.
    logging.exception('exc_info=True', exc_info=True)
    logging.exception('exc_info=False', exc_info=False)

  def test_logging_levels(self):
    old_level = logging.get_verbosity()

    logging.set_verbosity(logging.DEBUG)
    self.assertEqual(logging.get_verbosity(), logging.DEBUG)
    self.assertTrue(logging.level_debug())
    self.assertTrue(logging.level_info())
    self.assertTrue(logging.level_warning())
    self.assertTrue(logging.level_error())

    logging.set_verbosity(logging.INFO)
    self.assertEqual(logging.get_verbosity(), logging.INFO)
    self.assertFalse(logging.level_debug())
    self.assertTrue(logging.level_info())
    self.assertTrue(logging.level_warning())
    self.assertTrue(logging.level_error())

    logging.set_verbosity(logging.WARNING)
    self.assertEqual(logging.get_verbosity(), logging.WARNING)
    self.assertFalse(logging.level_debug())
    self.assertFalse(logging.level_info())
    self.assertTrue(logging.level_warning())
    self.assertTrue(logging.level_error())

    logging.set_verbosity(logging.ERROR)
    self.assertEqual(logging.get_verbosity(), logging.ERROR)
    self.assertFalse(logging.level_debug())
    self.assertFalse(logging.level_info())
    self.assertTrue(logging.level_error())

    logging.set_verbosity(old_level)

  def test_set_verbosity_strings(self):
    old_level = logging.get_verbosity()

    # Lowercase names.
    logging.set_verbosity('debug')
    self.assertEqual(logging.get_verbosity(), logging.DEBUG)
    logging.set_verbosity('info')
    self.assertEqual(logging.get_verbosity(), logging.INFO)
    logging.set_verbosity('warning')
    self.assertEqual(logging.get_verbosity(), logging.WARNING)
    logging.set_verbosity('warn')
    self.assertEqual(logging.get_verbosity(), logging.WARNING)
    logging.set_verbosity('error')
    self.assertEqual(logging.get_verbosity(), logging.ERROR)
    logging.set_verbosity('fatal')

    # Uppercase names.
    self.assertEqual(logging.get_verbosity(), logging.FATAL)
    logging.set_verbosity('DEBUG')
    self.assertEqual(logging.get_verbosity(), logging.DEBUG)
    logging.set_verbosity('INFO')
    self.assertEqual(logging.get_verbosity(), logging.INFO)
    logging.set_verbosity('WARNING')
    self.assertEqual(logging.get_verbosity(), logging.WARNING)
    logging.set_verbosity('WARN')
    self.assertEqual(logging.get_verbosity(), logging.WARNING)
    logging.set_verbosity('ERROR')
    self.assertEqual(logging.get_verbosity(), logging.ERROR)
    logging.set_verbosity('FATAL')
    self.assertEqual(logging.get_verbosity(), logging.FATAL)

    # Integers as strings.
    logging.set_verbosity(str(logging.DEBUG))
    self.assertEqual(logging.get_verbosity(), logging.DEBUG)
    logging.set_verbosity(str(logging.INFO))
    self.assertEqual(logging.get_verbosity(), logging.INFO)
    logging.set_verbosity(str(logging.WARNING))
    self.assertEqual(logging.get_verbosity(), logging.WARNING)
    logging.set_verbosity(str(logging.ERROR))
    self.assertEqual(logging.get_verbosity(), logging.ERROR)
    logging.set_verbosity(str(logging.FATAL))
    self.assertEqual(logging.get_verbosity(), logging.FATAL)

    logging.set_verbosity(old_level)

  def test_key_flags(self):
    key_flags = FLAGS.get_key_flags_for_module(logging)
    key_flag_names = [flag.name for flag in key_flags]
    self.assertIn('stderrthreshold', key_flag_names)
    self.assertIn('verbosity', key_flag_names)

  def test_get_absl_logger(self):
    self.assertIsInstance(
        logging.get_absl_logger(), logging.ABSLLogger)

  def test_get_absl_handler(self):
    self.assertIsInstance(
        logging.get_absl_handler(), logging.ABSLHandler)


@mock.patch.object(logging.ABSLLogger, 'register_frame_to_skip')
class LogSkipPrefixTest(absltest.TestCase):
  """Tests for logging.skip_log_prefix."""

  def _log_some_info(self):
    """Logging helper function for LogSkipPrefixTest."""
    logging.info('info')

  def _log_nested_outer(self):
    """Nested logging helper functions for LogSkipPrefixTest."""
    def _log_nested_inner():
      logging.info('info nested')
    return _log_nested_inner

  def test_skip_log_prefix_with_name(self, mock_skip_register):
    retval = logging.skip_log_prefix('_log_some_info')
    mock_skip_register.assert_called_once_with(__file__, '_log_some_info', None)
    self.assertEqual(retval, '_log_some_info')

  def test_skip_log_prefix_with_func(self, mock_skip_register):
    retval = logging.skip_log_prefix(self._log_some_info)
    mock_skip_register.assert_called_once_with(
        __file__, '_log_some_info', mock.ANY)
    self.assertEqual(retval, self._log_some_info)

  def test_skip_log_prefix_with_functools_partial(self, mock_skip_register):
    partial_input = functools.partial(self._log_some_info)
    with self.assertRaises(ValueError):
      _ = logging.skip_log_prefix(partial_input)
    mock_skip_register.assert_not_called()

  def test_skip_log_prefix_with_lambda(self, mock_skip_register):
    lambda_input = lambda _: self._log_some_info()
    retval = logging.skip_log_prefix(lambda_input)
    mock_skip_register.assert_called_once_with(__file__, '<lambda>', mock.ANY)
    self.assertEqual(retval, lambda_input)

  def test_skip_log_prefix_with_bad_input(self, mock_skip_register):
    dict_input = {1: 2, 2: 3}
    with self.assertRaises(TypeError):
      _ = logging.skip_log_prefix(dict_input)
    mock_skip_register.assert_not_called()

  def test_skip_log_prefix_with_nested_func(self, mock_skip_register):
    nested_input = self._log_nested_outer()
    retval = logging.skip_log_prefix(nested_input)
    mock_skip_register.assert_called_once_with(
        __file__, '_log_nested_inner', mock.ANY)
    self.assertEqual(retval, nested_input)

  def test_skip_log_prefix_decorator(self, mock_skip_register):

    @logging.skip_log_prefix
    def _log_decorated():
      logging.info('decorated')

    del _log_decorated
    mock_skip_register.assert_called_once_with(
        __file__, '_log_decorated', mock.ANY)


@contextlib.contextmanager
def override_python_handler_stream(stream):
  handler = logging.get_absl_handler().python_handler
  old_stream = handler.stream
  handler.stream = stream
  try:
    yield
  finally:
    handler.stream = old_stream


class GetLogFileNameTest(parameterized.TestCase):

  @parameterized.named_parameters(
      ('err', sys.stderr),
      ('out', sys.stdout),
  )
  def test_get_log_file_name_py_std(self, stream):
    with override_python_handler_stream(stream):
      self.assertEqual('', logging.get_log_file_name())

  def test_get_log_file_name_py_no_name(self):

    class FakeFile(object):
      pass

    with override_python_handler_stream(FakeFile()):
      self.assertEqual('', logging.get_log_file_name())

  def test_get_log_file_name_py_file(self):
    _, filename = tempfile.mkstemp(dir=absltest.TEST_TMPDIR.value)
    with open(filename, 'a') as stream:
      with override_python_handler_stream(stream):
        self.assertEqual(filename, logging.get_log_file_name())


@contextlib.contextmanager
def _mock_windows_os_getuid():
  yield mock.MagicMock()


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