"""Base class for logging tests."""

log_format = "%(name)s -> %(levelname)s: %(message)s"

expected_log_pat = r"^([\w.]+) -> (\w+): (\d+)$"

message_num = 0

def setUp(self):

"""Setup the default logging stream to an internal StringIO instance,

self._threading_key = threading_helper.threading_setup()

logger_dict = logging.getLogger().manager.loggerDict

with logging._lock:

self.saved_handlers = logging._handlers.copy()

self.saved_handler_list = logging._handlerList[:]

self.saved_loggers = saved_loggers = logger_dict.copy()

self.saved_name_to_level = logging._nameToLevel.copy()

self.saved_level_to_name = logging._levelToName.copy()

self.logger_states = logger_states = {}

for name in saved_loggers:

logger_states[name] = getattr(saved_loggers[name],

'disabled', None)

# Set two unused loggers

self.logger1 = logging.getLogger("\xab\xd7\xbb")

self.logger2 = logging.getLogger("\u013f\u00d6\u0047")

self.root_logger = logging.getLogger("")

self.original_logging_level = self.root_logger.getEffectiveLevel()

self.stream = io.StringIO()

self.root_logger.setLevel(logging.DEBUG)

self.root_hdlr = logging.StreamHandler(self.stream)

self.root_formatter = logging.Formatter(self.log_format)

self.root_hdlr.setFormatter(self.root_formatter)

if self.logger1.hasHandlers():

hlist = self.logger1.handlers + self.root_logger.handlers

raise AssertionError('Unexpected handlers: %s' % hlist)

if self.logger2.hasHandlers():

hlist = self.logger2.handlers + self.root_logger.handlers

raise AssertionError('Unexpected handlers: %s' % hlist)

self.root_logger.addHandler(self.root_hdlr)

self.assertTrue(self.logger1.hasHandlers())

self.assertTrue(self.logger2.hasHandlers())

def tearDown(self):

"""Remove our logging stream, and restore the original logging

self.stream.close()

self.root_logger.removeHandler(self.root_hdlr)

while self.root_logger.handlers:

h = self.root_logger.handlers[0]

self.root_logger.removeHandler(h)

h.close()

self.root_logger.setLevel(self.original_logging_level)

with logging._lock:

logging._levelToName.clear()

logging._levelToName.update(self.saved_level_to_name)

logging._nameToLevel.clear()

logging._nameToLevel.update(self.saved_name_to_level)

logging._handlers.clear()

logging._handlers.update(self.saved_handlers)

logging._handlerList[:] = self.saved_handler_list

manager = logging.getLogger().manager

manager.disable = 0

loggerDict = manager.loggerDict

loggerDict.clear()

loggerDict.update(self.saved_loggers)

logger_states = self.logger_states

for name in self.logger_states:

if logger_states[name] is not None:

self.saved_loggers[name].disabled = logger_states[name]

self.doCleanups()

threading_helper.threading_cleanup(*self._threading_key)

def assert_log_lines(self, expected_values, stream=None, pat=None):

"""Match the collected log lines against the regular expression

stream = stream or self.stream

pat = re.compile(pat or self.expected_log_pat)

actual_lines = stream.getvalue().splitlines()

self.assertEqual(len(actual_lines), len(expected_values))

for actual, expected in zip(actual_lines, expected_values):

match = pat.search(actual)

if not match:

self.fail("Log line does not match expected pattern:\n" +

actual)

self.assertEqual(tuple(match.groups()), expected)

s = stream.read()

if s:

self.fail("Remaining output at end of log stream:\n" + s)

def next_message(self):

"""Generate a message consisting solely of an auto-incrementing

self.message_num += 1

"""Test builtin levels and their inheritance."""

def test_flat(self):

# Logging levels in a flat logger namespace.

m = self.next_message

ERR = logging.getLogger("ERR")

ERR.setLevel(logging.ERROR)

INF = logging.LoggerAdapter(logging.getLogger("INF"), {})

INF.setLevel(logging.INFO)

DEB = logging.getLogger("DEB")

DEB.setLevel(logging.DEBUG)

