# Copyright 2001-2004 by Vinay Sajip. All Rights Reserved.
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose and without fee is hereby granted,
# provided that the above copyright notice appear in all copies and that
# both that copyright notice and this permission notice appear in
# supporting documentation, and that the name of Vinay Sajip
# not be used in advertising or publicity pertaining to distribution
# of the software without specific, written prior permission.
# VINAY SAJIP DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING
# ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
# VINAY SAJIP BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR
# ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER
# IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
# This file is part of the Python logging distribution. See
# http://www.red-dove.com/python_logging.html
"""Test harness for the logging module. Run all tests.
Copyright (C) 2001-2002 Vinay Sajip. All Rights Reserved.
import os
, sys
, string
, struct
, types
, cPickle
, cStringIO
import socket
, threading
, time
import logging
, logging
.handlers
, logging
.config
BANNER
= "-- %-10s %-6s ---------------------------------------------------\n"
FINISH_UP
= "Finish up, it's closing time. Messages should bear numbers 0 through 24."
#----------------------------------------------------------------------------
#----------------------------------------------------------------------------
from SocketServer
import ThreadingTCPServer
, StreamRequestHandler
class LogRecordStreamHandler(StreamRequestHandler
):
Handler for a streaming logging request. It basically logs the record
using whatever logging policy is configured locally.
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.
chunk
= self
.connection
.recv(4)
slen
= struct
.unpack(">L", chunk
)[0]
chunk
= self
.connection
.recv(slen
)
chunk
= chunk
+ self
.connection
.recv(slen
- len(chunk
))
obj
= self
.unPickle(chunk
)
record
= logging
.makeLogRecord(obj
)
self
.handleLogRecord(record
)
def unPickle(self
, data
):
return cPickle
.loads(data
)
def handleLogRecord(self
, record
):
logname
= "logrecv.tcp." + record
.name
#If the end-of-messages sentinel is seen, tell the server to terminate
if record
.msg
== FINISH_UP
:
record
.msg
= record
.msg
+ " (via " + logname
+ ")"
logger
= logging
.getLogger(logname
)
# The server sets socketDataProcessed when it's done.
socketDataProcessed
= threading
.Event()
class LogRecordSocketReceiver(ThreadingTCPServer
):
A simple-minded TCP socket-based logging receiver suitable for test
def __init__(self
, host
='localhost',
port
=logging
.handlers
.DEFAULT_TCP_LOGGING_PORT
,
handler
=LogRecordStreamHandler
):
ThreadingTCPServer
.__init
__(self
, (host
, port
), handler
)
def serve_until_stopped(self
):
rd
, wr
, ex
= select
.select([self
.socket
.fileno()],
#notify the main thread that we're about to exit
socketDataProcessed
.set()
def process_request(self
, request
, client_address
):
t
= threading
.Thread(target
= self
.finish_request
,
args
= (request
, client_address
))
tcpserver
.serve_until_stopped()
#----------------------------------------------------------------------------
#----------------------------------------------------------------------------
rv
= "Message %d" % msgcount
ERR
= logging
.getLogger("ERR")
ERR
.setLevel(logging
.ERROR
)
INF
= logging
.getLogger("INF")
INF
.setLevel(logging
.INFO
)
INF_ERR
= logging
.getLogger("INF.ERR")
INF_ERR
.setLevel(logging
.ERROR
)
DEB
= logging
.getLogger("DEB")
DEB
.setLevel(logging
.DEBUG
)
INF_UNDEF
= logging
.getLogger("INF.UNDEF")
INF_ERR_UNDEF
= logging
.getLogger("INF.ERR.UNDEF")
UNDEF
= logging
.getLogger("UNDEF")
GRANDCHILD
= logging
.getLogger("INF.BADPARENT.UNDEF")
CHILD
= logging
.getLogger("INF.BADPARENT")
ERR
.log(logging
.FATAL
, nextmessage())
INF
.log(logging
.FATAL
, nextmessage())
INF_UNDEF
.log(logging
.FATAL
, nextmessage())
INF_UNDEF
.error(nextmessage())
INF_UNDEF
.warn (nextmessage())
INF_UNDEF
.info (nextmessage())
INF_ERR
.log(logging
.FATAL
, nextmessage())
INF_ERR
.error(nextmessage())
INF_ERR_UNDEF
.log(logging
.FATAL
, nextmessage())
INF_ERR_UNDEF
.error(nextmessage())
DEB
.log(logging
.FATAL
, nextmessage())
UNDEF
.log(logging
.FATAL
, nextmessage())
UNDEF
.error(nextmessage())
UNDEF
.warn (nextmessage())
UNDEF
.info (nextmessage())
GRANDCHILD
.log(logging
.FATAL
, nextmessage())
CHILD
.log(logging
.FATAL
, nextmessage())
INF_UNDEF
.debug(nextmessage())
INF_ERR
.warn(nextmessage())
INF_ERR
.info(nextmessage())
INF_ERR
.debug(nextmessage())
INF_ERR_UNDEF
.warn(nextmessage())
INF_ERR_UNDEF
.info(nextmessage())
INF_ERR_UNDEF
.debug(nextmessage())
#----------------------------------------------------------------------------
#----------------------------------------------------------------------------
# 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
LEVEL_RANGE
= range(BORING
, SILENT
+ 1)
# Next, we define names for our levels. You don't need to do this - in which
# case the system will use "Level n" to denote the text for the level.
CHATTERBOX
: 'Chatterbox',
# Now, to demonstrate filtering: suppose for some perverse reason we only
# want to print out all except GARRULOUS messages. Let's create a filter for
class SpecificLevelFilter(logging
.Filter
):
def filter(self
, record
):
return self
.level
!= record
.levelno
class GarrulousFilter(SpecificLevelFilter
):
SpecificLevelFilter
.__init
__(self
, GARRULOUS
)
# Now, let's demonstrate filtering at the logger. This time, use a filter
# which excludes SOCIABLE and TACITURN messages. Note that GARRULOUS events
class VerySpecificFilter(logging
.Filter
):
def filter(self
, record
):
return record
.levelno
not in [SOCIABLE
, TACITURN
]
sys
.stdout
.write("%s\n" % s
)
SHOULD1
= "This should only be seen at the '%s' logging level (or lower)"
# Now, tell the logging system to associate names with our levels.
for lvl
in my_logging_levels
.keys():
logging
.addLevelName(lvl
, my_logging_levels
[lvl
])
# Now, define a test function which logs an event at each of our levels.
log
.log(lvl
, SHOULD1
, logging
.getLevelName(lvl
))
log
= logging
.getLogger("")
# Set the logging level to each different value and call the utility
# function to log events.
# In the output, you should see that each time round the loop, the number of
# logging events which are actually output decreases.
message("-- setting logging level to '%s' -----" %
logging
.getLevelName(lvl
))
# Now, we demonstrate level filtering at the handler level. Tell the
# handler defined above to filter at level 'SOCIABLE', and repeat the
# above loop. Compare the output from the two runs.
message("-- Filtering at handler level to SOCIABLE --")
message("-- setting logging level to '%s' -----" %
logging
.getLevelName(lvl
))
hdlr
.setLevel(0) #turn off level filtering at the handler
message("-- Filtering using GARRULOUS filter --")
message("-- setting logging level to '%s' -----" %
logging
.getLevelName(lvl
))
spec
= VerySpecificFilter()
message("-- Filtering using specific filter for SOCIABLE, TACITURN --")
message("-- setting logging level to '%s' -----" %
logging
.getLevelName(lvl
))
#Undo the one level which clashes...for regression tests
logging
.addLevelName(logging
.DEBUG
, "DEBUG")
#----------------------------------------------------------------------------
#----------------------------------------------------------------------------
MSG
= "-- logging %d at INFO, messages should be seen every 10 events --"
logger
= logging
.getLogger("")
mh
= logging
.handlers
.MemoryHandler(10,logging
.WARNING
, sh
)
logger
.setLevel(logging
.DEBUG
)
message("-- logging at DEBUG, nothing should be seen yet --")
logger
.debug("Debug message")
message("-- logging at INFO, nothing should be seen yet --")
logger
.info("Info message")
message("-- logging at WARNING, 3 messages should be seen --")
logger
.warn("Warn message")
logger
.info("Info index = %d", i
)
#----------------------------------------------------------------------------
#----------------------------------------------------------------------------
logging
.getLogger("a").info("Info 1")
logging
.getLogger("a.b").info("Info 2")
logging
.getLogger("a.c").info("Info 3")
logging
.getLogger("a.b.c").info("Info 4")
logging
.getLogger("a.b.c.d").info("Info 5")
logging
.getLogger("a.bb.c").info("Info 6")
logging
.getLogger("b").info("Info 7")
logging
.getLogger("b.a").info("Info 8")
logging
.getLogger("c.a.b").info("Info 9")
logging
.getLogger("a.bb").info("Info 10")
root
= logging
.getLogger()
root
.setLevel(logging
.DEBUG
)
message("Filtered with '%s'..." % FILTER
)
filt
= logging
.Filter(FILTER
)
#----------------------------------------------------------------------------
#----------------------------------------------------------------------------
rootLogger
= logging
.getLogger("")
rootLogger
.setLevel(logging
.DEBUG
)
hdlr
= logging
.StreamHandler(sys
.stdout
)
fmt
= logging
.Formatter(logging
.BASIC_FORMAT
)
rootLogger
.addHandler(hdlr
)
#Set up a handler such that all events are sent via a socket to the log
#The handler will only be added to the rootLogger for some of the tests
shdlr
= logging
.handlers
.SocketHandler('localhost',
logging
.handlers
.DEFAULT_TCP_LOGGING_PORT
)
#Configure the logger for logrecv so events do not propagate beyond it.
#The sockLogger output is buffered in memory until the end of the test,
sockOut
= cStringIO
.StringIO()
sockLogger
= logging
.getLogger("logrecv")
sockLogger
.setLevel(logging
.DEBUG
)
sockhdlr
= logging
.StreamHandler(sockOut
)
sockhdlr
.setFormatter(logging
.Formatter(
"%(name)s -> %(levelname)s: %(message)s"))
sockLogger
.addHandler(sockhdlr
)
tcpserver
= LogRecordSocketReceiver()
#sys.stdout.write("About to start TCP server...\n")
threads
.append(threading
.Thread(target
=runTCP
, args
=(tcpserver
,)))
banner("log_test0", "begin")
rootLogger
.addHandler(shdlr
)
rootLogger
.removeHandler(shdlr
)
banner("log_test0", "end")
banner("log_test1", "begin")
banner("log_test1", "end")
banner("log_test2", "begin")
banner("log_test2", "end")
banner("log_test3", "begin")
banner("log_test3", "end")
#wait for TCP receiver to terminate
socketDataProcessed
.wait()
banner("logrecv output", "begin")
sys
.stdout
.write(sockOut
.getvalue())
sockLogger
.removeHandler(sockhdlr
)
banner("logrecv output", "end")
rootLogger
.removeHandler(hdlr
)
# 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 so we can restore it at the end.
original_locale
= locale
.setlocale(locale
.LC_ALL
)
locale
.setlocale(locale
.LC_ALL
, '')
except (ValueError, locale
.Error
):
# this happens on a Solaris box which only supports "C" locale
# or a Mac OS X box which supports very little locale stuff at all
if original_locale
is not None:
locale
.setlocale(locale
.LC_ALL
, original_locale
)
if __name__
== "__main__":
sys
.stdout
.write("test_logging\n")