Menu
▾
▴
[Jython-checkins] SF.net SVN: jython: [4755] branches/asm/Lib/test/test_logging.py
|
From: <pj...@us...> - 2008-06-24 23:12:20
|
Revision: 4755
http://jython.svn.sourceforge.net/jython/?rev=4755&view=rev
Author: pjenvey
Date: 2008-06-24 16:12:17 -0700 (Tue, 24 Jun 2008)
Log Message:
-----------
from:
http://svn.python.org/projects/python/branches/release25-maint/Lib/test/test_logging.py@54275
Modified Paths:
--------------
branches/asm/Lib/test/test_logging.py
Modified: branches/asm/Lib/test/test_logging.py
===================================================================
--- branches/asm/Lib/test/test_logging.py 2008-06-24 21:31:42 UTC (rev 4754)
+++ branches/asm/Lib/test/test_logging.py 2008-06-24 23:12:17 UTC (rev 4755)
@@ -1,6 +1,6 @@
#!/usr/bin/env python
#
-# Copyright 2001-2002 by Vinay Sajip. All Rights Reserved.
+# 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,
@@ -24,10 +24,11 @@
Copyright (C) 2001-2002 Vinay Sajip. All Rights Reserved.
"""
+import select
import os, sys, string, struct, types, cPickle, cStringIO
-import socket, threading, time
+import socket, tempfile, threading, time
import logging, logging.handlers, logging.config
-from select import select
+from test.test_support import run_with_locale
BANNER = "-- %-10s %-6s ---------------------------------------------------\n"
@@ -99,22 +100,15 @@
self.timeout = 1
def serve_until_stopped(self):
- if sys.platform.startswith('java'):
- # XXX: There's a problem using cpython_compatibile_select
- # here: it seems to be due to the fact that
- # cpython_compatible_select switches blocking mode on while
- # a separate thread is reading from the same socket, causing
- # a read of 0 in LogRecordStreamHandler.handle (which
- # deadlocks this test)
- self.socket.setblocking(0)
- abort = 0
- while not abort:
- rd, wr, ex = select([self.socket.fileno()], [], [], self.timeout)
+ while not self.abort:
+ rd, wr, ex = select.select([self.socket.fileno()], [], [],
+ self.timeout)
if rd:
self.handle_request()
- abort = self.abort
#notify the main thread that we're about to exit
socketDataProcessed.set()
+ # close the listen socket
+ self.server_close()
def process_request(self, request, client_address):
#import threading
@@ -399,6 +393,184 @@
hand.removeFilter(filt)
#----------------------------------------------------------------------------
+# Test 4
+#----------------------------------------------------------------------------
+
+# config0 is a standard configuration.
+config0 = """
+[loggers]
+keys=root
+
+[handlers]
+keys=hand1
+
+[formatters]
+keys=form1
+
+[logger_root]
+level=NOTSET
+handlers=hand1
+
+[handler_hand1]
+class=StreamHandler
+level=NOTSET
+formatter=form1
+args=(sys.stdout,)
+
+[formatter_form1]
+format=%(levelname)s:%(name)s:%(message)s
+datefmt=
+"""
+
+# config1 adds a little to the standard configuration.
+config1 = """
+[loggers]
+keys=root,parser
+
+[handlers]
+keys=hand1
+
+[formatters]
+keys=form1
+
+[logger_root]
+level=NOTSET
+handlers=hand1
+
+[logger_parser]
+level=DEBUG
+handlers=hand1
+propagate=1
+qualname=compiler.parser
+
+[handler_hand1]
+class=StreamHandler
+level=NOTSET
+formatter=form1
+args=(sys.stdout,)
+
+[formatter_form1]
+format=%(levelname)s:%(name)s:%(message)s
+datefmt=
+"""
+
+# config2 has a subtle configuration error that should be reported
+config2 = string.replace(config1, "sys.stdout", "sys.stbout")
+
+# config3 has a less subtle configuration error
+config3 = string.replace(
+ config1, "formatter=form1", "formatter=misspelled_name")
+
+def test4():
+ for i in range(4):
+ conf = globals()['config%d' % i]
+ sys.stdout.write('config%d: ' % i)
+ loggerDict = logging.getLogger().manager.loggerDict
+ logging._acquireLock()
+ try:
+ saved_handlers = logging._handlers.copy()
+ saved_handler_list = logging._handlerList[:]
+ saved_loggers = loggerDict.copy()
+ finally:
+ logging._releaseLock()
+ try:
+ fn = tempfile.mktemp(".ini")
+ f = open(fn, "w")
+ f.write(conf)
+ f.close()
+ try:
+ logging.config.fileConfig(fn)
+ #call again to make sure cleanup is correct
+ logging.config.fileConfig(fn)
+ except:
+ t = sys.exc_info()[0]
+ message(str(t))
+ else:
+ message('ok.')
+ os.remove(fn)
+ finally:
+ logging._acquireLock()
+ try:
+ logging._handlers.clear()
+ logging._handlers.update(saved_handlers)
+ logging._handlerList[:] = saved_handler_list
+ loggerDict = logging.getLogger().manager.loggerDict
+ loggerDict.clear()
+ loggerDict.update(saved_loggers)
+ finally:
+ logging._releaseLock()
+
+#----------------------------------------------------------------------------
+# Test 5
+#----------------------------------------------------------------------------
+
+test5_config = """
+[loggers]
+keys=root
+
+[handlers]
+keys=hand1
+
+[formatters]
+keys=form1
+
+[logger_root]
+level=NOTSET
+handlers=hand1
+
+[handler_hand1]
+class=StreamHandler
+level=NOTSET
+formatter=form1
+args=(sys.stdout,)
+
+[formatter_form1]
+class=test.test_logging.FriendlyFormatter
+format=%(levelname)s:%(name)s:%(message)s
+datefmt=
+"""
+
+class FriendlyFormatter (logging.Formatter):
+ def formatException(self, ei):
+ return "%s... Don't panic!" % str(ei[0])
+
+
+def test5():
+ loggerDict = logging.getLogger().manager.loggerDict
+ logging._acquireLock()
+ try:
+ saved_handlers = logging._handlers.copy()
+ saved_handler_list = logging._handlerList[:]
+ saved_loggers = loggerDict.copy()
+ finally:
+ logging._releaseLock()
+ try:
+ fn = tempfile.mktemp(".ini")
+ f = open(fn, "w")
+ f.write(test5_config)
+ f.close()
+ logging.config.fileConfig(fn)
+ try:
+ raise KeyError
+ except KeyError:
+ logging.exception("just testing")
+ os.remove(fn)
+ hdlr = logging.getLogger().handlers[0]
+ logging.getLogger().handlers.remove(hdlr)
+ finally:
+ logging._acquireLock()
+ try:
+ logging._handlers.clear()
+ logging._handlers.update(saved_handlers)
+ logging._handlerList[:] = saved_handler_list
+ loggerDict = logging.getLogger().manager.loggerDict
+ loggerDict.clear()
+ loggerDict.update(saved_loggers)
+ finally:
+ logging._releaseLock()
+
+
+#----------------------------------------------------------------------------
# Test Harness
#----------------------------------------------------------------------------
def banner(nm, typ):
@@ -414,11 +586,23 @@
hdlr.setFormatter(fmt)
rootLogger.addHandler(hdlr)
+ # Find an unused port number
+ port = logging.handlers.DEFAULT_TCP_LOGGING_PORT
+ while port < logging.handlers.DEFAULT_TCP_LOGGING_PORT+100:
+ try:
+ tcpserver = LogRecordSocketReceiver(port=port)
+ except socket.error:
+ port += 1
+ else:
+ break
+ else:
+ raise ImportError, "Could not find unused port"
+
+
#Set up a handler such that all events are sent via a socket to the log
#receiver (logrecv).
#The handler will only be added to the rootLogger for some of the tests
- hdlr = logging.handlers.SocketHandler('localhost',
- logging.handlers.DEFAULT_TCP_LOGGING_PORT)
+ shdlr = logging.handlers.SocketHandler('localhost', 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,
@@ -434,7 +618,6 @@
#Set up servers
threads = []
- tcpserver = LogRecordSocketReceiver()
#sys.stdout.write("About to start TCP server...\n")
threads.append(threading.Thread(target=runTCP, args=(tcpserver,)))
@@ -443,54 +626,59 @@
try:
banner("log_test0", "begin")
- rootLogger.addHandler(hdlr)
+ rootLogger.addHandler(shdlr)
test0()
- hdlr.close()
- rootLogger.removeHandler(hdlr)
+ # XXX(nnorwitz): Try to fix timing related test failures.
+ # This sleep gives us some extra time to read messages.
+ # The test generally only fails on Solaris without this sleep.
+ time.sleep(2.0)
+ shdlr.close()
+ rootLogger.removeHandler(shdlr)
banner("log_test0", "end")
- banner("log_test1", "begin")
- test1()
- banner("log_test1", "end")
+ for t in range(1,6):
+ banner("log_test%d" % t, "begin")
+ globals()['test%d' % t]()
+ banner("log_test%d" % t, "end")
- banner("log_test2", "begin")
- test2()
- banner("log_test2", "end")
-
- banner("log_test3", "begin")
- test3()
- banner("log_test3", "end")
-
finally:
#wait for TCP receiver to terminate
socketDataProcessed.wait()
+ # ensure the server dies
+ tcpserver.abort = 1
for thread in threads:
- thread.join()
+ thread.join(2.0)
banner("logrecv output", "begin")
sys.stdout.write(sockOut.getvalue())
sockOut.close()
+ sockLogger.removeHandler(sockhdlr)
+ sockhdlr.close()
banner("logrecv output", "end")
sys.stdout.flush()
+ try:
+ hdlr.close()
+ except:
+ pass
+ 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 and restore it at the end.
+@run_with_locale('LC_ALL', '')
def test_main():
- import locale
- # 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.
+ # Save and restore the original root logger level across the tests.
+ # Otherwise, e.g., if any test using cookielib runs after test_logging,
+ # cookielib's debug-level logger tries to log messages, leading to
+ # confusing:
+ # No handlers could be found for logger "cookielib"
+ # output while the tests are running.
+ root_logger = logging.getLogger("")
+ original_logging_level = root_logger.getEffectiveLevel()
try:
- 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
- original_locale = None
-
- try:
test_main_inner()
finally:
- if original_locale is not None:
- locale.setlocale(locale.LC_ALL, original_locale)
+ root_logger.setLevel(original_logging_level)
if __name__ == "__main__":
sys.stdout.write("test_logging\n")
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|