# These should log.

ERR.log(logging.CRITICAL, m())

ERR.error(m())

INF.log(logging.CRITICAL, m())

INF.error(m())

INF.warning(m())

INF.info(m())

DEB.log(logging.CRITICAL, m())

DEB.error(m())

DEB.warning(m())

DEB.info(m())

DEB.debug(m())

# These should not log.

ERR.warning(m())

ERR.info(m())

ERR.debug(m())

INF.debug(m())

self.assert_log_lines([

('ERR', 'CRITICAL', '1'),

('ERR', 'ERROR', '2'),

('INF', 'CRITICAL', '3'),

('INF', 'ERROR', '4'),

('INF', 'WARNING', '5'),

('INF', 'INFO', '6'),

('DEB', 'CRITICAL', '7'),

('DEB', 'ERROR', '8'),

('DEB', 'WARNING', '9'),

('DEB', 'INFO', '10'),

('DEB', 'DEBUG', '11'),

])

def test_nested_explicit(self):

# Logging levels in a nested namespace, all explicitly set.

m = self.next_message

INF = logging.getLogger("INF")

INF.setLevel(logging.INFO)

INF_ERR = logging.getLogger("INF.ERR")

INF_ERR.setLevel(logging.ERROR)

# These should log.

INF_ERR.log(logging.CRITICAL, m())

INF_ERR.error(m())

# These should not log.

INF_ERR.warning(m())

INF_ERR.info(m())

INF_ERR.debug(m())

self.assert_log_lines([

('INF.ERR', 'CRITICAL', '1'),

('INF.ERR', 'ERROR', '2'),

])

def test_nested_inherited(self):

# Logging levels in a nested namespace, inherited from parent loggers.

m = self.next_message

INF = logging.getLogger("INF")

INF.setLevel(logging.INFO)

INF_ERR = logging.getLogger("INF.ERR")

INF_ERR.setLevel(logging.ERROR)

INF_UNDEF = logging.getLogger("INF.UNDEF")

INF_ERR_UNDEF = logging.getLogger("INF.ERR.UNDEF")

UNDEF = logging.getLogger("UNDEF")

# These should log.

INF_UNDEF.log(logging.CRITICAL, m())

INF_UNDEF.error(m())

INF_UNDEF.warning(m())

INF_UNDEF.info(m())

INF_ERR_UNDEF.log(logging.CRITICAL, m())

INF_ERR_UNDEF.error(m())

# These should not log.

INF_UNDEF.debug(m())

INF_ERR_UNDEF.warning(m())

INF_ERR_UNDEF.info(m())

INF_ERR_UNDEF.debug(m())

self.assert_log_lines([

('INF.UNDEF', 'CRITICAL', '1'),

('INF.UNDEF', 'ERROR', '2'),

('INF.UNDEF', 'WARNING', '3'),

('INF.UNDEF', 'INFO', '4'),

('INF.ERR.UNDEF', 'CRITICAL', '5'),

('INF.ERR.UNDEF', 'ERROR', '6'),

])

def test_nested_with_virtual_parent(self):

# Logging levels when some parent does not exist yet.

m = self.next_message

INF = logging.getLogger("INF")

GRANDCHILD = logging.getLogger("INF.BADPARENT.UNDEF")

CHILD = logging.getLogger("INF.BADPARENT")

INF.setLevel(logging.INFO)

# These should log.

GRANDCHILD.log(logging.FATAL, m())

GRANDCHILD.info(m())

CHILD.log(logging.FATAL, m())

CHILD.info(m())

# These should not log.

GRANDCHILD.debug(m())

CHILD.debug(m())

self.assert_log_lines([

('INF.BADPARENT.UNDEF', 'CRITICAL', '1'),

('INF.BADPARENT.UNDEF', 'INFO', '2'),

('INF.BADPARENT', 'CRITICAL', '3'),

('INF.BADPARENT', 'INFO', '4'),

])

def test_regression_22386(self):

"""See issue #22386 for more information."""

self.assertEqual(logging.getLevelName('INFO'), logging.INFO)

