Python logging 模块，makeLogRecord() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用logging.makeLogRecord()。
def _handle_logging(self, level, message):
"""
Handles an RPC logging request.
:param str level: One of "debug", "info", "warning", or "error".
:param str message: The log message.
"""
# manually create a record to log to the standard file handler.
# we format it to match the regular logs, but tack on the '.js' to
# indicate that it came from javascript.
record = logging.makeLogRecord({
"levelname": level.upper(),
"name": "%s.js" % (self.logger.name,),
"msg": message,
})
# forward this message to the base file handler so that it is logged
# appropriately.
if sgtk.LogManager().base_file_handler:
sgtk.LogManager().base_file_handler.handle(record)
def handle(self):
'''
Handle multiple requests - each expected to be a 4-byte length,
followed by the LogRecord in pickle format.
'''
while 1:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack('>L', chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unPickle(chunk)
record = logging.makeLogRecord(obj)
self.handleLogRecord(record)
def test_multiline_formatter():
import logging
from ldap2pg.config import MultilineFormatter
formatter = MultilineFormatter("prefix: %(message)s")
base_record = dict(
name='pouet', level=logging.DEBUG, fn="(unknown file)", lno=0, args=(),
exc_info=None,
)
record = logging.makeLogRecord(dict(base_record, msg="single line"))
payload = formatter.format(record)
assert "prefix: single line" == payload
record = logging.makeLogRecord(dict(base_record, msg="Uno\nDos\nTres"))
payload = formatter.format(record)
wanted = """\
prefix: Uno
prefix: Dos
prefix: Tres\
""".replace(' ', '')
assert wanted == payload
def read_logs(self):
try:
while True:
datagram = self.socket.recv(8192)
chunk = datagram[0:4]
struct.unpack(">L", chunk)[0]
chunk = datagram[4:]
obj = cPickle.loads(chunk)
record = logging.makeLogRecord(obj)
if (record.levelno >= self.level):
logger = logging.getLogger(record.name)
logger.handle(record)
except Exception as e:
print "ERROR: " + str(e)
finally:
self.socket.close()
def test_logger_format(self):
"""Test: CONFIG: logger format."""
config = SRBConfig(logger)
config.logger_add_stream_handler()
config.logger_add_file_handler('logs.log')
fmt = '%(name)s - %(levelname)s - %(message)s'
formatter = logging.Formatter(fmt)
config.logger_format = fmt
tst_record = {
'name': 'test_logger',
'level': logging.DEBUG,
'pathname': os.path.realpath(__file__),
'lineno': 42,
'msg': 'test_msg',
'args': None,
'exc_info': None,
'func': 'test_logger_format'
}
rec = logging.makeLogRecord(tst_record)
self.assertEqual(formatter.format(rec),
config.logger_stream_handler.format(rec))
def handle(self):
"""Handle multiple requests - each expected to be a 4-byte length,
followed by the LogRecord in pickle format. Logs the record
according to whatever policy is configured locally.
"""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack('>L', chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unpickle(chunk)
record = logging.makeLogRecord(obj)
self.handle_log_record(record)
def test_error_handling(self):
h = TestStreamHandler(BadStream())
r = logging.makeLogRecord({})
old_raise = logging.raiseExceptions
old_stderr = sys.stderr
try:
h.handle(r)
self.assertIs(h.error_record, r)
h = logging.StreamHandler(BadStream())
sys.stderr = sio = io.StringIO()
h.handle(r)
self.assertIn('\nRuntimeError: deliberate mistake\n',
sio.getvalue())
logging.raiseExceptions = False
sys.stderr = sio = io.StringIO()
h.handle(r)
self.assertEqual('', sio.getvalue())
finally:
logging.raiseExceptions = old_raise
sys.stderr = old_stderr
# -- The following section could be moved into a server_helper.py module
# -- if it proves to be of wider utility than just test_logging
def test_basic(self):
sockmap = {}
server = TestSMTPServer(('localhost', 0), self.process_message, 0.001,
sockmap)
server.start()
addr = ('localhost', server.port)
h = logging.handlers.SMTPHandler(addr, 'me', 'you', 'Log', timeout=5.0)
self.assertEqual(h.toaddrs, ['you'])
self.messages = []
r = logging.makeLogRecord({'msg': 'Hello'})
self.handled = threading.Event()
h.handle(r)
self.handled.wait(5.0) # 14314: don't wait forever
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'))
h.close()
def test_error_handling(self):
h = TestStreamHandler(BadStream())
r = logging.makeLogRecord({})
old_raise = logging.raiseExceptions
old_stderr = sys.stderr
try:
h.handle(r)
self.assertIs(h.error_record, r)
h = logging.StreamHandler(BadStream())
sys.stderr = sio = io.StringIO()
h.handle(r)
self.assertIn('\nRuntimeError: deliberate mistake\n',
sio.getvalue())
logging.raiseExceptions = False
sys.stderr = sio = io.StringIO()
h.handle(r)
self.assertEqual('', sio.getvalue())
finally:
logging.raiseExceptions = old_raise
sys.stderr = old_stderr
# -- The following section could be moved into a server_helper.py module
# -- if it proves to be of wider utility than just test_logging
def test_basic(self):
sockmap = {}
server = TestSMTPServer((HOST, 0), self.process_message, 0.001,
sockmap)
server.start()
addr = (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) # 14314: don't wait forever
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()
def handle(self):
"""
Handle multiple requests - each expected to be a 4-byte length,
followed by the LogRecord in pickle format. Logs the record
according to whatever policy is configured locally.
"""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack('>L', chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unPickle(chunk)
record = logging.makeLogRecord(obj)
self.handleLogRecord(record)
def test_error_handling(self):
h = TestStreamHandler(BadStream())
r = logging.makeLogRecord({})
old_raise = logging.raiseExceptions
old_stderr = sys.stderr
try:
h.handle(r)
self.assertIs(h.error_record, r)
h = logging.StreamHandler(BadStream())
sys.stderr = sio = io.StringIO()
h.handle(r)
self.assertIn('\nRuntimeError: deliberate mistake\n',
sio.getvalue())
logging.raiseExceptions = False
sys.stderr = sio = io.StringIO()
h.handle(r)
self.assertEqual('', sio.getvalue())
finally:
logging.raiseExceptions = old_raise
sys.stderr = old_stderr
# -- The following section could be moved into a server_helper.py module
# -- if it proves to be of wider utility than just test_logging
def test_basic(self):
sockmap = {}
server = TestSMTPServer(('localhost', 0), self.process_message, 0.001,
sockmap)
server.start()
addr = ('localhost', 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'})
self.handled = threading.Event()
h.handle(r)
self.handled.wait(self.TIMEOUT) # 14314: don't wait forever
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'))
h.close()
def test_color_handler():
import logging
from ldap2pg.config import ColoredStreamHandler
handler = ColoredStreamHandler()
record = logging.makeLogRecord(dict(
name='pouet', level=logging.DEBUG, fn="(unknown file)", msg="Message",
lno=0, args=(), exc_info=None,
))
payload = handler.format(record)
assert "\033[0" in payload
def update(self,frame,events):
if self._socket:
while self._socket.new_data:
t,s = self._socket.recv()
self.on_log(logging.makeLogRecord(s))
def recent_events(self,events):
if self._socket and self._socket.new_data:
t,s = self._socket.recv()
self.on_log(logging.makeLogRecord(s))
self.alpha -= min(.2,events['dt'])
def test_emitReturnsLoggedMessageId(self):
"""
MDKHandler.emit returns the LoggedMessageId from the Session.
"""
mdk, tracer = create_mdk_with_faketracer()
session = mdk.session()
handler = MDKHandler(mdk, lambda: session)
record = logging.makeLogRecord({"name": "", "levelname": "INFO",
"message": "hello"})
mid = handler.emit(record)
self.assertIsInstance(mid, LoggedMessageId)
self.assertEqual(mid.traceId, session._context.traceId)
def test_emitReturnsLoggedMessageId(self):
"""
MDKLoggingHandler.emit returns the LoggedMessageId.
"""
mdk, tracer = create_mdk_with_faketracer()
handler = MDKLoggingHandler(mdk)
record = logging.makeLogRecord({"name": "", "levelname": "INFO",
"message": "hello"})
mid = handler.emit(record)
self.assertIsInstance(mid, LoggedMessageId)
def handle(self):
"""Deal with the incoming log data"""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
struct_len = struct.unpack('>L', chunk)[0]
chunk = self.connection.recv(struct_len)
while len(chunk) < struct_len:
chunk = chunk + self.connection.recv(struct_len - len(chunk))
obj = self.unpickle(chunk)
record = logging.makeLogRecord(obj)
self.handle_log_record(record)
def run(self):
"""Run loop.
Receives log messages from connected publishers and logs them via
a python logging interface.
"""
log = logging.getLogger('sip.logging_aggregator')
fail_count = 0
fail_count_limit = 100
# Exponential relaxation of timeout in event loop.
timeout = np.logspace(-6, -2, fail_count_limit)
while not self._stop_requested.is_set():
try:
topic, values = self._subscriber.recv_multipart(zmq.NOBLOCK)
str_values = values.decode('utf-8')
try:
dict_values = json.loads(str_values)
record = logging.makeLogRecord(dict_values)
log.handle(record)
fail_count = 0
except json.decoder.JSONDecodeError:
print('ERROR: Unable to convert JSON log record.')
raise
except zmq.ZMQError as e:
if e.errno == zmq.EAGAIN:
fail_count += 1
else:
raise # Re-raise the exception
if fail_count < fail_count_limit:
_timeout = timeout[fail_count]
else:
_timeout = timeout[-1]
self._stop_requested.wait(_timeout)
def handle(self):
"""Handle multiple requests - each expected to be of 4-byte length,
followed by the LogRecord in pickle format. Logs the record
according to whatever policy is configured locally."""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack(">L", chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unpickle(chunk)
record = logging.makeLogRecord(obj)
self.handle_log_record(record)
def get_record(self, name=None):
result = dict(self.common)
if name is not None:
result.update(self.variants[name])
return logging.makeLogRecord(result)
def test_html_formatter_format():
formatter = formatters.HtmlFormatter()
logrecord = logging.makeLogRecord(dict(name='<foo>', func='<module>', msg='Whatsup?', funcName='test'))
s = formatter.format(logrecord)
assert '<foo>' not in s
assert '<module>' not in s
def test_html_formatter_emoji():
formatter = formatters.HtmlFormatter(use_emoji=True)
emoji_level_map = {
formatters.EMOJI.WHITE_CIRCLE: [logging.DEBUG],
formatters.EMOJI.BLUE_CIRCLE: [logging.INFO],
formatters.EMOJI.RED_CIRCLE: [logging.WARNING, logging.ERROR]
}
for emoji, levels in emoji_level_map.items():
for level in levels:
logrecord = logging.makeLogRecord({'levelno': level, 'levelname': logging.getLevelName(level)})
s = formatter.format(logrecord)
assert s.find(emoji) > 0, 'Emoji not found in %s' % level
def test_emit(handler):
record = logging.makeLogRecord({'msg': 'hello'})
with mock.patch('requests.post') as patch:
handler.emit(record)
assert patch.called
assert patch.call_count == 1
assert patch.call_args[1]['json']['chat_id'] == 'bar'
assert 'hello' in patch.call_args[1]['json']['text']
assert patch.call_args[1]['json']['parse_mode'] == 'HTML'
def test_emit_http_exception(handler):
record = logging.makeLogRecord({'msg': 'hello'})
with mock.patch('requests.post') as patch:
response = requests.Response()
response.status_code = 500
response._content = 'Server error'.encode()
patch.return_value = response
handler.emit(record)
assert telegram_handler.handlers.logger.handlers[0].messages['error']
assert telegram_handler.handlers.logger.handlers[0].messages['debug']
def test_emit_telegram_error(handler):
record = logging.makeLogRecord({'msg': 'hello'})
with mock.patch('requests.post') as patch:
response = requests.Response()
response.status_code = 200
response._content = json.dumps({'ok': False}).encode()
patch.return_value = response
handler.emit(record)
assert telegram_handler.handlers.logger.handlers[0].messages['warning']
def test_adds_padding(self):
self.handler.emit(logging.makeLogRecord({}))
self.d.infobox.assert_called_once_with(mock.ANY, 4, 10)
def test_args_in_msg_get_replaced(self):
assert len('123456') <= self.handler.width
self.handler.emit(logging.makeLogRecord(
{'msg': '123%s', 'args': (456,)}))
self.d.infobox.assert_called_once_with('123456', mock.ANY, mock.ANY)
def test_wraps_nospace_is_greedy(self):
assert len('1234567') > self.handler.width
self.handler.emit(logging.makeLogRecord({'msg': '1234567'}))
self.d.infobox.assert_called_once_with('123456\n7', mock.ANY, mock.ANY)
def test_wraps_at_whitespace(self):
assert len('123 567') > self.handler.width
self.handler.emit(logging.makeLogRecord({'msg': '123 567'}))
self.d.infobox.assert_called_once_with('123\n567', mock.ANY, mock.ANY)
def test_only_last_lines_are_printed(self):
assert len('a\nb\nc'.split()) > self.handler.height
self.handler.emit(logging.makeLogRecord({'msg': 'a\n\nb\nc'}))
self.d.infobox.assert_called_once_with('b\nc', mock.ANY, mock.ANY)
def handle(self):
"""Handle multiple requests - each expected to be of 4-byte length,
followed by the LogRecord in pickle format. Logs the record
according to whatever policy is configured locally."""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack(">L", chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unpickle(chunk)
record = logging.makeLogRecord(obj)
self.handle_log_record(record)
def test_race(self):
# Issue #14632 refers.
def remove_loop(fname, tries):
for _ in range(tries):
try:
os.unlink(fname)
except OSError:
pass
time.sleep(0.004 * random.randint(0, 4))
del_count = 500
log_count = 500
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' })
h.handle(r)
finally:
remover.join()
try:
h.close()
except ValueError:
pass
if os.path.exists(fn):
os.unlink(fn)
# Set the locale to the platform-dependent default. I have no idea
# why the test does this, but in any case we save the current locale
# first and restore it at the end.
def handle(self):
"""Handle multiple requests - each expected to be of 4-byte length,
followed by the LogRecord in pickle format. Logs the record
according to whatever policy is configured locally."""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack(">L", chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unpickle(chunk)
record = logging.makeLogRecord(obj)
self.handle_log_record(record)
def test_race(self):
# Issue #14632 refers.
def remove_loop(fname, tries):
for _ in range(tries):
try:
os.unlink(fname)
except OSError:
pass
time.sleep(0.004 * random.randint(0, 4))
del_count = 500
log_count = 500
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' })
h.handle(r)
finally:
remover.join()
try:
h.close()
except ValueError:
pass
if os.path.exists(fn):
os.unlink(fn)
# Set the locale to the platform-dependent default. I have no idea
# why the test does this, but in any case we save the current locale
# first and restore it at the end.
def log(self, msg: Message):
d = msg._asdict()
d['asctime'] = datetime.fromtimestamp(msg.time, self.tz).strftime('%Y-%m-%d %H:%M:%S')
d['srcname'] = msg.src.alias
d['srcid'] = msg.src.id
self.loghandler.emit(logging.makeLogRecord({'msg': self.FORMAT % d}))
def post(self):
self.log.handle(logging.makeLogRecord(self.get_json_body()))
def test_empty_filter(self):
f = logging.Filter()
r = logging.makeLogRecord({'name': 'spam.eggs'})
self.assertTrue(f.filter(r))
#
# First, we define our levels. There can be as many as you want - the only
# limitations are that they should be integers, the lowest should be > 0 and
# larger values mean less information being logged. If you need specific
# level values which do not fit into these limitations, you can use a
# mapping dictionary to convert between your application levels and the
# logging system.
#
def test_race(self):
# Issue #14632 refers.
def remove_loop(fname, tries):
for _ in range(tries):
try:
os.unlink(fname)
except OSError:
pass
time.sleep(0.004 * random.randint(0, 4))
del_count = 500
log_count = 500
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' })
h.handle(r)
finally:
remover.join()
h.close()
if os.path.exists(fn):
os.unlink(fn)
def handle_datagram(self, request):
slen = struct.pack('>L', 0) # length of prefix
packet = request.packet[len(slen):]
obj = pickle.loads(packet)
record = logging.makeLogRecord(obj)
self.log_output += record.msg + '\n'
self.handled.set()
def get_record(self, name=None):
result = dict(self.common)
if name is not None:
result.update(self.variants[name])
return logging.makeLogRecord(result)
def setUp(self):
self.records = [
logging.makeLogRecord({'msg': 'one'}),
logging.makeLogRecord({'msg': 'two'}),
]
def test_str_rep(self):
r = logging.makeLogRecord({})
s = str(r)
self.assertTrue(s.startswith('<LogRecord: '))
self.assertTrue(s.endswith('>'))
def test_multiprocessing(self):
r = logging.makeLogRecord({})
self.assertEqual(r.processName, 'MainProcess')
try:
import multiprocessing as mp
r = logging.makeLogRecord({})
self.assertEqual(r.processName, mp.current_process().name)
except ImportError:
pass
def test_delay(self):
os.unlink(self.fn)
fh = logging.FileHandler(self.fn, delay=True)
self.assertIsNone(fh.stream)
self.assertFalse(os.path.exists(self.fn))
fh.handle(logging.makeLogRecord({}))
self.assertIsNotNone(fh.stream)
self.assertTrue(os.path.exists(self.fn))
fh.close()
def test_rollover(self):
fh = logging.handlers.TimedRotatingFileHandler(self.fn, 'S',
backupCount=1)
fmt = logging.Formatter('%(asctime)s %(message)s')
fh.setFormatter(fmt)
r1 = logging.makeLogRecord({'msg': 'testing - initial'})
fh.emit(r1)
self.assertLogFile(self.fn)
time.sleep(1.1) # a little over a second ...
r2 = logging.makeLogRecord({'msg': 'testing - after delay'})
fh.emit(r2)
fh.close()
# At this point, we should have a recent rotated file which we
# can test for the existence of. However, in practice, on some
# machines which run really slowly, we don't know how far back
# in time to go to look for the log file. So, we go back a fair
# bit, and stop as soon as we see a rotated file. In theory this
# could of course still fail, but the chances are lower.
found = False
now = datetime.datetime.now()
GO_BACK = 5 * 60 # seconds
for secs in range(GO_BACK):
prev = now - datetime.timedelta(seconds=secs)
fn = self.fn + prev.strftime(".%Y-%m-%d_%H-%M-%S")
found = os.path.exists(fn)
if found:
self.rmfiles.append(fn)
break
msg = 'No rotated files found, went back %d seconds' % GO_BACK
if not found:
#print additional diagnostics
dn, fn = os.path.split(self.fn)
files = [f for f in os.listdir(dn) if f.startswith(fn)]
print('Test time: %s' % now.strftime("%Y-%m-%d %H-%M-%S"), file=sys.stderr)
print('The only matching files are: %s' % files, file=sys.stderr)
for f in files:
print('Contents of %s:' % f)
path = os.path.join(dn, f)
with open(path, 'r') as tf:
print(tf.read())
self.assertTrue(found, msg=msg)
def test_basic(self):
logtype = 'Application'
elh = win32evtlog.OpenEventLog(None, logtype)
num_recs = win32evtlog.GetNumberOfEventLogRecords(elh)
h = logging.handlers.NTEventLogHandler('test_logging')
r = logging.makeLogRecord({'msg': 'Test Log Message'})
h.handle(r)
h.close()
# Now see if the event is recorded
self.assertTrue(num_recs < win32evtlog.GetNumberOfEventLogRecords(elh))
flags = win32evtlog.EVENTLOG_BACKWARDS_READ | \
win32evtlog.EVENTLOG_SEQUENTIAL_READ
found = False
GO_BACK = 100
events = win32evtlog.ReadEventLog(elh, flags, GO_BACK)
for e in events:
if e.SourceName != 'test_logging':
continue
msg = win32evtlogutil.SafeFormatMessage(e, logtype)
if msg != 'Test Log Message\r\n':
continue
found = True
break
msg = 'Record not found in event log, went back %d records' % GO_BACK
self.assertTrue(found, msg=msg)
# Set the locale to the platform-dependent default. I have no idea
# why the test does this, but in any case we save the current locale
# first and restore it at the end.
def handle(self):
"""Handle multiple requests - each expected to be of 4-byte length,
followed by the LogRecord in pickle format. Logs the record
according to whatever policy is configured locally."""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack(">L", chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unpickle(chunk)
record = logging.makeLogRecord(obj)
self.handle_log_record(record)
def test_race(self):
# Issue #14632 refers.
def remove_loop(fname, tries):
for _ in range(tries):
try:
os.unlink(fname)
except OSError:
pass
time.sleep(0.004 * random.randint(0, 4))
del_count = 500
log_count = 500
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' })
h.handle(r)
finally:
remover.join()
try:
h.close()
except ValueError:
pass
if os.path.exists(fn):
os.unlink(fn)
# Set the locale to the platform-dependent default. I have no idea
# why the test does this, but in any case we save the current locale
# first and restore it at the end.