"""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 = support.threading_setup()

logger_dict = logging.getLogger().manager.loggerDict

logging._acquireLock()

try:

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)

finally:

logging._releaseLock()

# 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)

logging._acquireLock()

try:

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]

finally:

logging._releaseLock()

self.doCleanups()

support.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_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), '')

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_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', 'darwin'):

for existing in (True, False):

fd, fn = tempfile.mkstemp()

os.close(fd)

if not existing:

os.unlink(fn)

h = logging.handlers.WatchedFileHandler(fn, 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_path_objects(self):

"""

fd, fn = tempfile.mkstemp()

os.close(fd)

os.unlink(fn)

pfn = pathlib.Path(fn)

cases = (

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

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

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

)

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

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

for cls, args in cases:

h = cls(*args)

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

h.close()

os.unlink(fn)

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

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):

fd, fn = tempfile.mkstemp('.log', 'test_logging-3-')

os.close(fd)

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

remover.daemon = True

remover.start()

h = logging.handlers.WatchedFileHandler(fn, 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):

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)

"""

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.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.setDaemon(True)

t.start()

def serve_forever(self, poll_interval):

"""

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

def stop(self, timeout=None):

"""

self.close()

support.join_thread(self._thread, timeout)

self._thread = None

"""

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.setDaemon(True)

t.start()

def serve_forever(self, poll_interval):

"""

self.ready.set()

super(ControlMixin, self).serve_forever(poll_interval)

def stop(self, timeout=None):

"""

self.shutdown()

if self._thread is not None:

support.join_thread(self._thread, timeout)

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

raise AttributeError(name)

def process_request(self):

self.server._handler(self)

def log_message(self, format, *args):

if log:

super(DelegatingHTTPRequestHandler,

self).log_message(format, *args)

HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)

ControlMixin.__init__(self, handler, poll_interval)

self.sslctx = sslctx

def get_request(self):

try:

sock, addr = self.socket.accept()

if self.sslctx:

sock = self.sslctx.wrap_socket(sock, server_side=True)

except OSError as e:

# socket errors are silenced by the caller, print them here

sys.stderr.write("Got an error:\n%s\n" % e)

raise

"""

allow_reuse_address = True

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

bind_and_activate=True):

class DelegatingTCPRequestHandler(StreamRequestHandler):

def handle(self):

self.server._handler(self)

ThreadingTCPServer.__init__(self, addr, DelegatingTCPRequestHandler,

bind_and_activate)

ControlMixin.__init__(self, handler, poll_interval)

def server_bind(self):

super(TestTCPServer, self).server_bind()

"""

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

bind_and_activate=True):

class DelegatingUDPRequestHandler(DatagramRequestHandler):

def handle(self):

self.server._handler(self)

def finish(self):

data = self.wfile.getvalue()

if data:

try:

super(DelegatingUDPRequestHandler, self).finish()

except OSError:

if not self.server._closed:

raise

ThreadingUDPServer.__init__(self, addr,

DelegatingUDPRequestHandler,

bind_and_activate)

ControlMixin.__init__(self, handler, poll_interval)

self._closed = False

def server_bind(self):

super(TestUDPServer, self).server_bind()

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

def server_close(self):

super(TestUDPServer, self).server_close()

# bpo-14314, bpo-19665, bpo-34092: don't wait forever, timeout of 1 minute

TIMEOUT = 60.0

def test_basic(self):

sockmap = {}

server = TestSMTPServer((support.HOST, 0), self.process_message, 0.001,

sockmap)

server.start()

addr = (support.HOST, server.port)

h = logging.handlers.SMTPHandler(addr, 'me', 'you', 'Log',

timeout=self.TIMEOUT)

self.assertEqual(h.toaddrs, ['you'])

self.messages = []

r = logging.makeLogRecord({'msg': 'Hello \u2713'})

self.handled = threading.Event()

h.handle(r)

self.handled.wait(self.TIMEOUT)

server.stop()

self.assertTrue(self.handled.is_set())

self.assertEqual(len(self.messages), 1)

peer, mailfrom, rcpttos, data = self.messages[0]

self.assertEqual(mailfrom, 'me')

self.assertEqual(rcpttos, ['you'])

self.assertIn('\nSubject: Log\n', data)

self.assertTrue(data.endswith('\n\nHello \u2713'))

h.close()