self.assertEqual(logging.getLevelName(logging.INFO), 'INFO')

def test_issue27935(self):

fatal = logging.getLevelName('FATAL')

self.assertEqual(fatal, logging.FATAL)

def test_regression_29220(self):

"""See issue #29220 for more information."""

logging.addLevelName(logging.INFO, '')

self.addCleanup(logging.addLevelName, logging.INFO, 'INFO')

self.assertEqual(logging.getLevelName(logging.INFO), '')

self.assertEqual(logging.getLevelName(logging.NOTSET), 'NOTSET')

"""Test the bundled Filter class."""

def test_filter(self):

# Only messages satisfying the specified criteria pass through the

# filter.

filter_ = logging.Filter("spam.eggs")

handler = self.root_logger.handlers[0]

try:

handler.addFilter(filter_)

spam = logging.getLogger("spam")

spam_eggs = logging.getLogger("spam.eggs")

spam_eggs_fish = logging.getLogger("spam.eggs.fish")

spam_bakedbeans = logging.getLogger("spam.bakedbeans")

spam.info(self.next_message())

spam_eggs.info(self.next_message()) # Good.

spam_eggs_fish.info(self.next_message()) # Good.

spam_bakedbeans.info(self.next_message())

self.assert_log_lines([

('spam.eggs', 'INFO', '2'),

('spam.eggs.fish', 'INFO', '3'),

])

finally:

handler.removeFilter(filter_)

def test_callable_filter(self):

# Only messages satisfying the specified criteria pass through the

# filter.

def filterfunc(record):

parts = record.name.split('.')

prefix = '.'.join(parts[:2])

return prefix == 'spam.eggs'

handler = self.root_logger.handlers[0]

try:

handler.addFilter(filterfunc)

spam = logging.getLogger("spam")

spam_eggs = logging.getLogger("spam.eggs")

spam_eggs_fish = logging.getLogger("spam.eggs.fish")

spam_bakedbeans = logging.getLogger("spam.bakedbeans")

spam.info(self.next_message())

spam_eggs.info(self.next_message()) # Good.

spam_eggs_fish.info(self.next_message()) # Good.

spam_bakedbeans.info(self.next_message())

self.assert_log_lines([

('spam.eggs', 'INFO', '2'),

('spam.eggs.fish', 'INFO', '3'),

])

finally:

handler.removeFilter(filterfunc)

def test_empty_filter(self):

f = logging.Filter()

r = logging.makeLogRecord({'name': 'spam.eggs'})

SILENT : 'Silent',

TACITURN : 'Taciturn',

TERSE : 'Terse',

EFFUSIVE : 'Effusive',

SOCIABLE : 'Sociable',

VERBOSE : 'Verbose',

TALKATIVE : 'Talkative',

GARRULOUS : 'Garrulous',

CHATTERBOX : 'Chatterbox',

BORING : 'Boring',

"""A filter which blocks garrulous messages."""

def filter(self, record):

"""A filter which blocks sociable and taciturn messages."""

def filter(self, record):

"""Test various filtering possibilities with custom logging levels."""

# Skip the logger name group.

expected_log_pat = r"^[\w.]+ -> (\w+): (\d+)$"

def setUp(self):

BaseTest.setUp(self)

for k, v in my_logging_levels.items():

logging.addLevelName(k, v)

def log_at_all_levels(self, logger):

for lvl in LEVEL_RANGE:

logger.log(lvl, self.next_message())

def test_handler_filter_replaces_record(self):

def replace_message(record: logging.LogRecord):

record = copy.copy(record)

record.msg = "new message!"

return record

# Set up a logging hierarchy such that "child" and it's handler

# (and thus `replace_message()`) always get called before

# propagating up to "parent".

# Then we can confirm that `replace_message()` was able to

# replace the log record without having a side effect on

# other loggers or handlers.

parent = logging.getLogger("parent")

child = logging.getLogger("parent.child")

stream_1 = io.StringIO()

stream_2 = io.StringIO()

handler_1 = logging.StreamHandler(stream_1)

