3 # Copyright 2001-2010 by Vinay Sajip. All Rights Reserved.
5 # Permission to use, copy, modify, and distribute this software and its
6 # documentation for any purpose and without fee is hereby granted,
7 # provided that the above copyright notice appear in all copies and that
8 # both that copyright notice and this permission notice appear in
9 # supporting documentation, and that the name of Vinay Sajip
10 # not be used in advertising or publicity pertaining to distribution
11 # of the software without specific, written prior permission.
12 # VINAY SAJIP DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING
13 # ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
14 # VINAY SAJIP BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR
15 # ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER
16 # IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
17 # OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
19 """Test harness for the logging module. Run all tests.
21 Copyright (C) 2001-2010 Vinay Sajip. All Rights Reserved.
25 import logging
.handlers
37 from SocketServer
import ThreadingTCPServer
, StreamRequestHandler
41 from test
.test_support
import captured_stdout
, run_with_locale
, run_unittest
51 class BaseTest(unittest
.TestCase
):
53 """Base class for logging tests."""
55 log_format
= "%(name)s -> %(levelname)s: %(message)s"
56 expected_log_pat
= r
"^([\w.]+) -> ([\w]+): ([\d]+)$"
60 """Setup the default logging stream to an internal StringIO instance,
61 so that we can examine log output as we want."""
62 logger_dict
= logging
.getLogger().manager
.loggerDict
63 logging
._acquireLock
()
65 self
.saved_handlers
= logging
._handlers
.copy()
66 self
.saved_handler_list
= logging
._handlerList
[:]
67 self
.saved_loggers
= logger_dict
.copy()
68 self
.saved_level_names
= logging
._levelNames
.copy()
70 logging
._releaseLock
()
72 # Set two unused loggers: one non-ASCII and one Unicode.
73 # This is to test correct operation when sorting existing
74 # loggers in the configuration code. See issue 8201.
75 logging
.getLogger("\xab\xd7\xbb")
76 logging
.getLogger(u
"\u013f\u00d6\u0047")
78 self
.root_logger
= logging
.getLogger("")
79 self
.original_logging_level
= self
.root_logger
.getEffectiveLevel()
81 self
.stream
= cStringIO
.StringIO()
82 self
.root_logger
.setLevel(logging
.DEBUG
)
83 self
.root_hdlr
= logging
.StreamHandler(self
.stream
)
84 self
.root_formatter
= logging
.Formatter(self
.log_format
)
85 self
.root_hdlr
.setFormatter(self
.root_formatter
)
86 self
.root_logger
.addHandler(self
.root_hdlr
)
89 """Remove our logging stream, and restore the original logging
92 self
.root_logger
.removeHandler(self
.root_hdlr
)
93 while self
.root_logger
.handlers
:
94 h
= self
.root_logger
.handlers
[0]
95 self
.root_logger
.removeHandler(h
)
97 self
.root_logger
.setLevel(self
.original_logging_level
)
98 logging
._acquireLock
()
100 logging
._levelNames
.clear()
101 logging
._levelNames
.update(self
.saved_level_names
)
102 logging
._handlers
.clear()
103 logging
._handlers
.update(self
.saved_handlers
)
104 logging
._handlerList
[:] = self
.saved_handler_list
105 loggerDict
= logging
.getLogger().manager
.loggerDict
107 loggerDict
.update(self
.saved_loggers
)
109 logging
._releaseLock
()
111 def assert_log_lines(self
, expected_values
, stream
=None):
112 """Match the collected log lines against the regular expression
113 self.expected_log_pat, and compare the extracted group values to
114 the expected_values list of tuples."""
115 stream
= stream
or self
.stream
116 pat
= re
.compile(self
.expected_log_pat
)
119 actual_lines
= stream
.readlines()
120 except AttributeError:
121 # StringIO.StringIO lacks a reset() method.
122 actual_lines
= stream
.getvalue().splitlines()
123 self
.assertEquals(len(actual_lines
), len(expected_values
))
124 for actual
, expected
in zip(actual_lines
, expected_values
):
125 match
= pat
.search(actual
)
127 self
.fail("Log line does not match expected pattern:\n" +
129 self
.assertEquals(tuple(match
.groups()), expected
)
132 self
.fail("Remaining output at end of log stream:\n" + s
)
134 def next_message(self
):
135 """Generate a message consisting solely of an auto-incrementing
137 self
.message_num
+= 1
138 return "%d" % self
.message_num
141 class BuiltinLevelsTest(BaseTest
):
142 """Test builtin levels and their inheritance."""
145 #Logging levels in a flat logger namespace.
146 m
= self
.next_message
148 ERR
= logging
.getLogger("ERR")
149 ERR
.setLevel(logging
.ERROR
)
150 INF
= logging
.getLogger("INF")
151 INF
.setLevel(logging
.INFO
)
152 DEB
= logging
.getLogger("DEB")
153 DEB
.setLevel(logging
.DEBUG
)
156 ERR
.log(logging
.CRITICAL
, m())
159 INF
.log(logging
.CRITICAL
, m())
164 DEB
.log(logging
.CRITICAL
, m())
170 # These should not log.
177 self
.assert_log_lines([
178 ('ERR', 'CRITICAL', '1'),
179 ('ERR', 'ERROR', '2'),
180 ('INF', 'CRITICAL', '3'),
181 ('INF', 'ERROR', '4'),
182 ('INF', 'WARNING', '5'),
183 ('INF', 'INFO', '6'),
184 ('DEB', 'CRITICAL', '7'),
185 ('DEB', 'ERROR', '8'),
186 ('DEB', 'WARNING', '9'),
187 ('DEB', 'INFO', '10'),
188 ('DEB', 'DEBUG', '11'),
191 def test_nested_explicit(self
):
192 # Logging levels in a nested namespace, all explicitly set.
193 m
= self
.next_message
195 INF
= logging
.getLogger("INF")
196 INF
.setLevel(logging
.INFO
)
197 INF_ERR
= logging
.getLogger("INF.ERR")
198 INF_ERR
.setLevel(logging
.ERROR
)
201 INF_ERR
.log(logging
.CRITICAL
, m())
204 # These should not log.
209 self
.assert_log_lines([
210 ('INF.ERR', 'CRITICAL', '1'),
211 ('INF.ERR', 'ERROR', '2'),
214 def test_nested_inherited(self
):
215 #Logging levels in a nested namespace, inherited from parent loggers.
216 m
= self
.next_message
218 INF
= logging
.getLogger("INF")
219 INF
.setLevel(logging
.INFO
)
220 INF_ERR
= logging
.getLogger("INF.ERR")
221 INF_ERR
.setLevel(logging
.ERROR
)
222 INF_UNDEF
= logging
.getLogger("INF.UNDEF")
223 INF_ERR_UNDEF
= logging
.getLogger("INF.ERR.UNDEF")
224 UNDEF
= logging
.getLogger("UNDEF")
227 INF_UNDEF
.log(logging
.CRITICAL
, m())
231 INF_ERR_UNDEF
.log(logging
.CRITICAL
, m())
232 INF_ERR_UNDEF
.error(m())
234 # These should not log.
236 INF_ERR_UNDEF
.warn(m())
237 INF_ERR_UNDEF
.info(m())
238 INF_ERR_UNDEF
.debug(m())
240 self
.assert_log_lines([
241 ('INF.UNDEF', 'CRITICAL', '1'),
242 ('INF.UNDEF', 'ERROR', '2'),
243 ('INF.UNDEF', 'WARNING', '3'),
244 ('INF.UNDEF', 'INFO', '4'),
245 ('INF.ERR.UNDEF', 'CRITICAL', '5'),
246 ('INF.ERR.UNDEF', 'ERROR', '6'),
249 def test_nested_with_virtual_parent(self
):
250 # Logging levels when some parent does not exist yet.
251 m
= self
.next_message
253 INF
= logging
.getLogger("INF")
254 GRANDCHILD
= logging
.getLogger("INF.BADPARENT.UNDEF")
255 CHILD
= logging
.getLogger("INF.BADPARENT")
256 INF
.setLevel(logging
.INFO
)
259 GRANDCHILD
.log(logging
.FATAL
, m())
261 CHILD
.log(logging
.FATAL
, m())
264 # These should not log.
265 GRANDCHILD
.debug(m())
268 self
.assert_log_lines([
269 ('INF.BADPARENT.UNDEF', 'CRITICAL', '1'),
270 ('INF.BADPARENT.UNDEF', 'INFO', '2'),
271 ('INF.BADPARENT', 'CRITICAL', '3'),
272 ('INF.BADPARENT', 'INFO', '4'),
276 class BasicFilterTest(BaseTest
):
278 """Test the bundled Filter class."""
280 def test_filter(self
):
281 # Only messages satisfying the specified criteria pass through the
283 filter_
= logging
.Filter("spam.eggs")
284 handler
= self
.root_logger
.handlers
[0]
286 handler
.addFilter(filter_
)
287 spam
= logging
.getLogger("spam")
288 spam_eggs
= logging
.getLogger("spam.eggs")
289 spam_eggs_fish
= logging
.getLogger("spam.eggs.fish")
290 spam_bakedbeans
= logging
.getLogger("spam.bakedbeans")
292 spam
.info(self
.next_message())
293 spam_eggs
.info(self
.next_message()) # Good.
294 spam_eggs_fish
.info(self
.next_message()) # Good.
295 spam_bakedbeans
.info(self
.next_message())
297 self
.assert_log_lines([
298 ('spam.eggs', 'INFO', '2'),
299 ('spam.eggs.fish', 'INFO', '3'),
302 handler
.removeFilter(filter_
)
306 # First, we define our levels. There can be as many as you want - the only
307 # limitations are that they should be integers, the lowest should be > 0 and
308 # larger values mean less information being logged. If you need specific
309 # level values which do not fit into these limitations, you can use a
310 # mapping dictionary to convert between your application levels and the
324 LEVEL_RANGE
= range(BORING
, SILENT
+ 1)
327 # Next, we define names for our levels. You don't need to do this - in which
328 # case the system will use "Level n" to denote the text for the level.
330 my_logging_levels
= {
332 TACITURN
: 'Taciturn',
334 EFFUSIVE
: 'Effusive',
335 SOCIABLE
: 'Sociable',
337 TALKATIVE
: 'Talkative',
338 GARRULOUS
: 'Garrulous',
339 CHATTERBOX
: 'Chatterbox',
343 class GarrulousFilter(logging
.Filter
):
345 """A filter which blocks garrulous messages."""
347 def filter(self
, record
):
348 return record
.levelno
!= GARRULOUS
350 class VerySpecificFilter(logging
.Filter
):
352 """A filter which blocks sociable and taciturn messages."""
354 def filter(self
, record
):
355 return record
.levelno
not in [SOCIABLE
, TACITURN
]
358 class CustomLevelsAndFiltersTest(BaseTest
):
360 """Test various filtering possibilities with custom logging levels."""
362 # Skip the logger name group.
363 expected_log_pat
= r
"^[\w.]+ -> ([\w]+): ([\d]+)$"
367 for k
, v
in my_logging_levels
.items():
368 logging
.addLevelName(k
, v
)
370 def log_at_all_levels(self
, logger
):
371 for lvl
in LEVEL_RANGE
:
372 logger
.log(lvl
, self
.next_message())
374 def test_logger_filter(self
):
375 # Filter at logger level.
376 self
.root_logger
.setLevel(VERBOSE
)
377 # Levels >= 'Verbose' are good.
378 self
.log_at_all_levels(self
.root_logger
)
379 self
.assert_log_lines([
388 def test_handler_filter(self
):
389 # Filter at handler level.
390 self
.root_logger
.handlers
[0].setLevel(SOCIABLE
)
392 # Levels >= 'Sociable' are good.
393 self
.log_at_all_levels(self
.root_logger
)
394 self
.assert_log_lines([
402 self
.root_logger
.handlers
[0].setLevel(logging
.NOTSET
)
404 def test_specific_filters(self
):
405 # Set a specific filter object on the handler, and then add another
406 # filter object on the logger itself.
407 handler
= self
.root_logger
.handlers
[0]
408 specific_filter
= None
409 garr
= GarrulousFilter()
410 handler
.addFilter(garr
)
412 self
.log_at_all_levels(self
.root_logger
)
414 # Notice how 'Garrulous' is missing
425 self
.assert_log_lines(first_lines
)
427 specific_filter
= VerySpecificFilter()
428 self
.root_logger
.addFilter(specific_filter
)
429 self
.log_at_all_levels(self
.root_logger
)
430 self
.assert_log_lines(first_lines
+ [
431 # Not only 'Garrulous' is still missing, but also 'Sociable'
434 ('Chatterbox', '12'),
443 self
.root_logger
.removeFilter(specific_filter
)
444 handler
.removeFilter(garr
)
447 class MemoryHandlerTest(BaseTest
):
449 """Tests for the MemoryHandler."""
451 # Do not bother with a logger name group.
452 expected_log_pat
= r
"^[\w.]+ -> ([\w]+): ([\d]+)$"
456 self
.mem_hdlr
= logging
.handlers
.MemoryHandler(10, logging
.WARNING
,
458 self
.mem_logger
= logging
.getLogger('mem')
459 self
.mem_logger
.propagate
= 0
460 self
.mem_logger
.addHandler(self
.mem_hdlr
)
463 self
.mem_hdlr
.close()
464 BaseTest
.tearDown(self
)
466 def test_flush(self
):
467 # The memory handler flushes to its target handler based on specific
468 # criteria (message count and message level).
469 self
.mem_logger
.debug(self
.next_message())
470 self
.assert_log_lines([])
471 self
.mem_logger
.info(self
.next_message())
472 self
.assert_log_lines([])
473 # This will flush because the level is >= logging.WARNING
474 self
.mem_logger
.warn(self
.next_message())
480 self
.assert_log_lines(lines
)
483 self
.mem_logger
.debug(self
.next_message())
484 self
.assert_log_lines(lines
)
485 # This will flush because it's the 10th message since the last
487 self
.mem_logger
.debug(self
.next_message())
488 lines
= lines
+ [('DEBUG', str(i
)) for i
in range(n
, n
+ 10)]
489 self
.assert_log_lines(lines
)
491 self
.mem_logger
.debug(self
.next_message())
492 self
.assert_log_lines(lines
)
495 class ExceptionFormatter(logging
.Formatter
):
496 """A special exception formatter."""
497 def formatException(self
, ei
):
498 return "Got a [%s]" % ei
[0].__name
__
501 class ConfigFileTest(BaseTest
):
503 """Reading logging config from a .ini-style config file."""
505 expected_log_pat
= r
"^([\w]+) \+\+ ([\w]+)$"
507 # config0 is a standard configuration.
529 format=%(levelname)s ++ %(message)s
533 # config1 adds a little to the standard configuration.
552 qualname=compiler.parser
561 format=%(levelname)s ++ %(message)s
565 # config2 has a subtle configuration error that should be reported
566 config2
= config1
.replace("sys.stdout", "sys.stbout")
568 # config3 has a less subtle configuration error
569 config3
= config1
.replace("formatter=form1", "formatter=misspelled_name")
571 # config4 specifies a custom formatter class to be loaded
593 class=""" + __name__
+ """.ExceptionFormatter
594 format=%(levelname)s:%(name)s:%(message)s
598 # config5 specifies a custom handler class to be loaded
599 config5
= config1
.replace('class=StreamHandler', 'class=logging.StreamHandler')
601 # config6 uses ', ' delimiters in the handlers and formatters sections
620 qualname=compiler.parser
635 format=%(levelname)s ++ %(message)s
643 def apply_config(self
, conf
):
644 file = cStringIO
.StringIO(textwrap
.dedent(conf
))
645 logging
.config
.fileConfig(file)
647 def test_config0_ok(self
):
648 # A simple config file which overrides the default settings.
649 with
captured_stdout() as output
:
650 self
.apply_config(self
.config0
)
651 logger
= logging
.getLogger()
652 # Won't output anything
653 logger
.info(self
.next_message())
655 logger
.error(self
.next_message())
656 self
.assert_log_lines([
659 # Original logger output is empty.
660 self
.assert_log_lines([])
662 def test_config1_ok(self
, config
=config1
):
663 # A config file defining a sub-parser as well.
664 with
captured_stdout() as output
:
665 self
.apply_config(config
)
666 logger
= logging
.getLogger("compiler.parser")
667 # Both will output a message
668 logger
.info(self
.next_message())
669 logger
.error(self
.next_message())
670 self
.assert_log_lines([
674 # Original logger output is empty.
675 self
.assert_log_lines([])
677 def test_config2_failure(self
):
678 # A simple config file which overrides the default settings.
679 self
.assertRaises(StandardError, self
.apply_config
, self
.config2
)
681 def test_config3_failure(self
):
682 # A simple config file which overrides the default settings.
683 self
.assertRaises(StandardError, self
.apply_config
, self
.config3
)
685 def test_config4_ok(self
):
686 # A config file specifying a custom formatter class.
687 with
captured_stdout() as output
:
688 self
.apply_config(self
.config4
)
689 logger
= logging
.getLogger()
693 logging
.exception("just testing")
695 self
.assertEquals(output
.getvalue(),
696 "ERROR:root:just testing\nGot a [RuntimeError]\n")
697 # Original logger output is empty
698 self
.assert_log_lines([])
700 def test_config5_ok(self
):
701 self
.test_config1_ok(config
=self
.config5
)
703 def test_config6_ok(self
):
704 self
.test_config1_ok(config
=self
.config6
)
706 class LogRecordStreamHandler(StreamRequestHandler
):
708 """Handler for a streaming logging request. It saves the log message in the
709 TCP server's 'log_output' attribute."""
711 TCP_LOG_END
= "!!!END!!!"
714 """Handle multiple requests - each expected to be of 4-byte length,
715 followed by the LogRecord in pickle format. Logs the record
716 according to whatever policy is configured locally."""
718 chunk
= self
.connection
.recv(4)
721 slen
= struct
.unpack(">L", chunk
)[0]
722 chunk
= self
.connection
.recv(slen
)
723 while len(chunk
) < slen
:
724 chunk
= chunk
+ self
.connection
.recv(slen
- len(chunk
))
725 obj
= self
.unpickle(chunk
)
726 record
= logging
.makeLogRecord(obj
)
727 self
.handle_log_record(record
)
729 def unpickle(self
, data
):
730 return cPickle
.loads(data
)
732 def handle_log_record(self
, record
):
733 # If the end-of-messages sentinel is seen, tell the server to
735 if self
.TCP_LOG_END
in record
.msg
:
736 self
.server
.abort
= 1
738 self
.server
.log_output
+= record
.msg
+ "\n"
741 class LogRecordSocketReceiver(ThreadingTCPServer
):
743 """A simple-minded TCP socket-based logging receiver suitable for test
746 allow_reuse_address
= 1
749 def __init__(self
, host
='localhost',
750 port
=logging
.handlers
.DEFAULT_TCP_LOGGING_PORT
,
751 handler
=LogRecordStreamHandler
):
752 ThreadingTCPServer
.__init
__(self
, (host
, port
), handler
)
755 self
.finished
= threading
.Event()
757 def serve_until_stopped(self
):
758 while not self
.abort
:
759 rd
, wr
, ex
= select
.select([self
.socket
.fileno()], [], [],
762 self
.handle_request()
763 # Notify the main thread that we're about to exit
765 # close the listen socket
769 @unittest.skipUnless(threading
, 'Threading required for this test.')
770 class SocketHandlerTest(BaseTest
):
772 """Test for SocketHandler objects."""
775 """Set up a TCP server to receive log messages, and a SocketHandler
776 pointing to that server's address and port."""
778 self
.tcpserver
= LogRecordSocketReceiver(port
=0)
779 self
.port
= self
.tcpserver
.socket
.getsockname()[1]
781 threading
.Thread(target
=self
.tcpserver
.serve_until_stopped
)]
782 for thread
in self
.threads
:
785 self
.sock_hdlr
= logging
.handlers
.SocketHandler('localhost', self
.port
)
786 self
.sock_hdlr
.setFormatter(self
.root_formatter
)
787 self
.root_logger
.removeHandler(self
.root_logger
.handlers
[0])
788 self
.root_logger
.addHandler(self
.sock_hdlr
)
791 """Shutdown the TCP server."""
793 self
.tcpserver
.abort
= True
795 self
.root_logger
.removeHandler(self
.sock_hdlr
)
796 self
.sock_hdlr
.close()
797 for thread
in self
.threads
:
800 BaseTest
.tearDown(self
)
802 def get_output(self
):
803 """Get the log output as received by the TCP server."""
804 # Signal the TCP receiver and wait for it to terminate.
805 self
.root_logger
.critical(LogRecordStreamHandler
.TCP_LOG_END
)
806 self
.tcpserver
.finished
.wait(2.0)
807 return self
.tcpserver
.log_output
809 def test_output(self
):
810 # The log message sent to the SocketHandler is properly received.
811 logger
= logging
.getLogger("tcp")
814 self
.assertEquals(self
.get_output(), "spam\neggs\n")
817 class MemoryTest(BaseTest
):
819 """Test memory persistence of logger objects."""
822 """Create a dict to remember potentially destroyed objects."""
826 def _watch_for_survival(self
, *args
):
827 """Watch the given objects for survival, by creating weakrefs to
830 key
= id(obj
), repr(obj
)
831 self
._survivors
[key
] = weakref
.ref(obj
)
833 def _assertTruesurvival(self
):
834 """Assert that all objects watched for survival have survived."""
835 # Trigger cycle breaking.
838 for (id_
, repr_
), ref
in self
._survivors
.items():
842 self
.fail("%d objects should have survived "
843 "but have been destroyed: %s" % (len(dead
), ", ".join(dead
)))
845 def test_persistent_loggers(self
):
846 # Logger objects are persistent and retain their configuration, even
847 # if visible references are destroyed.
848 self
.root_logger
.setLevel(logging
.INFO
)
849 foo
= logging
.getLogger("foo")
850 self
._watch
_for
_survival
(foo
)
851 foo
.setLevel(logging
.DEBUG
)
852 self
.root_logger
.debug(self
.next_message())
853 foo
.debug(self
.next_message())
854 self
.assert_log_lines([
855 ('foo', 'DEBUG', '2'),
859 self
._assertTruesurvival
()
860 # foo has retained its settings.
861 bar
= logging
.getLogger("foo")
862 bar
.debug(self
.next_message())
863 self
.assert_log_lines([
864 ('foo', 'DEBUG', '2'),
865 ('foo', 'DEBUG', '3'),
869 class EncodingTest(BaseTest
):
870 def test_encoding_plain_file(self
):
871 # In Python 2.x, a plain file object is treated as having no encoding.
872 log
= logging
.getLogger("test")
873 fn
= tempfile
.mktemp(".log")
874 # the non-ascii data we write to the log.
877 handler
= logging
.FileHandler(fn
)
878 log
.addHandler(handler
)
880 # write non-ascii data to the log.
883 log
.removeHandler(handler
)
885 # check we wrote exactly those bytes, ignoring trailing \n etc
888 self
.assertEqual(f
.read().rstrip(), data
)
892 if os
.path
.isfile(fn
):
895 def test_encoding_cyrillic_unicode(self
):
896 log
= logging
.getLogger("test")
897 #Get a message in Unicode: Do svidanya in Cyrillic (meaning goodbye)
898 message
= u
'\u0434\u043e \u0441\u0432\u0438\u0434\u0430\u043d\u0438\u044f'
899 #Ensure it's written in a Cyrillic encoding
900 writer_class
= codecs
.getwriter('cp1251')
901 writer_class
.encoding
= 'cp1251'
902 stream
= cStringIO
.StringIO()
903 writer
= writer_class(stream
, 'strict')
904 handler
= logging
.StreamHandler(writer
)
905 log
.addHandler(handler
)
909 log
.removeHandler(handler
)
911 # check we wrote exactly those bytes, ignoring trailing \n etc
912 s
= stream
.getvalue()
913 #Compare against what the data should be when encoded in CP-1251
914 self
.assertEqual(s
, '\xe4\xee \xf1\xe2\xe8\xe4\xe0\xed\xe8\xff\n')
917 class WarningsTest(BaseTest
):
919 def test_warnings(self
):
920 with warnings
.catch_warnings():
921 logging
.captureWarnings(True)
923 warnings
.filterwarnings("always", category
=UserWarning)
924 file = cStringIO
.StringIO()
925 h
= logging
.StreamHandler(file)
926 logger
= logging
.getLogger("py.warnings")
928 warnings
.warn("I'm warning you...")
929 logger
.removeHandler(h
)
932 self
.assertTrue(s
.find("UserWarning: I'm warning you...\n") > 0)
934 #See if an explicit file uses the original implementation
935 file = cStringIO
.StringIO()
936 warnings
.showwarning("Explicit", UserWarning, "dummy.py", 42,
941 "dummy.py:42: UserWarning: Explicit\n Dummy line\n")
943 logging
.captureWarnings(False)
946 def formatFunc(format
, datefmt
=None):
947 return logging
.Formatter(format
, datefmt
)
950 return logging
.StreamHandler()
952 class CustomHandler(logging
.StreamHandler
):
955 class ConfigDictTest(BaseTest
):
957 """Reading logging config from a dictionary."""
959 expected_log_pat
= r
"^([\w]+) \+\+ ([\w]+)$"
961 # config0 is a standard configuration.
966 'format' : '%(levelname)s ++ %(message)s',
971 'class' : 'logging.StreamHandler',
972 'formatter' : 'form1',
974 'stream' : 'ext://sys.stdout',
979 'handlers' : ['hand1'],
983 # config1 adds a little to the standard configuration.
988 'format' : '%(levelname)s ++ %(message)s',
993 'class' : 'logging.StreamHandler',
994 'formatter' : 'form1',
996 'stream' : 'ext://sys.stdout',
1000 'compiler.parser' : {
1002 'handlers' : ['hand1'],
1006 'level' : 'WARNING',
1010 # config2 has a subtle configuration error that should be reported
1015 'format' : '%(levelname)s ++ %(message)s',
1020 'class' : 'logging.StreamHandler',
1021 'formatter' : 'form1',
1023 'stream' : 'ext://sys.stdbout',
1027 'compiler.parser' : {
1029 'handlers' : ['hand1'],
1033 'level' : 'WARNING',
1037 #As config1 but with a misspelt level on a handler
1042 'format' : '%(levelname)s ++ %(message)s',
1047 'class' : 'logging.StreamHandler',
1048 'formatter' : 'form1',
1050 'stream' : 'ext://sys.stdout',
1054 'compiler.parser' : {
1056 'handlers' : ['hand1'],
1060 'level' : 'WARNING',
1065 #As config1 but with a misspelt level on a logger
1070 'format' : '%(levelname)s ++ %(message)s',
1075 'class' : 'logging.StreamHandler',
1076 'formatter' : 'form1',
1078 'stream' : 'ext://sys.stdout',
1082 'compiler.parser' : {
1084 'handlers' : ['hand1'],
1088 'level' : 'WRANING',
1092 # config3 has a less subtle configuration error
1097 'format' : '%(levelname)s ++ %(message)s',
1102 'class' : 'logging.StreamHandler',
1103 'formatter' : 'misspelled_name',
1105 'stream' : 'ext://sys.stdout',
1109 'compiler.parser' : {
1111 'handlers' : ['hand1'],
1115 'level' : 'WARNING',
1119 # config4 specifies a custom formatter class to be loaded
1124 '()' : __name__
+ '.ExceptionFormatter',
1125 'format' : '%(levelname)s:%(name)s:%(message)s',
1130 'class' : 'logging.StreamHandler',
1131 'formatter' : 'form1',
1133 'stream' : 'ext://sys.stdout',
1138 'handlers' : ['hand1'],
1142 # As config4 but using an actual callable rather than a string
1147 '()' : ExceptionFormatter
,
1148 'format' : '%(levelname)s:%(name)s:%(message)s',
1151 '()' : __name__
+ '.formatFunc',
1152 'format' : '%(levelname)s:%(name)s:%(message)s',
1156 'format' : '%(levelname)s:%(name)s:%(message)s',
1161 'class' : 'logging.StreamHandler',
1162 'formatter' : 'form1',
1164 'stream' : 'ext://sys.stdout',
1172 'handlers' : ['hand1'],
1176 # config5 specifies a custom handler class to be loaded
1181 'format' : '%(levelname)s ++ %(message)s',
1186 'class' : __name__
+ '.CustomHandler',
1187 'formatter' : 'form1',
1189 'stream' : 'ext://sys.stdout',
1193 'compiler.parser' : {
1195 'handlers' : ['hand1'],
1199 'level' : 'WARNING',
1203 # config6 specifies a custom handler class to be loaded
1204 # but has bad arguments
1209 'format' : '%(levelname)s ++ %(message)s',
1214 'class' : __name__
+ '.CustomHandler',
1215 'formatter' : 'form1',
1217 'stream' : 'ext://sys.stdout',
1218 '9' : 'invalid parameter name',
1222 'compiler.parser' : {
1224 'handlers' : ['hand1'],
1228 'level' : 'WARNING',
1232 #config 7 does not define compiler.parser but defines compiler.lexer
1233 #so compiler.parser should be disabled after applying it
1238 'format' : '%(levelname)s ++ %(message)s',
1243 'class' : 'logging.StreamHandler',
1244 'formatter' : 'form1',
1246 'stream' : 'ext://sys.stdout',
1250 'compiler.lexer' : {
1252 'handlers' : ['hand1'],
1256 'level' : 'WARNING',
1262 'disable_existing_loggers' : False,
1265 'format' : '%(levelname)s ++ %(message)s',
1270 'class' : 'logging.StreamHandler',
1271 'formatter' : 'form1',
1273 'stream' : 'ext://sys.stdout',
1279 'handlers' : ['hand1'],
1281 'compiler.lexer' : {
1285 'level' : 'WARNING',
1293 'format' : '%(levelname)s ++ %(message)s',
1298 'class' : 'logging.StreamHandler',
1299 'formatter' : 'form1',
1300 'level' : 'WARNING',
1301 'stream' : 'ext://sys.stdout',
1305 'compiler.parser' : {
1306 'level' : 'WARNING',
1307 'handlers' : ['hand1'],
1317 'incremental' : True,
1320 'level' : 'WARNING',
1324 'compiler.parser' : {
1332 'incremental' : True,
1339 'compiler.parser' : {
1345 #As config1 but with a filter added
1350 'format' : '%(levelname)s ++ %(message)s',
1355 'name' : 'compiler.parser',
1360 'class' : 'logging.StreamHandler',
1361 'formatter' : 'form1',
1363 'stream' : 'ext://sys.stdout',
1364 'filters' : ['filt1'],
1368 'compiler.parser' : {
1370 'filters' : ['filt1'],
1374 'level' : 'WARNING',
1375 'handlers' : ['hand1'],
1379 #As config1 but using cfg:// references
1382 'true_formatters': {
1384 'format' : '%(levelname)s ++ %(message)s',
1387 'handler_configs': {
1389 'class' : 'logging.StreamHandler',
1390 'formatter' : 'form1',
1392 'stream' : 'ext://sys.stdout',
1395 'formatters' : 'cfg://true_formatters',
1397 'hand1' : 'cfg://handler_configs[hand1]',
1400 'compiler.parser' : {
1402 'handlers' : ['hand1'],
1406 'level' : 'WARNING',
1410 #As config11 but missing the version key
1412 'true_formatters': {
1414 'format' : '%(levelname)s ++ %(message)s',
1417 'handler_configs': {
1419 'class' : 'logging.StreamHandler',
1420 'formatter' : 'form1',
1422 'stream' : 'ext://sys.stdout',
1425 'formatters' : 'cfg://true_formatters',
1427 'hand1' : 'cfg://handler_configs[hand1]',
1430 'compiler.parser' : {
1432 'handlers' : ['hand1'],
1436 'level' : 'WARNING',
1440 #As config11 but using an unsupported version
1443 'true_formatters': {
1445 'format' : '%(levelname)s ++ %(message)s',
1448 'handler_configs': {
1450 'class' : 'logging.StreamHandler',
1451 'formatter' : 'form1',
1453 'stream' : 'ext://sys.stdout',
1456 'formatters' : 'cfg://true_formatters',
1458 'hand1' : 'cfg://handler_configs[hand1]',
1461 'compiler.parser' : {
1463 'handlers' : ['hand1'],
1467 'level' : 'WARNING',
1471 def apply_config(self
, conf
):
1472 logging
.config
.dictConfig(conf
)
1474 def test_config0_ok(self
):
1475 # A simple config which overrides the default settings.
1476 with
captured_stdout() as output
:
1477 self
.apply_config(self
.config0
)
1478 logger
= logging
.getLogger()
1479 # Won't output anything
1480 logger
.info(self
.next_message())
1482 logger
.error(self
.next_message())
1483 self
.assert_log_lines([
1486 # Original logger output is empty.
1487 self
.assert_log_lines([])
1489 def test_config1_ok(self
, config
=config1
):
1490 # A config defining a sub-parser as well.
1491 with
captured_stdout() as output
:
1492 self
.apply_config(config
)
1493 logger
= logging
.getLogger("compiler.parser")
1494 # Both will output a message
1495 logger
.info(self
.next_message())
1496 logger
.error(self
.next_message())
1497 self
.assert_log_lines([
1501 # Original logger output is empty.
1502 self
.assert_log_lines([])
1504 def test_config2_failure(self
):
1505 # A simple config which overrides the default settings.
1506 self
.assertRaises(StandardError, self
.apply_config
, self
.config2
)
1508 def test_config2a_failure(self
):
1509 # A simple config which overrides the default settings.
1510 self
.assertRaises(StandardError, self
.apply_config
, self
.config2a
)
1512 def test_config2b_failure(self
):
1513 # A simple config which overrides the default settings.
1514 self
.assertRaises(StandardError, self
.apply_config
, self
.config2b
)
1516 def test_config3_failure(self
):
1517 # A simple config which overrides the default settings.
1518 self
.assertRaises(StandardError, self
.apply_config
, self
.config3
)
1520 def test_config4_ok(self
):
1521 # A config specifying a custom formatter class.
1522 with
captured_stdout() as output
:
1523 self
.apply_config(self
.config4
)
1524 #logger = logging.getLogger()
1526 raise RuntimeError()
1527 except RuntimeError:
1528 logging
.exception("just testing")
1530 self
.assertEquals(output
.getvalue(),
1531 "ERROR:root:just testing\nGot a [RuntimeError]\n")
1532 # Original logger output is empty
1533 self
.assert_log_lines([])
1535 def test_config4a_ok(self
):
1536 # A config specifying a custom formatter class.
1537 with
captured_stdout() as output
:
1538 self
.apply_config(self
.config4a
)
1539 #logger = logging.getLogger()
1541 raise RuntimeError()
1542 except RuntimeError:
1543 logging
.exception("just testing")
1545 self
.assertEquals(output
.getvalue(),
1546 "ERROR:root:just testing\nGot a [RuntimeError]\n")
1547 # Original logger output is empty
1548 self
.assert_log_lines([])
1550 def test_config5_ok(self
):
1551 self
.test_config1_ok(config
=self
.config5
)
1553 def test_config6_failure(self
):
1554 self
.assertRaises(StandardError, self
.apply_config
, self
.config6
)
1556 def test_config7_ok(self
):
1557 with
captured_stdout() as output
:
1558 self
.apply_config(self
.config1
)
1559 logger
= logging
.getLogger("compiler.parser")
1560 # Both will output a message
1561 logger
.info(self
.next_message())
1562 logger
.error(self
.next_message())
1563 self
.assert_log_lines([
1567 # Original logger output is empty.
1568 self
.assert_log_lines([])
1569 with
captured_stdout() as output
:
1570 self
.apply_config(self
.config7
)
1571 logger
= logging
.getLogger("compiler.parser")
1572 self
.assertTrue(logger
.disabled
)
1573 logger
= logging
.getLogger("compiler.lexer")
1574 # Both will output a message
1575 logger
.info(self
.next_message())
1576 logger
.error(self
.next_message())
1577 self
.assert_log_lines([
1581 # Original logger output is empty.
1582 self
.assert_log_lines([])
1584 #Same as test_config_7_ok but don't disable old loggers.
1585 def test_config_8_ok(self
):
1586 with
captured_stdout() as output
:
1587 self
.apply_config(self
.config1
)
1588 logger
= logging
.getLogger("compiler.parser")
1589 # Both will output a message
1590 logger
.info(self
.next_message())
1591 logger
.error(self
.next_message())
1592 self
.assert_log_lines([
1596 # Original logger output is empty.
1597 self
.assert_log_lines([])
1598 with
captured_stdout() as output
:
1599 self
.apply_config(self
.config8
)
1600 logger
= logging
.getLogger("compiler.parser")
1601 self
.assertFalse(logger
.disabled
)
1602 # Both will output a message
1603 logger
.info(self
.next_message())
1604 logger
.error(self
.next_message())
1605 logger
= logging
.getLogger("compiler.lexer")
1606 # Both will output a message
1607 logger
.info(self
.next_message())
1608 logger
.error(self
.next_message())
1609 self
.assert_log_lines([
1615 # Original logger output is empty.
1616 self
.assert_log_lines([])
1618 def test_config_9_ok(self
):
1619 with
captured_stdout() as output
:
1620 self
.apply_config(self
.config9
)
1621 logger
= logging
.getLogger("compiler.parser")
1622 #Nothing will be output since both handler and logger are set to WARNING
1623 logger
.info(self
.next_message())
1624 self
.assert_log_lines([], stream
=output
)
1625 self
.apply_config(self
.config9a
)
1626 #Nothing will be output since both handler is still set to WARNING
1627 logger
.info(self
.next_message())
1628 self
.assert_log_lines([], stream
=output
)
1629 self
.apply_config(self
.config9b
)
1630 #Message should now be output
1631 logger
.info(self
.next_message())
1632 self
.assert_log_lines([
1636 def test_config_10_ok(self
):
1637 with
captured_stdout() as output
:
1638 self
.apply_config(self
.config10
)
1639 logger
= logging
.getLogger("compiler.parser")
1640 logger
.warning(self
.next_message())
1641 logger
= logging
.getLogger('compiler')
1642 #Not output, because filtered
1643 logger
.warning(self
.next_message())
1644 logger
= logging
.getLogger('compiler.lexer')
1645 #Not output, because filtered
1646 logger
.warning(self
.next_message())
1647 logger
= logging
.getLogger("compiler.parser.codegen")
1648 #Output, as not filtered
1649 logger
.error(self
.next_message())
1650 self
.assert_log_lines([
1655 def test_config11_ok(self
):
1656 self
.test_config1_ok(self
.config11
)
1658 def test_config12_failure(self
):
1659 self
.assertRaises(StandardError, self
.apply_config
, self
.config12
)
1661 def test_config13_failure(self
):
1662 self
.assertRaises(StandardError, self
.apply_config
, self
.config13
)
1664 @unittest.skipUnless(threading
, 'listen() needs threading to work')
1665 def setup_via_listener(self
, text
):
1666 # Ask for a randomly assigned port (by using port 0)
1667 t
= logging
.config
.listen(0)
1670 # Now get the port allocated
1674 sock
= socket
.socket(socket
.AF_INET
, socket
.SOCK_STREAM
)
1675 sock
.settimeout(2.0)
1676 sock
.connect(('localhost', port
))
1678 slen
= struct
.pack('>L', len(text
))
1683 sent
= sock
.send(s
[sentsofar
:])
1689 logging
.config
.stopListening()
1692 def test_listen_config_10_ok(self
):
1693 with
captured_stdout() as output
:
1694 self
.setup_via_listener(json
.dumps(self
.config10
))
1695 logger
= logging
.getLogger("compiler.parser")
1696 logger
.warning(self
.next_message())
1697 logger
= logging
.getLogger('compiler')
1698 #Not output, because filtered
1699 logger
.warning(self
.next_message())
1700 logger
= logging
.getLogger('compiler.lexer')
1701 #Not output, because filtered
1702 logger
.warning(self
.next_message())
1703 logger
= logging
.getLogger("compiler.parser.codegen")
1704 #Output, as not filtered
1705 logger
.error(self
.next_message())
1706 self
.assert_log_lines([
1711 def test_listen_config_1_ok(self
):
1712 with
captured_stdout() as output
:
1713 self
.setup_via_listener(textwrap
.dedent(ConfigFileTest
.config1
))
1714 logger
= logging
.getLogger("compiler.parser")
1715 # Both will output a message
1716 logger
.info(self
.next_message())
1717 logger
.error(self
.next_message())
1718 self
.assert_log_lines([
1722 # Original logger output is empty.
1723 self
.assert_log_lines([])
1726 class ManagerTest(BaseTest
):
1727 def test_manager_loggerclass(self
):
1730 class MyLogger(logging
.Logger
):
1731 def _log(self
, level
, msg
, args
, exc_info
=None, extra
=None):
1734 man
= logging
.Manager(None)
1735 self
.assertRaises(TypeError, man
.setLoggerClass
, int)
1736 man
.setLoggerClass(MyLogger
)
1737 logger
= man
.getLogger('test')
1738 logger
.warning('should appear in logged')
1739 logging
.warning('should not appear in logged')
1741 self
.assertEqual(logged
, ['should appear in logged'])
1744 class ChildLoggerTest(BaseTest
):
1745 def test_child_loggers(self
):
1746 r
= logging
.getLogger()
1747 l1
= logging
.getLogger('abc')
1748 l2
= logging
.getLogger('def.ghi')
1749 c1
= r
.getChild('xyz')
1750 c2
= r
.getChild('uvw.xyz')
1751 self
.assertTrue(c1
is logging
.getLogger('xyz'))
1752 self
.assertTrue(c2
is logging
.getLogger('uvw.xyz'))
1753 c1
= l1
.getChild('def')
1754 c2
= c1
.getChild('ghi')
1755 c3
= l1
.getChild('def.ghi')
1756 self
.assertTrue(c1
is logging
.getLogger('abc.def'))
1757 self
.assertTrue(c2
is logging
.getLogger('abc.def.ghi'))
1758 self
.assertTrue(c2
is c3
)
1761 # Set the locale to the platform-dependent default. I have no idea
1762 # why the test does this, but in any case we save the current locale
1763 # first and restore it at the end.
1764 @run_with_locale('LC_ALL', '')
1766 run_unittest(BuiltinLevelsTest
, BasicFilterTest
,
1767 CustomLevelsAndFiltersTest
, MemoryHandlerTest
,
1768 ConfigFileTest
, SocketHandlerTest
, MemoryTest
,
1769 EncodingTest
, WarningsTest
, ConfigDictTest
, ManagerTest
,
1772 if __name__
== "__main__":