# 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__, '', 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()