handler_2 = logging.StreamHandler(stream_2)

handler_2.addFilter(replace_message)

parent.addHandler(handler_1)

child.addHandler(handler_2)

child.info("original message")

handler_1.flush()

handler_2.flush()

self.assertEqual(stream_1.getvalue(), "original message\n")

self.assertEqual(stream_2.getvalue(), "new message!\n")

def test_logging_filter_replaces_record(self):

records = set()

class RecordingFilter(logging.Filter):

def filter(self, record: logging.LogRecord):

records.add(id(record))

return copy.copy(record)

logger = logging.getLogger("logger")

logger.setLevel(logging.INFO)

logger.addFilter(RecordingFilter())

logger.addFilter(RecordingFilter())

logger.info("msg")

self.assertEqual(2, len(records))

def test_logger_filter(self):

# Filter at logger level.

self.root_logger.setLevel(VERBOSE)

# Levels >= 'Verbose' are good.

self.log_at_all_levels(self.root_logger)

self.assert_log_lines([

('Verbose', '5'),

('Sociable', '6'),

('Effusive', '7'),

('Terse', '8'),

('Taciturn', '9'),

('Silent', '10'),

])

def test_handler_filter(self):

# Filter at handler level.

self.root_logger.handlers[0].setLevel(SOCIABLE)

try:

# Levels >= 'Sociable' are good.

self.log_at_all_levels(self.root_logger)

self.assert_log_lines([

('Sociable', '6'),

('Effusive', '7'),

('Terse', '8'),

('Taciturn', '9'),

('Silent', '10'),

])

finally:

self.root_logger.handlers[0].setLevel(logging.NOTSET)

def test_specific_filters(self):

# Set a specific filter object on the handler, and then add another

# filter object on the logger itself.

handler = self.root_logger.handlers[0]

specific_filter = None

garr = GarrulousFilter()

handler.addFilter(garr)

try:

self.log_at_all_levels(self.root_logger)

first_lines = [

# Notice how 'Garrulous' is missing

('Boring', '1'),

('Chatterbox', '2'),

('Talkative', '4'),

('Verbose', '5'),

('Sociable', '6'),

('Effusive', '7'),

('Terse', '8'),

('Taciturn', '9'),

('Silent', '10'),

]

self.assert_log_lines(first_lines)

specific_filter = VerySpecificFilter()

self.root_logger.addFilter(specific_filter)

self.log_at_all_levels(self.root_logger)

self.assert_log_lines(first_lines + [

# Not only 'Garrulous' is still missing, but also 'Sociable'

# and 'Taciturn'

('Boring', '11'),

('Chatterbox', '12'),

('Talkative', '14'),

('Verbose', '15'),

('Effusive', '17'),

('Terse', '18'),

('Silent', '20'),

])

finally:

if specific_filter:

self.root_logger.removeFilter(specific_filter)

def test_name(self):

h = logging.Handler()

h.name = 'generic'

self.assertEqual(h.name, 'generic')

h.name = 'anothergeneric'

self.assertEqual(h.name, 'anothergeneric')

self.assertRaises(NotImplementedError, h.emit, None)

def test_builtin_handlers(self):

# We can't actually *use* too many handlers in the tests,

# but we can try instantiating them with various options

if sys.platform in ('linux', 'android', 'darwin'):

for existing in (True, False):

fn = make_temp_file()

if not existing:

os.unlink(fn)

h = logging.handlers.WatchedFileHandler(fn, encoding='utf-8', delay=True)

if existing:

dev, ino = h.dev, h.ino

self.assertEqual(dev, -1)

self.assertEqual(ino, -1)

r = logging.makeLogRecord({'msg': 'Test'})

h.handle(r)

# Now remove the file.

os.unlink(fn)

self.assertFalse(os.path.exists(fn))

# The next call should recreate the file.

h.handle(r)

self.assertTrue(os.path.exists(fn))

else:

self.assertEqual(h.dev, -1)

self.assertEqual(h.ino, -1)

h.close()

if existing:

os.unlink(fn)

if sys.platform == 'darwin':

sockname = '/var/run/syslog'

else:

sockname = '/dev/log'

try:

h = logging.handlers.SysLogHandler(sockname)

self.assertEqual(h.facility, h.LOG_USER)

self.assertTrue(h.unixsocket)

h.close()

except OSError: # syslogd might not be available

pass

for method in ('GET', 'POST', 'PUT'):

if method == 'PUT':

self.assertRaises(ValueError, logging.handlers.HTTPHandler,

'localhost', '/log', method)

else:

h = logging.handlers.HTTPHandler('localhost', '/log', method)

h.close()

h = logging.handlers.BufferingHandler(0)

r = logging.makeLogRecord({})

self.assertTrue(h.shouldFlush(r))

h.close()

h = logging.handlers.BufferingHandler(1)

self.assertFalse(h.shouldFlush(r))

h.close()

def test_pathlike_objects(self):

"""

fn = make_temp_file()

os.unlink(fn)

pfn = os_helper.FakePath(fn)

cases = (

(logging.FileHandler, (pfn, 'w')),

(logging.handlers.RotatingFileHandler, (pfn, 'a')),

(logging.handlers.TimedRotatingFileHandler, (pfn, 'h')),

)

if sys.platform in ('linux', 'android', 'darwin'):

cases += ((logging.handlers.WatchedFileHandler, (pfn, 'w')),)

for cls, args in cases:

h = cls(*args, encoding="utf-8")

self.assertTrue(os.path.exists(fn))

h.close()

os.unlink(fn)

@unittest.skipIf(os.name == 'nt', 'WatchedFileHandler not appropriate for Windows.')

@unittest.skipIf(

@threading_helper.requires_working_threading()

@support.requires_resource('walltime')

def test_race(self):

# Issue #14632 refers.

def remove_loop(fname, tries):

for _ in range(tries):

try:

os.unlink(fname)

self.deletion_time = time.time()

except OSError:

pass

time.sleep(0.004 * random.randint(0, 4))

del_count = 500

log_count = 500

self.handle_time = None

self.deletion_time = None

for delay in (False, True):

fn = make_temp_file('.log', 'test_logging-3-')

remover = threading.Thread(target=remove_loop, args=(fn, del_count))

remover.daemon = True

remover.start()

h = logging.handlers.WatchedFileHandler(fn, encoding='utf-8', delay=delay)

f = logging.Formatter('%(asctime)s: %(levelname)s: %(message)s')

h.setFormatter(f)

try:

for _ in range(log_count):

time.sleep(0.005)

r = logging.makeLogRecord({'msg': 'testing' })

try:

self.handle_time = time.time()

h.handle(r)

except Exception:

print('Deleted at %s, '

'opened at %s' % (self.deletion_time,

self.handle_time))

raise

finally:

remover.join()

h.close()

if os.path.exists(fn):

os.unlink(fn)

# The implementation relies on os.register_at_fork existing, but we test

# based on os.fork existing because that is what users and this test use.

# This helps ensure that when fork exists (the important concept) that the

# register_at_fork mechanism is also present and used.

@support.requires_fork()

@threading_helper.requires_working_threading()

@skip_if_asan_fork

@skip_if_tsan_fork

def test_post_fork_child_no_deadlock(self):

"""Ensure child logging locks are not held; bpo-6721 & bpo-36533."""

class _OurHandler(logging.Handler):

def __init__(self):

super().__init__()

self.sub_handler = logging.StreamHandler(

stream=open('/dev/null', 'wt', encoding='utf-8'))

def emit(self, record):

with self.sub_handler.lock:

self.sub_handler.emit(record)

self.assertEqual(len(logging._handlers), 0)

refed_h = _OurHandler()

self.addCleanup(refed_h.sub_handler.stream.close)

refed_h.name = 'because we need at least one for this test'

self.assertGreater(len(logging._handlers), 0)

self.assertGreater(len(logging._at_fork_reinit_lock_weakset), 1)

test_logger = logging.getLogger('test_post_fork_child_no_deadlock')

test_logger.addHandler(refed_h)

test_logger.setLevel(logging.DEBUG)

locks_held__ready_to_fork = threading.Event()

fork_happened__release_locks_and_end_thread = threading.Event()

def lock_holder_thread_fn():

with logging._lock, refed_h.lock:

# Tell the main thread to do the fork.

locks_held__ready_to_fork.set()

# If the deadlock bug exists, the fork will happen

# without dealing with the locks we hold, deadlocking

# the child.

# Wait for a successful fork or an unreasonable amount of

# time before releasing our locks. To avoid a timing based

# test we'd need communication from os.fork() as to when it

# has actually happened. Given this is a regression test

# for a fixed issue, potentially less reliably detecting

# regression via timing is acceptable for simplicity.

# The test will always take at least this long. :(

fork_happened__release_locks_and_end_thread.wait(0.5)

lock_holder_thread = threading.Thread(

target=lock_holder_thread_fn,

name='test_post_fork_child_no_deadlock lock holder')

lock_holder_thread.start()

locks_held__ready_to_fork.wait()

pid = os.fork()

if pid == 0:

# Child process

try:

test_logger.info(r'Child process did not deadlock. \o/')

finally:

os._exit(0)

else:

# Parent process

test_logger.info(r'Parent process returned from fork. \o/')

fork_happened__release_locks_and_end_thread.set()

lock_holder_thread.join()

def test_error_handling(self):

h = TestStreamHandler(BadStream())

r = logging.makeLogRecord({})

old_raise = logging.raiseExceptions

try:

h.handle(r)

self.assertIs(h.error_record, r)

h = logging.StreamHandler(BadStream())

with support.captured_stderr() as stderr:

h.handle(r)

msg = '\nRuntimeError: deliberate mistake\n'

self.assertIn(msg, stderr.getvalue())

logging.raiseExceptions = False

with support.captured_stderr() as stderr:

h.handle(r)

self.assertEqual('', stderr.getvalue())

finally:

logging.raiseExceptions = old_raise

def test_stream_setting(self):

"""

h = logging.StreamHandler()

stream = io.StringIO()

old = h.setStream(stream)

self.assertIs(old, sys.stderr)

actual = h.setStream(old)

self.assertIs(actual, stream)

# test that setting to existing value returns None

actual = h.setStream(old)

self.assertIsNone(actual)

def test_can_represent_stream_with_int_name(self):

h = logging.StreamHandler(StreamWithIntName())

"""

def __init__(self, addr, handler, poll_interval, sockmap):

smtpd.SMTPServer.__init__(self, addr, None, map=sockmap,

decode_data=True)

self.port = self.socket.getsockname()[1]

self._handler = handler

self._thread = None

self._quit = False

self.poll_interval = poll_interval

def process_message(self, peer, mailfrom, rcpttos, data):

"""

self._handler(peer, mailfrom, rcpttos, data)

def start(self):

"""

self._thread = t = threading.Thread(target=self.serve_forever,

args=(self.poll_interval,))

t.daemon = True

t.start()

def serve_forever(self, poll_interval):

"""

while not self._quit:

asyncore.loop(poll_interval, map=self._map, count=1)

def stop(self):

"""

self._quit = True

threading_helper.join_thread(self._thread)

self._thread = None

self.close()

"""

def __init__(self, handler, poll_interval):

self._thread = None

self.poll_interval = poll_interval

self._handler = handler

self.ready = threading.Event()

def start(self):

"""

self._thread = t = threading.Thread(target=self.serve_forever,

args=(self.poll_interval,))

t.daemon = True

t.start()

def serve_forever(self, poll_interval):

"""

self.ready.set()

super(ControlMixin, self).serve_forever(poll_interval)

def stop(self):

"""

self.shutdown()

if self._thread is not None:

threading_helper.join_thread(self._thread)

self._thread = None

self.server_close()

"""

def __init__(self, addr, handler, poll_interval=0.5,

log=False, sslctx=None):

class DelegatingHTTPRequestHandler(BaseHTTPRequestHandler):

def __getattr__(self, name, default=None):

if name.startswith('do_'):

return self.process_request