xref: /third_party/python/Lib/test/test_logging.py

87     def setUp(self):
90         self._threading_key = threading_helper.threading_setup()
95             self.saved_handlers = logging._handlers.copy()
96             self.saved_handler_list = logging._handlerList[:]
97             self.saved_loggers = saved_loggers = logger_dict.copy()
98             self.saved_name_to_level = logging._nameToLevel.copy()
99             self.saved_level_to_name = logging._levelToName.copy()
100             self.logger_states = logger_states = {}
108         self.logger1 = logging.getLogger("\xab\xd7\xbb")
109         self.logger2 = logging.getLogger("\u013f\u00d6\u0047")
111         self.root_logger = logging.getLogger("")
112         self.original_logging_level = self.root_logger.getEffectiveLevel()
114         self.stream = io.StringIO()
115         self.root_logger.setLevel(logging.DEBUG)
116         self.root_hdlr = logging.StreamHandler(self.stream)
117         self.root_formatter = logging.Formatter(self.log_format)
118         self.root_hdlr.setFormatter(self.root_formatter)
119         if self.logger1.hasHandlers():
120             hlist = self.logger1.handlers + self.root_logger.handlers
122         if self.logger2.hasHandlers():
123             hlist = self.logger2.handlers + self.root_logger.handlers
125         self.root_logger.addHandler(self.root_hdlr)
126         self.assertTrue(self.logger1.hasHandlers())
127         self.assertTrue(self.logger2.hasHandlers())
129     def tearDown(self):
132         self.stream.close()
133         self.root_logger.removeHandler(self.root_hdlr)
134         while self.root_logger.handlers:
135             h = self.root_logger.handlers[0]
136             self.root_logger.removeHandler(h)
138         self.root_logger.setLevel(self.original_logging_level)
142             logging._levelToName.update(self.saved_level_to_name)
144             logging._nameToLevel.update(self.saved_name_to_level)
146             logging._handlers.update(self.saved_handlers)
147             logging._handlerList[:] = self.saved_handler_list
152             loggerDict.update(self.saved_loggers)
153             logger_states = self.logger_states
154             for name in self.logger_states:
156                     self.saved_loggers[name].disabled = logger_states[name]
160         self.doCleanups()
161         threading_helper.threading_cleanup(*self._threading_key)
163     def assert_log_lines(self, expected_values, stream=None, pat=None):
165         self.expected_log_pat, and compare the extracted group values to
167         stream = stream or self.stream
168         pat = re.compile(pat or self.expected_log_pat)
170         self.assertEqual(len(actual_lines), len(expected_values))
174                 self.fail("Log line does not match expected pattern:\n" +
176             self.assertEqual(tuple(match.groups()), expected)
179             self.fail("Remaining output at end of log stream:\n" + s)
181     def next_message(self):
184         self.message_num += 1
185         return "%d" % self.message_num
191     def test_flat(self):
193         m = self.next_message
224         self.assert_log_lines([
238     def test_nested_explicit(self):
240         m = self.next_message
256         self.assert_log_lines([
261     def test_nested_inherited(self):
263         m = self.next_message
287         self.assert_log_lines([
296     def test_nested_with_virtual_parent(self):
298         m = self.next_message
315         self.assert_log_lines([
322     def test_regression_22386(self):
324         self.assertEqual(logging.getLevelName('INFO'), logging.INFO)
325         self.assertEqual(logging.getLevelName(logging.INFO), 'INFO')
327     def test_issue27935(self):
329         self.assertEqual(fatal, logging.FATAL)
331     def test_regression_29220(self):
334         self.addCleanup(logging.addLevelName, logging.INFO, 'INFO')
335         self.assertEqual(logging.getLevelName(logging.INFO), '')
336         self.assertEqual(logging.getLevelName(logging.NOTSET), 'NOTSET')
337         self.assertEqual(logging.getLevelName('NOTSET'), logging.NOTSET)
343     def test_filter(self):
347         handler = self.root_logger.handlers[0]
355             spam.info(self.next_message())
356             spam_eggs.info(self.next_message())  # Good.
357             spam_eggs_fish.info(self.next_message())  # Good.
358             spam_bakedbeans.info(self.next_message())
360             self.assert_log_lines([
367     def test_callable_filter(self):
376         handler = self.root_logger.handlers[0]
384             spam.info(self.next_message())
385             spam_eggs.info(self.next_message())  # Good.
386             spam_eggs_fish.info(self.next_message())  # Good.
387             spam_bakedbeans.info(self.next_message())
389             self.assert_log_lines([
396     def test_empty_filter(self):
399         self.assertTrue(f.filter(r))
443     def filter(self, record):
450     def filter(self, record):
461     def setUp(self):
462         BaseTest.setUp(self)
466     def log_at_all_levels(self, logger):
468             logger.log(lvl, self.next_message())
470     def test_logger_filter(self):
472         self.root_logger.setLevel(VERBOSE)
474         self.log_at_all_levels(self.root_logger)
475         self.assert_log_lines([
484     def test_handler_filter(self):
486         self.root_logger.handlers[0].setLevel(SOCIABLE)
489             self.log_at_all_levels(self.root_logger)
490             self.assert_log_lines([
498             self.root_logger.handlers[0].setLevel(logging.NOTSET)
500     def test_specific_filters(self):
503         handler = self.root_logger.handlers[0]
508             self.log_at_all_levels(self.root_logger)
521             self.assert_log_lines(first_lines)
524             self.root_logger.addFilter(specific_filter)
525             self.log_at_all_levels(self.root_logger)
526             self.assert_log_lines(first_lines + [
539                 self.root_logger.removeFilter(specific_filter)
544     def test_name(self):
547         self.assertEqual(h.name, 'generic')
549         self.assertEqual(h.name, 'anothergeneric')
550         self.assertRaises(NotImplementedError, h.emit, None)
552     def test_builtin_handlers(self):
564                     self.assertEqual(dev, -1)
565                     self.assertEqual(ino, -1)
570                     self.assertFalse(os.path.exists(fn))
573                     self.assertTrue(os.path.exists(fn))
575                     self.assertEqual(h.dev, -1)
576                     self.assertEqual(h.ino, -1)
586                 self.assertEqual(h.facility, h.LOG_USER)
587                 self.assertTrue(h.unixsocket)
593                 self.assertRaises(ValueError, logging.handlers.HTTPHandler,
600         self.assertTrue(h.shouldFlush(r))
603         self.assertFalse(h.shouldFlush(r))
606     def test_path_objects(self):
625             self.assertTrue(os.path.exists(fn))
634     def test_race(self):
640                     self.deletion_time = time.time()
648         self.handle_time = None
649         self.deletion_time = None
665                         self.handle_time = time.time()
669                               'opened at %s' % (self.deletion_time,
670                                                 self.handle_time))
684     def test_post_fork_child_no_deadlock(self):
687             def __init__(self):
689                 self.sub_handler = logging.StreamHandler(
692             def emit(self, record):
693                 self.sub_handler.acquire()
695                     self.sub_handler.emit(record)
697                     self.sub_handler.release()
699         self.assertEqual(len(logging._handlers), 0)
701         self.addCleanup(refed_h.sub_handler.stream.close)
703         self.assertGreater(len(logging._handlers), 0)
704         self.assertGreater(len(logging._at_fork_reinit_lock_weakset), 1)
760     def write(self, data):
764     def handleError(self, record):
765         self.error_record = record
772     def test_error_handling(self):
779             self.assertIs(h.error_record, r)
785                 self.assertIn(msg, stderr.getvalue())
790                 self.assertEqual('', stderr.getvalue())
794     def test_stream_setting(self):
801         self.assertIs(old, sys.stderr)
803         self.assertIs(actual, stream)
806         self.assertIsNone(actual)
808     def test_can_represent_stream_with_int_name(self):
810         self.assertEqual(repr(h), '<StreamHandler 2 (NOTSET)>')
836     def __init__(self, addr, handler, poll_interval, sockmap):
837         smtpd.SMTPServer.__init__(self, addr, None, map=sockmap,
839         self.port = self.socket.getsockname()[1]
840         self._handler = handler
841         self._thread = None
842         self._quit = False
843         self.poll_interval = poll_interval
845     def process_message(self, peer, mailfrom, rcpttos, data):
855         self._handler(peer, mailfrom, rcpttos, data)
857     def start(self):
861         self._thread = t = threading.Thread(target=self.serve_forever,
862                                             args=(self.poll_interval,))
866     def serve_forever(self, poll_interval):
874         while not self._quit:
875             asyncore.loop(poll_interval, map=self._map, count=1)
877     def stop(self):
882         self._quit = True
883         threading_helper.join_thread(self._thread)
884         self._thread = None
885         self.close()
886         asyncore.close_all(map=self._map, ignore_all=True)
905     def __init__(self, handler, poll_interval):
906         self._thread = None
907         self.poll_interval = poll_interval
908         self._handler = handler
909         self.ready = threading.Event()
911     def start(self):
915         self._thread = t = threading.Thread(target=self.serve_forever,
916                                             args=(self.poll_interval,))
920     def serve_forever(self, poll_interval):
925         self.ready.set()
926         super(ControlMixin, self).serve_forever(poll_interval)
928     def stop(self):
932         self.shutdown()
933         if self._thread is not None:
934             threading_helper.join_thread(self._thread)
935             self._thread = None
936         self.server_close()
937         self.ready.clear()
950     def __init__(self, addr, handler, poll_interval=0.5,
953             def __getattr__(self, name, default=None):
955                     return self.process_request
958             def process_request(self):
959                 self.server._handler(self)
961             def log_message(self, format, *args):
964                           self).log_message(format, *args)
965         HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)
966         ControlMixin.__init__(self, handler, poll_interval)
967         self.sslctx = sslctx
969     def get_request(self):
971             sock, addr = self.socket.accept()
972             if self.sslctx:
973                 sock = self.sslctx.wrap_socket(sock, server_side=True)
997     def __init__(self, addr, handler, poll_interval=0.5,
1001             def handle(self):
1002                 self.server._handler(self)
1003         ThreadingTCPServer.__init__(self, addr, DelegatingTCPRequestHandler,
1005         ControlMixin.__init__(self, handler, poll_interval)
1007     def server_bind(self):
1008         super(TestTCPServer, self).server_bind()
1009         self.port = self.socket.getsockname()[1]
1028     def __init__(self, addr, handler, poll_interval=0.5,
1032             def handle(self):
1033                 self.server._handler(self)
1035             def finish(self):
1036                 data = self.wfile.getvalue()
1039                         super(DelegatingUDPRequestHandler, self).finish()
1041                         if not self.server._closed:
1044         ThreadingUDPServer.__init__(self, addr,
1047         ControlMixin.__init__(self, handler, poll_interval)
1048         self._closed = False
1050     def server_bind(self):
1051         super(TestUDPServer, self).server_bind()
1052         self.port = self.socket.getsockname()[1]
1054     def server_close(self):
1055         super(TestUDPServer, self).server_close()
1056         self._closed = True
1073     def test_basic(self):
1075         server = TestSMTPServer((socket_helper.HOST, 0), self.process_message, 0.001,
1080                                          timeout=self.TIMEOUT)
1081         self.assertEqual(h.toaddrs, ['you'])
1082         self.messages = []
1084         self.handled = threading.Event()
1086         self.handled.wait(self.TIMEOUT)
1088         self.assertTrue(self.handled.is_set())
1089         self.assertEqual(len(self.messages), 1)
1090         peer, mailfrom, rcpttos, data = self.messages[0]
1091         self.assertEqual(mailfrom, 'me')
1092         self.assertEqual(rcpttos, ['you'])
1093         self.assertIn('\nSubject: Log\n', data)
1094         self.assertTrue(data.endswith('\n\nHello \u2713'))
1097     def process_message(self, *args):
1098         self.messages.append(args)
1099         self.handled.set()
1108     def setUp(self):
1109         BaseTest.setUp(self)
1110         self.mem_hdlr = logging.handlers.MemoryHandler(10, logging.WARNING,
1111                                                        self.root_hdlr)
1112         self.mem_logger = logging.getLogger('mem')
1113         self.mem_logger.propagate = 0
1114         self.mem_logger.addHandler(self.mem_hdlr)
1116     def tearDown(self):
1117         self.mem_hdlr.close()
1118         BaseTest.tearDown(self)
1120     def test_flush(self):
1123         self.mem_logger.debug(self.next_message())
1124         self.assert_log_lines([])
1125         self.mem_logger.info(self.next_message())
1126         self.assert_log_lines([])
1128         self.mem_logger.warning(self.next_message())
1134         self.assert_log_lines(lines)
1137                 self.mem_logger.debug(self.next_message())
1138             self.assert_log_lines(lines)
1141             self.mem_logger.debug(self.next_message())
1143             self.assert_log_lines(lines)
1145         self.mem_logger.debug(self.next_message())
1146         self.assert_log_lines(lines)
1148     def test_flush_on_close(self):
1152         self.mem_logger.debug(self.next_message())
1153         self.assert_log_lines([])
1154         self.mem_logger.info(self.next_message())
1155         self.assert_log_lines([])
1156         self.mem_logger.removeHandler(self.mem_hdlr)
1158         self.mem_hdlr.close()
1163         self.assert_log_lines(lines)
1165         self.mem_hdlr = logging.handlers.MemoryHandler(10, logging.WARNING,
1166                                                        self.root_hdlr,
1168         self.mem_logger.addHandler(self.mem_hdlr)
1169         self.mem_logger.debug(self.next_message())
1170         self.assert_log_lines(lines)  # no change
1171         self.mem_logger.info(self.next_message())
1172         self.assert_log_lines(lines)  # no change
1173         self.mem_logger.removeHandler(self.mem_hdlr)
1174         self.mem_hdlr.close()
1176         self.assert_log_lines(lines)  # no change
1179     def test_race_between_set_target_and_flush(self):
1181             def __init__(self, mem_hdlr):
1182                 self.mem_hdlr = mem_hdlr
1183                 self.threads = []
1185             def removeTarget(self):
1186                 self.mem_hdlr.setTarget(None)
1188             def handle(self, msg):
1189                 thread = threading.Thread(target=self.removeTarget)
1190                 self.threads.append(thread)
1193         target = MockRaceConditionHandler(self.mem_hdlr)
1195             self.mem_hdlr.setTarget(target)
1199                 self.mem_logger.info("not flushed")
1200                 self.mem_logger.warning("flushed")
1208     def formatException(self, ei):
1468     def apply_config(self, conf, **kwargs):
1472     def test_config0_ok(self):
1475             self.apply_config(self.config0)
1478             logger.info(self.next_message())
1480             logger.error(self.next_message())
1481             self.assert_log_lines([
1485             self.assert_log_lines([])
1487     def test_config0_using_cp_ok(self):
1490             file = io.StringIO(textwrap.dedent(self.config0))
1496             logger.info(self.next_message())
1498             logger.error(self.next_message())
1499             self.assert_log_lines([
1503             self.assert_log_lines([])
1505     def test_config1_ok(self, config=config1):
1508             self.apply_config(config)
1511             logger.info(self.next_message())
1512             logger.error(self.next_message())
1513             self.assert_log_lines([
1518             self.assert_log_lines([])
1520     def test_config2_failure(self):
1522         self.assertRaises(Exception, self.apply_config, self.config2)
1524     def test_config3_failure(self):
1526         self.assertRaises(Exception, self.apply_config, self.config3)
1528     def test_config4_ok(self):
1531             self.apply_config(self.config4)
1538             self.assertEqual(output.getvalue(),
1541             self.assert_log_lines([])
1543     def test_config5_ok(self):
1544         self.test_config1_ok(config=self.config5)
1546     def test_config6_ok(self):
1547         self.test_config1_ok(config=self.config6)
1549     def test_config7_ok(self):
1551             self.apply_config(self.config1a)
1560             logger.info(self.next_message())
1561             logger.error(self.next_message())
1562             hyphenated.critical(self.next_message())
1563             self.assert_log_lines([
1569             self.assert_log_lines([])
1571             self.apply_config(self.config7)
1573             self.assertFalse(logger.disabled)
1575             logger.info(self.next_message())
1576             logger.error(self.next_message())
1579             logger.info(self.next_message())
1580             logger.error(self.next_message())
1582             hyphenated.critical(self.next_message())
1583             self.assert_log_lines([
1590             self.assert_log_lines([])
1592     def test_config8_ok(self):
1598         with self.check_no_resource_warning():
1607             config8 = self.config8.format(tempfile=fn)
1609             self.apply_config(config8)
1610             self.apply_config(config8)
1613         self.addCleanup(cleanup, handler, fn)
1615     def test_logger_disabling(self):
1616         self.apply_config(self.disable_test)
1618         self.assertFalse(logger.disabled)
1619         self.apply_config(self.disable_test)
1620         self.assertTrue(logger.disabled)
1621         self.apply_config(self.disable_test, disable_existing_loggers=False)
1622         self.assertFalse(logger.disabled)
1624     def test_config_set_handler_names(self):
1645         self.apply_config(test_config)
1646         self.assertEqual(logging.getLogger().handlers[0].name, 'hand1')
1648     def test_exception_if_confg_file_is_invalid(self):
1673         self.assertRaises(RuntimeError, logging.config.fileConfig, file)
1675     def test_exception_if_confg_file_is_empty(self):
1678         self.assertRaises(RuntimeError, logging.config.fileConfig, fn)
1681     def test_exception_if_config_file_does_not_exist(self):
1682         self.assertRaises(FileNotFoundError, logging.config.fileConfig, 'filenotfound')
1684     def test_defaults_do_no_interpolation(self):
1738     def setUp(self):
1741         BaseTest.setUp(self)
1743         self.server = self.sock_hdlr = self.server_exception = None
1745             self.server = server = self.server_class(self.address,
1746                                                      self.handle_socket, 0.01)
1751             self.server_exception = e
1756             self.sock_hdlr = hcls('localhost', server.port)
1758             self.sock_hdlr = hcls(server.server_address, None)
1759         self.log_output = ''
1760         self.root_logger.removeHandler(self.root_logger.handlers[0])
1761         self.root_logger.addHandler(self.sock_hdlr)
1762         self.handled = threading.Semaphore(0)
1764     def tearDown(self):
1767             if self.sock_hdlr:
1768                 self.root_logger.removeHandler(self.sock_hdlr)
1769                 self.sock_hdlr.close()
1770             if self.server:
1771                 self.server.stop()
1773             BaseTest.tearDown(self)
1775     def handle_socket(self, request):
1787             self.log_output += record.msg + '\n'
1788             self.handled.release()
1790     def test_output(self):
1792         if self.server_exception:
1793             self.skipTest(self.server_exception)
1796         self.handled.acquire()
1798         self.handled.acquire()
1799         self.assertEqual(self.log_output, "spam\neggs\n")
1801     def test_noserver(self):
1802         if self.server_exception:
1803             self.skipTest(self.server_exception)
1806         self.sock_hdlr.retryStart = 2.5
1808         self.server.stop()
1813             self.root_logger.exception('Never sent')
1814         self.root_logger.error('Never sent, either')
1816         self.assertGreater(self.sock_hdlr.retryTime, now)
1817         time.sleep(self.sock_hdlr.retryTime - now + 0.001)
1818         self.root_logger.error('Nor this')
1836     def setUp(self):
1838         self.address = _get_temp_domain_socket()
1839         SocketHandlerTest.setUp(self)
1841     def tearDown(self):
1842         SocketHandlerTest.tearDown(self)
1843         os_helper.unlink(self.address)
1854     def setUp(self):
1857         BaseTest.setUp(self)
1859         self.server = self.sock_hdlr = self.server_exception = None
1861             self.server = server = self.server_class(self.address,
1862                                                      self.handle_datagram, 0.01)
1867             self.server_exception = e
1872             self.sock_hdlr = hcls('localhost', server.port)
1874             self.sock_hdlr = hcls(server.server_address, None)
1875         self.log_output = ''
1876         self.root_logger.removeHandler(self.root_logger.handlers[0])
1877         self.root_logger.addHandler(self.sock_hdlr)
1878         self.handled = threading.Event()
1880     def tearDown(self):
1883             if self.server:
1884                 self.server.stop()
1885             if self.sock_hdlr:
1886                 self.root_logger.removeHandler(self.sock_hdlr)
1887                 self.sock_hdlr.close()
1889             BaseTest.tearDown(self)
1891     def handle_datagram(self, request):
1896         self.log_output += record.msg + '\n'
1897         self.handled.set()
1899     def test_output(self):
1901         if self.server_exception:
1902             self.skipTest(self.server_exception)
1905         self.handled.wait()
1906         self.handled.clear()
1908         self.handled.wait()
1909         self.assertEqual(self.log_output, "spam\neggs\n")
1919     def setUp(self):
1921         self.address = _get_temp_domain_socket()
1922         DatagramHandlerTest.setUp(self)
1924     def tearDown(self):
1925         DatagramHandlerTest.tearDown(self)
1926         os_helper.unlink(self.address)
1937     def setUp(self):
1940         BaseTest.setUp(self)
1942         self.server = self.sl_hdlr = self.server_exception = None
1944             self.server = server = self.server_class(self.address,
1945                                                      self.handle_datagram, 0.01)
1950             self.server_exception = e
1955             self.sl_hdlr = hcls((server.server_address[0], server.port))
1957             self.sl_hdlr = hcls(server.server_address)
1958         self.log_output = b''
1959         self.root_logger.removeHandler(self.root_logger.handlers[0])
1960         self.root_logger.addHandler(self.sl_hdlr)
1961         self.handled = threading.Event()
1963     def tearDown(self):
1966             if self.server:
1967                 self.server.stop()
1968             if self.sl_hdlr:
1969                 self.root_logger.removeHandler(self.sl_hdlr)
1970                 self.sl_hdlr.close()
1972             BaseTest.tearDown(self)
1974     def handle_datagram(self, request):
1975         self.log_output = request.packet
1976         self.handled.set()
1978     def test_output(self):
1979         if self.server_exception:
1980             self.skipTest(self.server_exception)
1984         self.handled.wait()
1985         self.assertEqual(self.log_output, b'<11>sp\xc3\xa4m\x00')
1986         self.handled.clear()
1987         self.sl_hdlr.append_nul = False
1989         self.handled.wait()
1990         self.assertEqual(self.log_output, b'<11>sp\xc3\xa4m')
1991         self.handled.clear()
1992         self.sl_hdlr.ident = "h\xe4m-"
1994         self.handled.wait()
1995         self.assertEqual(self.log_output, b'<11>h\xc3\xa4m-sp\xc3\xa4m')
1997     def test_udp_reconnection(self):
1999         self.sl_hdlr.close()
2000         self.handled.clear()
2002         self.handled.wait(0.1)
2003         self.assertEqual(self.log_output, b'<11>sp\xc3\xa4m\x00')
2013     def setUp(self):
2015         self.address = _get_temp_domain_socket()
2016         SysLogHandlerTest.setUp(self)
2018     def tearDown(self):
2019         SysLogHandlerTest.tearDown(self)
2020         os_helper.unlink(self.address)
2031     def setUp(self):
2032         self.server_class.address_family = socket.AF_INET6
2033         super(IPv6SysLogHandlerTest, self).setUp()
2035     def tearDown(self):
2036         self.server_class.address_family = socket.AF_INET
2037         super(IPv6SysLogHandlerTest, self).tearDown()
2044     def setUp(self):
2047         BaseTest.setUp(self)
2048         self.handled = threading.Event()
2050     def handle_request(self, request):
2051         self.command = request.command
2052         self.log_data = urlparse(request.path)
2053         if self.command == 'POST':
2056                 self.post_data = request.rfile.read(rlen)
2058                 self.post_data = None
2061         self.handled.set()
2063     def test_output(self):
2066         root_logger = self.root_logger
2067         root_logger.removeHandler(self.root_logger.handlers[0])
2085             self.server = server = TestHTTPServer(addr, self.handle_request,
2091             self.h_hdlr = logging.handlers.HTTPHandler(host, '/frob',
2095             self.log_data = None
2096             root_logger.addHandler(self.h_hdlr)
2099                 self.h_hdlr.method = method
2100                 self.handled.clear()
2103                 self.handled.wait()
2104                 self.assertEqual(self.log_data.path, '/frob')
2105                 self.assertEqual(self.command, method)
2107                     d = parse_qs(self.log_data.query)
2109                     d = parse_qs(self.post_data.decode('utf-8'))
2110                 self.assertEqual(d['name'], ['http'])
2111                 self.assertEqual(d['funcName'], ['test_output'])
2112                 self.assertEqual(d['msg'], [msg])
2114             self.server.stop()
2115             self.root_logger.removeHandler(self.h_hdlr)
2116             self.h_hdlr.close()
2122     def setUp(self):
2124         BaseTest.setUp(self)
2125         self._survivors = {}
2127     def _watch_for_survival(self, *args):
2132             self._survivors[key] = weakref.ref(obj)
2134     def _assertTruesurvival(self):
2139         for (id_, repr_), ref in self._survivors.items():
2143             self.fail("%d objects should have survived "
2146     def test_persistent_loggers(self):
2149         self.root_logger.setLevel(logging.INFO)
2151         self._watch_for_survival(foo)
2153         self.root_logger.debug(self.next_message())
2154         foo.debug(self.next_message())
2155         self.assert_log_lines([
2160         self._assertTruesurvival()
2163         bar.debug(self.next_message())
2164         self.assert_log_lines([
2171     def test_encoding_plain_file(self):
2190                 self.assertEqual(f.read().rstrip(), data)
2197     def test_encoding_cyrillic_unicode(self):
2216         self.assertEqual(s, b'\xe4\xee \xf1\xe2\xe8\xe4\xe0\xed\xe8\xff\n')
2221     def test_warnings(self):
2224             self.addCleanup(logging.captureWarnings, False)
2234             self.assertGreater(s.find("UserWarning: I'm warning you...\n"), 0)
2242             self.assertEqual(s,
2245     def test_warnings_no_handlers(self):
2248             self.addCleanup(logging.captureWarnings, False)
2252             self.assertEqual(logger.handlers, [])
2255             self.assertEqual(len(logger.handlers), 1)
2256             self.assertIsInstance(logger.handlers[0], logging.NullHandler)
2263     def __init__(self, fmt, datefmt=None):
3014     def apply_config(self, conf):
3017     def test_config0_ok(self):
3020             self.apply_config(self.config0)
3023             logger.info(self.next_message())
3025             logger.error(self.next_message())
3026             self.assert_log_lines([
3030             self.assert_log_lines([])
3032     def test_config1_ok(self, config=config1):
3035             self.apply_config(config)
3038             logger.info(self.next_message())
3039             logger.error(self.next_message())
3040             self.assert_log_lines([
3045             self.assert_log_lines([])
3047     def test_config2_failure(self):
3049         self.assertRaises(Exception, self.apply_config, self.config2)
3051     def test_config2a_failure(self):
3053         self.assertRaises(Exception, self.apply_config, self.config2a)
3055     def test_config2b_failure(self):
3057         self.assertRaises(Exception, self.apply_config, self.config2b)
3059     def test_config3_failure(self):
3061         self.assertRaises(Exception, self.apply_config, self.config3)
3063     def test_config4_ok(self):
3066             self.apply_config(self.config4)
3073             self.assertEqual(output.getvalue(),
3076             self.assert_log_lines([])
3078     def test_config4a_ok(self):
3081             self.apply_config(self.config4a)
3088             self.assertEqual(output.getvalue(),
3091             self.assert_log_lines([])
3093     def test_config5_ok(self):
3094         self.test_config1_ok(config=self.config5)
3096     def test_config6_failure(self):
3097         self.assertRaises(Exception, self.apply_config, self.config6)
3099     def test_config7_ok(self):
3101             self.apply_config(self.config1)
3104             logger.info(self.next_message())
3105             logger.error(self.next_message())
3106             self.assert_log_lines([
3111             self.assert_log_lines([])
3113             self.apply_config(self.config7)
3115             self.assertTrue(logger.disabled)
3118             logger.info(self.next_message())
3119             logger.error(self.next_message())
3120             self.assert_log_lines([
3125             self.assert_log_lines([])
3128     def test_config_8_ok(self):
3130             self.apply_config(self.config1)
3133             logger.info(self.next_message())
3134             logger.error(self.next_message())
3135             self.assert_log_lines([
3140             self.assert_log_lines([])
3142             self.apply_config(self.config8)
3144             self.assertFalse(logger.disabled)
3146             logger.info(self.next_message())
3147             logger.error(self.next_message())
3150             logger.info(self.next_message())
3151             logger.error(self.next_message())
3152             self.assert_log_lines([
3159             self.assert_log_lines([])
3161     def test_config_8a_ok(self):
3163             self.apply_config(self.config1a)
3172             logger.info(self.next_message())
3173             logger.error(self.next_message())
3174             hyphenated.critical(self.next_message())
3175             self.assert_log_lines([
3181             self.assert_log_lines([])
3183             self.apply_config(self.config8a)
3185             self.assertFalse(logger.disabled)
3187             logger.info(self.next_message())
3188             logger.error(self.next_message())
3191             logger.info(self.next_message())
3192             logger.error(self.next_message())
3194             hyphenated.critical(self.next_message())
3195             self.assert_log_lines([
3202             self.assert_log_lines([])
3204     def test_config_9_ok(self):
3206             self.apply_config(self.config9)
3209             logger.info(self.next_message())
3210             self.assert_log_lines([], stream=output)
3211             self.apply_config(self.config9a)
3213             logger.info(self.next_message())
3214             self.assert_log_lines([], stream=output)
3215             self.apply_config(self.config9b)
3217             logger.info(self.next_message())
3218             self.assert_log_lines([
3222     def test_config_10_ok(self):
3224             self.apply_config(self.config10)
3226             logger.warning(self.next_message())
3229             logger.warning(self.next_message())
3232             logger.warning(self.next_message())
3235             logger.error(self.next_message())
3236             self.assert_log_lines([
3241     def test_config11_ok(self):
3242         self.test_config1_ok(self.config11)
3244     def test_config12_failure(self):
3245         self.assertRaises(Exception, self.apply_config, self.config12)
3247     def test_config13_failure(self):
3248         self.assertRaises(Exception, self.apply_config, self.config13)
3250     def test_config14_ok(self):
3252             self.apply_config(self.config14)
3254             self.assertEqual(h.foo, 'bar')
3255             self.assertEqual(h.terminator, '!\n')
3257             self.assertTrue(output.getvalue().endswith('Exclamation!\n'))
3259     def test_config15_ok(self):
3265         with self.check_no_resource_warning():
3283             self.apply_config(config)
3284             self.apply_config(config)
3287         self.addCleanup(cleanup, handler, fn)
3289     def setup_via_listener(self, text, verify=None):
3318     def test_listen_config_10_ok(self):
3320             self.setup_via_listener(json.dumps(self.config10))
3322             logger.warning(self.next_message())
3325             logger.warning(self.next_message())
3328             logger.warning(self.next_message())
3331             logger.error(self.next_message())
3332             self.assert_log_lines([
3338     def test_listen_config_1_ok(self):
3340             self.setup_via_listener(textwrap.dedent(ConfigFileTest.config1))
3343             logger.info(self.next_message())
3344             logger.error(self.next_message())
3345             self.assert_log_lines([
3350             self.assert_log_lines([])
3353     def test_listen_verify(self):
3367             self.setup_via_listener(to_send, verify_fail)
3369             logger.info(self.next_message())
3370             logger.error(self.next_message())
3371         self.assert_log_lines([], stream=output)
3373         self.assert_log_lines([
3382             self.setup_via_listener(to_send)    # no verify callable specified
3385             logger.info(self.next_message())
3386             logger.error(self.next_message())
3387         self.assert_log_lines([
3392         self.assert_log_lines([
3400             self.setup_via_listener(to_send[::-1], verify_reverse)
3403             logger.info(self.next_message())
3404             logger.error(self.next_message())
3405         self.assert_log_lines([
3410         self.assert_log_lines([
3415     def test_out_of_order(self):
3416         self.assertRaises(ValueError, self.apply_config, self.out_of_order)
3418     def test_out_of_order_with_dollar_style(self):
3419         config = copy.deepcopy(self.out_of_order)
3422         self.apply_config(config)
3424         self.assertIsInstance(handler.target, logging.Handler)
3425         self.assertIsInstance(handler.formatter._style,
3428     def test_custom_formatter_class_with_validate(self):
3429         self.apply_config(self.custom_formatter_class_validate)
3431         self.assertIsInstance(handler.formatter, ExceptionFormatter)
3433     def test_custom_formatter_class_with_validate2(self):
3434         self.apply_config(self.custom_formatter_class_validate2)
3436         self.assertIsInstance(handler.formatter, ExceptionFormatter)
3438     def test_custom_formatter_class_with_validate2_with_wrong_fmt(self):
3439         config = self.custom_formatter_class_validate.copy()
3443         self.apply_config(config)
3445         self.assertIsInstance(handler.formatter, ExceptionFormatter)
3447     def test_custom_formatter_class_with_validate3(self):
3448         self.assertRaises(ValueError, self.apply_config, self.custom_formatter_class_validate3)
3450     def test_custom_formatter_function_with_validate(self):
3451         self.assertRaises(ValueError, self.apply_config, self.custom_formatter_with_function)
3453     def test_baseconfig(self):
3463         self.assertEqual(bc.convert('cfg://atuple[1]'), 2)
3464         self.assertEqual(bc.convert('cfg://alist[1]'), 'b')
3465         self.assertEqual(bc.convert('cfg://nest1[1][0]'), 'h')
3466         self.assertEqual(bc.convert('cfg://nest2[1][1]'), 'm')
3467         self.assertEqual(bc.convert('cfg://adict.d'), 'e')
3468         self.assertEqual(bc.convert('cfg://adict[f]'), 3)
3470         self.assertEqual(v.pop(1), ['a', 'b', 'c'])
3471         self.assertRaises(KeyError, bc.convert, 'cfg://nosuch')
3472         self.assertRaises(ValueError, bc.convert, 'cfg://!')
3473         self.assertRaises(KeyError, bc.convert, 'cfg://adict[2]')
3475     def test_namedtuple(self):
3480             def __init__(self, resource, *args, **kwargs):
3482                 self.resource: namedtuple = resource
3484             def emit(self, record):
3485                 record.msg += f' {self.resource.type}'
3502             self.apply_config(config)
3504         self.assertEqual(stderr.getvalue(), 'some log my_type\n')
3506     def test_config_callable_filter_works(self):
3509         self.apply_config({
3515     def test_config_filter_works(self):
3517         self.apply_config({
3523     def test_config_filter_method_works(self):
3525             def filter(self, _):
3528         self.apply_config({
3534     def test_invalid_type_raises(self):
3537             self.assertRaises(
3539                 self.apply_config,
3543     def test_90195(self):
3562         self.assertFalse(logger.disabled)
3563         self.apply_config(config)
3564         self.assertFalse(logger.disabled)
3566         self.apply_config({'version': 1})
3567         self.assertTrue(logger.disabled)
3569         self.apply_config(config)
3571         self.assertFalse(logger.disabled)
3574     def test_manager_loggerclass(self):
3578             def _log(self, level, msg, args, exc_info=None, extra=None):
3582         self.assertRaises(TypeError, man.setLoggerClass, int)
3588         self.assertEqual(logged, ['should appear in logged'])
3590     def test_set_log_record_factory(self):
3594         self.assertEqual(man.logRecordFactory, expected)
3597     def test_child_loggers(self):
3603         self.assertIs(c1, logging.getLogger('xyz'))
3604         self.assertIs(c2, logging.getLogger('uvw.xyz'))
3608         self.assertIs(c1, logging.getLogger('abc.def'))
3609         self.assertIs(c2, logging.getLogger('abc.def.ghi'))
3610         self.assertIs(c2, c3)
3618     def setUp(self):
3620             def __init__(self, cls):
3621                 self.cls = cls
3623             def filter(self, record):
3625                 if t is not self.cls:
3627                             self.cls)
3631         BaseTest.setUp(self)
3632         self.filter = CheckingFilter(DerivedLogRecord)
3633         self.root_logger.addFilter(self.filter)
3634         self.orig_factory = logging.getLogRecordFactory()
3636     def tearDown(self):
3637         self.root_logger.removeFilter(self.filter)
3638         BaseTest.tearDown(self)
3639         logging.setLogRecordFactory(self.orig_factory)
3641     def test_logrecord_class(self):
3642         self.assertRaises(TypeError, self.root_logger.warning,
3643                           self.next_message())
3645         self.root_logger.error(self.next_message())
3646         self.assert_log_lines([
3656     def setUp(self):
3657         BaseTest.setUp(self)
3658         self.queue = queue.Queue(-1)
3659         self.que_hdlr = logging.handlers.QueueHandler(self.queue)
3660         self.name = 'que'
3661         self.que_logger = logging.getLogger('que')
3662         self.que_logger.propagate = False
3663         self.que_logger.setLevel(logging.WARNING)
3664         self.que_logger.addHandler(self.que_hdlr)
3666     def tearDown(self):
3667         self.que_hdlr.close()
3668         BaseTest.tearDown(self)
3670     def test_queue_handler(self):
3671         self.que_logger.debug(self.next_message())
3672         self.assertRaises(queue.Empty, self.queue.get_nowait)
3673         self.que_logger.info(self.next_message())
3674         self.assertRaises(queue.Empty, self.queue.get_nowait)
3675         msg = self.next_message()
3676         self.que_logger.warning(msg)
3677         data = self.queue.get_nowait()
3678         self.assertTrue(isinstance(data, logging.LogRecord))
3679         self.assertEqual(data.name, self.que_logger.name)
3680         self.assertEqual((data.msg, data.args), (msg, None))
3682     def test_formatting(self):
3683         msg = self.next_message()
3686         formatted_msg = log_format_str.format(name=self.name,
3688         formatter = logging.Formatter(self.log_format)
3689         self.que_hdlr.setFormatter(formatter)
3690         self.que_logger.warning(msg)
3691         log_record = self.queue.get_nowait()
3692         self.assertEqual(formatted_msg, log_record.msg)
3693         self.assertEqual(formatted_msg, log_record.message)
3697     def test_queue_listener(self):
3699         listener = logging.handlers.QueueListener(self.queue, handler)
3702             self.que_logger.warning(self.next_message())
3703             self.que_logger.error(self.next_message())
3704             self.que_logger.critical(self.next_message())
3707         self.assertTrue(handler.matches(levelno=logging.WARNING, message='1'))
3708         self.assertTrue(handler.matches(levelno=logging.ERROR, message='2'))
3709         self.assertTrue(handler.matches(levelno=logging.CRITICAL, message='3'))
3716         listener = logging.handlers.QueueListener(self.queue, handler,
3720             self.que_logger.warning(self.next_message())
3721             self.que_logger.error(self.next_message())
3722             self.que_logger.critical(self.next_message())
3725         self.assertFalse(handler.matches(levelno=logging.WARNING, message='4'))
3726         self.assertFalse(handler.matches(levelno=logging.ERROR, message='5'))
3727         self.assertTrue(handler.matches(levelno=logging.CRITICAL, message='6'))
3732     def test_queue_listener_with_StreamHandler(self):
3734         listener = logging.handlers.QueueListener(self.queue, self.root_hdlr)
3740             self.que_logger.exception(self.next_message(), exc_info=exc)
3741         self.que_logger.error(self.next_message(), stack_info=True)
3743         self.assertEqual(self.stream.getvalue().strip().count('Traceback'), 1)
3744         self.assertEqual(self.stream.getvalue().strip().count('Stack'), 1)
3748     def test_queue_listener_with_multiple_handlers(self):
3750         self.que_hdlr.setFormatter(self.root_formatter)
3751         self.que_logger.addHandler(self.root_hdlr)
3753         listener = logging.handlers.QueueListener(self.queue, self.que_hdlr)
3755         self.que_logger.error("error")
3757         self.assertEqual(self.stream.getvalue().strip(), "que -> ERROR: error")
3797         def test_handle_called_with_queue_queue(self, mock_handle):
3798             for i in range(self.repeat):
3800                 self.setup_and_log(log_queue, '%s_%s' % (self.id(), i))
3801             self.assertEqual(mock_handle.call_count, 5 * self.repeat,
3805         def test_handle_called_with_mp_queue(self, mock_handle):
3809             for i in range(self.repeat):
3811                 self.setup_and_log(log_queue, '%s_%s' % (self.id(), i))
3814             self.assertEqual(mock_handle.call_count, 5 * self.repeat,
3825         def test_no_messages_in_queue_after_stop(self):
3835             for i in range(self.repeat):
3837                 self.setup_and_log(queue, '%s_%s' %(self.id(), i))
3839                 items = list(self.get_all_from_queue(queue))
3844                 self.assertIn(items, expected,
3849         def test_calls_task_done_after_stop(self):
3855             with self.assertRaises(ValueError):
3863     def utcoffset(self, dt):
3868     def tzname(self, dt):
3875     def assert_error_message(self, exception, message, *args, **kwargs):
3877             self.assertRaises((), *args, **kwargs)
3879             self.assertEqual(message, str(e))
3882     def setUp(self):
3883         self.common = {
3893         self.variants = {
3899     def get_record(self, name=None):
3900         result = dict(self.common)
3902             result.update(self.variants[name])
3905     def test_percent(self):
3907         r = self.get_record()
3909         self.assertEqual(f.format(r), '${Message with 2 placeholders}')
3911         self.assertRaises(ValueError, f.format, r)
3912         self.assertFalse(f.usesTime())
3914         self.assertTrue(f.usesTime())
3916         self.assertTrue(f.usesTime())
3918         self.assertTrue(f.usesTime())
3920     def test_braces(self):
3922         r = self.get_record()
3924         self.assertEqual(f.format(r), '$%Message with 2 placeholders%$')
3926         self.assertRaises(ValueError, f.format, r)
3928         self.assertFalse(f.usesTime())
3930         self.assertTrue(f.usesTime())
3932         self.assertTrue(f.usesTime())
3934         self.assertTrue(f.usesTime())
3936     def test_dollars(self):
3938         r = self.get_record()
3940         self.assertEqual(f.format(r), 'Message with 2 placeholders')
3942         self.assertEqual(f.format(r), 'Message with 2 placeholders')
3944         self.assertEqual(f.format(r), '$%Message with 2 placeholders%$')
3946         self.assertRaises(ValueError, f.format, r)
3947         self.assertFalse(f.usesTime())
3949         self.assertTrue(f.usesTime())
3951         self.assertTrue(f.usesTime())
3953         self.assertFalse(f.usesTime())
3955         self.assertTrue(f.usesTime())
3957     def test_format_validate(self):
3961         self.assertEqual(f._fmt, "%(levelname)-15s - %(message) 5s - %(process)03d - %(module) - %(asctime)*.3s")
3963         self.assertEqual(f._fmt, "%(asctime)*s - %(asctime)*.3s - %(process)-34.33o")
3965         self.assertEqual(f._fmt, "%(process)#+027.23X")
3967         self.assertEqual(f._fmt, "%(foo)#.*g")
3971         self.assertEqual(f._fmt, "$%{message}%$ - {asctime!a:15} - {customfield['key']}")
3973         self.assertEqual(f._fmt, "{process:.2f} - {custom.f:.4f}")
3975         self.assertEqual(f._fmt, "{customfield!s:#<30}")
3977         self.assertEqual(f._fmt, "{message!r}")
3979         self.assertEqual(f._fmt, "{message!s}")
3981         self.assertEqual(f._fmt, "{message!a}")
3983         self.assertEqual(f._fmt, "{process!r:4.2}")
3985         self.assertEqual(f._fmt, "{process!s:<#30,.12f}- {custom:=+#30,.1d} - {module:^30}")
3987         self.assertEqual(f._fmt, "{process!s:{w},.{p}}")
3989         self.assertEqual(f._fmt, "{foo:12.{p}}")
3991         self.assertEqual(f._fmt, "{foo:{w}.6}")
3993         self.assertEqual(f._fmt, "{foo[0].bar[1].baz}")
3995         self.assertEqual(f._fmt, "{foo[k1].bar[k2].baz}")
3997         self.assertEqual(f._fmt, "{12[k1].bar[k2].baz}")
4001         self.assertEqual(f._fmt, "${asctime} - $message")
4003         self.assertEqual(f._fmt, "$bar $$")
4005         self.assertEqual(f._fmt, "$bar $$$$")  # this would print two $($$)
4009         self.assertRaises(ValueError, logging.Formatter, "%(asctime)Z")
4010         self.assertRaises(ValueError, logging.Formatter, "%(asctime)b")
4011         self.assertRaises(ValueError, logging.Formatter, "%(asctime)*")
4012         self.assertRaises(ValueError, logging.Formatter, "%(asctime)*3s")
4013         self.assertRaises(ValueError, logging.Formatter, "%(asctime)_")
4014         self.assertRaises(ValueError, logging.Formatter, '{asctime}')
4015         self.assertRaises(ValueError, logging.Formatter, '${message}')
4016         self.assertRaises(ValueError, logging.Formatter, '%(foo)#12.3*f')  # with both * and decimal number as precision
4017         self.assertRaises(ValueError, logging.Formatter, '%(foo)0*.8*f')
4021         self.assert_error_message(
4027         self.assert_error_message(
4033         self.assert_error_message(
4037         self.assertRaises(ValueError, logging.Formatter, '{asctime!s:#30,15f}', style='{')
4038         self.assert_error_message(
4044         self.assert_error_message(
4049         self.assertRaises(ValueError, logging.Formatter, '{process:.2Z}', style='{')
4050         self.assertRaises(ValueError, logging.Formatter, '{process!s:<##30,12g}', style='{')
4051         self.assertRaises(ValueError, logging.Formatter, '{process!s:<#30#,12g}', style='{')
4052         self.assertRaises(ValueError, logging.Formatter, '{process!s:{{w}},{{p}}}', style='{')
4054         self.assert_error_message(
4059         self.assert_error_message(
4064         self.assertRaises(ValueError, logging.Formatter, '{{foo!r:4.2}', style='{')
4065         self.assertRaises(ValueError, logging.Formatter, '{{foo!r:4.2}}', style='{')
4066         self.assertRaises(ValueError, logging.Formatter, '{foo/bar}', style='{')
4067         self.assertRaises(ValueError, logging.Formatter, '{foo:{{w}}.{{p}}}}', style='{')
4068         self.assertRaises(ValueError, logging.Formatter, '{foo!X:{{w}}.{{p}}}', style='{')
4069         self.assertRaises(ValueError, logging.Formatter, '{foo!a:random}', style='{')
4070         self.assertRaises(ValueError, logging.Formatter, '{foo!a:ran{dom}', style='{')
4071         self.assertRaises(ValueError, logging.Formatter, '{foo!a:ran{d}om}', style='{')
4072         self.assertRaises(ValueError, logging.Formatter, '{foo.!a:d}', style='{')
4076         self.assert_error_message(
4081         self.assert_error_message(
4086         self.assert_error_message(
4092         self.assert_error_message(
4097         self.assertRaises(ValueError, logging.Formatter, '%(asctime)s', style='$')
4100         self.assert_error_message(
4105         self.assertRaises(ValueError, logging.Formatter, '${asctime', style='$')
4107     def test_defaults_parameter(self):
4112             r = self.get_record()
4113             self.assertEqual(f.format(r), 'Default Message with 2 placeholders')
4114             r = self.get_record("custom")
4115             self.assertEqual(f.format(r), '1234 Message with 2 placeholders')
4119             r = self.get_record()
4120             self.assertRaises(ValueError, f.format, r)
4124             r = self.get_record("custom")
4125             self.assertEqual(f.format(r), '1234 Message with 2 placeholders')
4127     def test_invalid_style(self):
4128         self.assertRaises(ValueError, logging.Formatter, None, None, 'x')
4130     def test_time(self):
4131         r = self.get_record()
4139         self.assertEqual(f.formatTime(r), '1993-04-21 08:03:00,123')
4140         self.assertEqual(f.formatTime(r, '%Y:%d'), '1993:21')
4142         self.assertEqual(r.asctime, '1993-04-21 08:03:00,123')
4144     def test_default_msec_format_none(self):
4149         r = self.get_record()
4154         self.assertEqual(f.formatTime(r), '21/04/1993 08:03:00')
4156     def test_issue_89047(self):
4162             self.assertNotIn('.1000', s)
4166     def formatHeader(self, records):
4169     def formatFooter(self, records):
4173     def setUp(self):
4174         self.records = [
4179     def test_default(self):
4181         self.assertEqual('', f.format([]))
4182         self.assertEqual('onetwo', f.format(self.records))
4184     def test_custom(self):
4186         self.assertEqual('[(2)onetwo(2)]', f.format(self.records))
4189         self.assertEqual('[(2)<one><two>(2)]', f.format(self.records))
4192     def test_formatting(self):
4193         r = self.root_logger
4203         self.assertTrue(r.exc_text.startswith('Traceback (most recent '
4205         self.assertTrue(r.exc_text.endswith('\nRuntimeError: '
4207         self.assertTrue(r.stack_info.startswith('Stack (most recent '
4209         self.assertTrue(r.stack_info.endswith('logging.exception(\'failed\', '
4214     def test_last_resort(self):
4216         root = self.root_logger
4217         root.removeHandler(self.root_hdlr)
4224                 self.assertEqual(stderr.getvalue(), '')
4226                 self.assertEqual(stderr.getvalue(), 'Final chance!\n')
4233                 self.assertEqual(stderr.getvalue(), msg)
4238                 self.assertEqual(stderr.getvalue(), '')
4245                 self.assertEqual(stderr.getvalue(), '')
4247             root.addHandler(self.root_hdlr)
4254     def __init__(self, identifier, called):
4256             setattr(self, method, self.record_call(identifier, method, called))
4258     def record_call(self, identifier, method_name, called):
4266     def __init__(self, *args, **kwargs):
4267         super(RecordingHandler, self).__init__(*args, **kwargs)
4268         self.records = []
4270     def handle(self, record):
4272         self.records.append(record)
4279     def setUp(self):
4280         super(ShutdownTest, self).setUp()
4281         self.called = []
4284         self.addCleanup(setattr, logging, 'raiseExceptions', raise_exceptions)
4286     def raise_error(self, error):
4291     def test_no_failure(self):
4293         handler0 = FakeHandler(0, self.called)
4294         handler1 = FakeHandler(1, self.called)
4295         handler2 = FakeHandler(2, self.called)
4305         self.assertEqual(expected, self.called)
4307     def _test_with_failure_in_method(self, method, error):
4308         handler = FakeHandler(0, self.called)
4309         setattr(handler, method, self.raise_error(error))
4314         self.assertEqual('0 - release', self.called[-1])
4316     def test_with_ioerror_in_acquire(self):
4317         self._test_with_failure_in_method('acquire', OSError)
4319     def test_with_ioerror_in_flush(self):
4320         self._test_with_failure_in_method('flush', OSError)
4322     def test_with_ioerror_in_close(self):
4323         self._test_with_failure_in_method('close', OSError)
4325     def test_with_valueerror_in_acquire(self):
4326         self._test_with_failure_in_method('acquire', ValueError)
4328     def test_with_valueerror_in_flush(self):
4329         self._test_with_failure_in_method('flush', ValueError)
4331     def test_with_valueerror_in_close(self):
4332         self._test_with_failure_in_method('close', ValueError)
4334     def test_with_other_error_in_acquire_without_raise(self):
4336         self._test_with_failure_in_method('acquire', IndexError)
4338     def test_with_other_error_in_flush_without_raise(self):
4340         self._test_with_failure_in_method('flush', IndexError)
4342     def test_with_other_error_in_close_without_raise(self):
4344         self._test_with_failure_in_method('close', IndexError)
4346     def test_with_other_error_in_acquire_with_raise(self):
4348         self.assertRaises(IndexError, self._test_with_failure_in_method,
4351     def test_with_other_error_in_flush_with_raise(self):
4353         self.assertRaises(IndexError, self._test_with_failure_in_method,
4356     def test_with_other_error_in_close_with_raise(self):
4358         self.assertRaises(IndexError, self._test_with_failure_in_method,
4366     def test_disable(self):
4369         self.assertEqual(old_disable, 0)
4370         self.addCleanup(logging.disable, old_disable)
4373         self.assertEqual(logging.root.manager.disable, 83)
4375         self.assertRaises(ValueError, logging.disable, "doesnotexists")
4380         self.assertRaises(TypeError, logging.disable, _NotAnIntOrString())
4387         self.assertEqual(logging.root.manager.disable, logging.CRITICAL)
4389     def _test_log(self, method, level=None):
4391         support.patch(self, logging, 'basicConfig',
4403         self.assertEqual(len(recording.records), 1)
4405         self.assertEqual(record.getMessage(), "test me: %r" % recording)
4408         self.assertEqual(record.levelno, expected_level)
4411         self.assertEqual(called, [])
4413     def test_log(self):
4414         self._test_log('log', logging.ERROR)
4416     def test_debug(self):
4417         self._test_log('debug')
4419     def test_info(self):
4420         self._test_log('info')
4422     def test_warning(self):
4423         self._test_log('warning')
4425     def test_error(self):
4426         self._test_log('error')
4428     def test_critical(self):
4429         self._test_log('critical')
4431     def test_set_logger_class(self):
4432         self.assertRaises(TypeError, logging.setLoggerClass, object)
4438         self.assertEqual(logging.getLoggerClass(), MyLogger)
4441         self.assertEqual(logging.getLoggerClass(), logging.Logger)
4443     def test_subclass_logger_cache(self):
4448             def __init__(self, name='MyLogger', level=logging.NOTSET):
4454         self.assertEqual(message, ['initialized'])
4461             self.assertEqual(stream.getvalue().strip(), "hello")
4468             self.assertEqual(stream.getvalue(), "")
4474     def test_logging_at_shutdown(self):
4480                 def __del__(self):
4490         self.assertIn("exception in __del__", err)
4491         self.assertIn("ValueError: some error", err)
4493     def test_logging_at_shutdown_open(self):
4498         self.addCleanup(os_helper.unlink, filename)
4505                 def __del__(self):
4522             self.assertEqual(fp.read().rstrip(), "ERROR:root:log in __del__")
4524     def test_recursion_error(self):
4537         self.assertNotIn("Cannot recover from stack overflow.", err)
4538         self.assertEqual(rc, 1)
4540     def test_get_level_names_mapping(self):
4542         self.assertEqual(logging._nameToLevel, mapping)  # value is equivalent
4543         self.assertIsNot(logging._nameToLevel, mapping)  # but not the internal data
4545         self.assertIsNot(mapping, new_mapping)           # verify not the same object as before
4546         self.assertEqual(mapping, new_mapping)           # but equivalent in value
4550     def test_str_rep(self):
4553         self.assertTrue(s.startswith('<LogRecord: '))
4554         self.assertTrue(s.endswith('>'))
4556     def test_dict_arg(self):
4562         self.assertIs(h.records[0].args, d)
4563         self.assertEqual(h.records[0].message, 'less is more')
4592     def test_multiprocessing(self):
4597             self.assertEqual(logging.logMultiprocessing, True)
4603             self.assertEqual(r.processName, 'MainProcess')
4605             results = self._extract_logrecord_process_name(1, LOG_MULTI_PROCESSING)
4606             self.assertEqual('MainProcess', results['processName'])
4607             self.assertEqual('MainProcess', results['r1.processName'])
4608             self.assertEqual('MainProcess', results['r2.processName'])
4615                 target=self._extract_logrecord_process_name,
4620             self.assertNotEqual('MainProcess', results['processName'])
4621             self.assertEqual(results['processName'], results['r1.processName'])
4622             self.assertEqual('MainProcess', results['r2.processName'])
4629     def test_optional(self):
4631         NOT_NONE = self.assertIsNotNone
4644             NONE = self.assertIsNone
4658     def setUp(self):
4659         super(BasicConfigTest, self).setUp()
4660         self.handlers = logging.root.handlers
4661         self.saved_handlers = logging._handlers.copy()
4662         self.saved_handler_list = logging._handlerList[:]
4663         self.original_logging_level = logging.root.level
4664         self.addCleanup(self.cleanup)
4667     def tearDown(self):
4671         super(BasicConfigTest, self).tearDown()
4673     def cleanup(self):
4674         setattr(logging.root, 'handlers', self.handlers)
4676         logging._handlers.update(self.saved_handlers)
4677         logging._handlerList[:] = self.saved_handler_list
4678         logging.root.setLevel(self.original_logging_level)
4680     def test_no_kwargs(self):
4684         self.assertEqual(len(logging.root.handlers), 1)
4686         self.assertIsInstance(handler, logging.StreamHandler)
4687         self.assertEqual(handler.stream, sys.stderr)
4691         self.assertEqual(formatter._style._fmt, logging.BASIC_FORMAT)
4693         self.assertIsNone(formatter.datefmt)
4695         self.assertIsInstance(formatter._style, logging.PercentStyle)
4698         self.assertEqual(logging.root.level, self.original_logging_level)
4700     def test_strformatstyle(self):
4705             self.assertEqual(output.getvalue().strip(),
4708     def test_stringtemplatestyle(self):
4713             self.assertEqual(output.getvalue().strip(),
4716     def test_filename(self):
4725         self.assertEqual(len(logging.root.handlers), 1)
4727         self.assertIsInstance(handler, logging.FileHandler)
4730         self.assertEqual(handler.stream.mode, expected.stream.mode)
4731         self.assertEqual(handler.stream.name, expected.stream.name)
4732         self.addCleanup(cleanup, handler, expected, 'test.log')
4734     def test_filemode(self):
4745         self.assertEqual(handler.stream.mode, expected.stream.mode)
4746         self.addCleanup(cleanup, handler, expected, 'test.log')
4748     def test_stream(self):
4750         self.addCleanup(stream.close)
4753         self.assertEqual(len(logging.root.handlers), 1)
4755         self.assertIsInstance(handler, logging.StreamHandler)
4756         self.assertEqual(handler.stream, stream)
4758     def test_format(self):
4762         self.assertEqual(formatter._style._fmt, '%(asctime)s - %(message)s')
4764     def test_datefmt(self):
4768         self.assertEqual(formatter.datefmt, 'bar')
4770     def test_style(self):
4774         self.assertIsInstance(formatter._style, logging.StringTemplateStyle)
4776     def test_level(self):
4778         self.addCleanup(logging.root.setLevel, old_level)
4781         self.assertEqual(logging.root.level, 57)
4784         self.assertEqual(logging.root.level, 57)
4786     def test_incompatible(self):
4787         assertRaises = self.assertRaises
4801     def test_handlers(self):
4810         self.assertIs(handlers[0], logging.root.handlers[0])
4811         self.assertIs(handlers[1], logging.root.handlers[1])
4812         self.assertIs(handlers[2], logging.root.handlers[2])
4813         self.assertIsNotNone(handlers[0].formatter)
4814         self.assertIsNotNone(handlers[1].formatter)
4815         self.assertIs(handlers[2].formatter, f)
4816         self.assertIs(handlers[0].formatter, handlers[1].formatter)
4818     def test_force(self):
4827         self.assertEqual(len(logging.root.handlers), 1)
4833         self.assertEqual(len(logging.root.handlers), 1)
4834         self.assertEqual(old_string_io.getvalue().strip(),
4836         self.assertEqual(new_string_io.getvalue().strip(),
4839     def test_encoding(self):
4846             self.assertEqual(len(logging.root.handlers), 1)
4848             self.assertIsInstance(handler, logging.FileHandler)
4849             self.assertEqual(handler.encoding, encoding)
4856             self.assertEqual(data,
4859     def test_encoding_errors(self):
4866             self.assertEqual(len(logging.root.handlers), 1)
4868             self.assertIsInstance(handler, logging.FileHandler)
4869             self.assertEqual(handler.encoding, encoding)
4876             self.assertEqual(data, 'The resund Bridge joins Copenhagen to Malm')
4878     def test_encoding_errors_default(self):
4884             self.assertEqual(len(logging.root.handlers), 1)
4886             self.assertIsInstance(handler, logging.FileHandler)
4887             self.assertEqual(handler.encoding, encoding)
4888             self.assertEqual(handler.errors, 'backslashreplace')
4895             self.assertEqual(data, r'\U0001f602: \u2603\ufe0f: The \xd8resund '
4898     def test_encoding_errors_none(self):
4906             self.assertEqual(len(logging.root.handlers), 1)
4908             self.assertIsInstance(handler, logging.FileHandler)
4909             self.assertEqual(handler.encoding, encoding)
4910             self.assertIsNone(handler.errors)
4920             self.assertTrue(message)
4921             self.assertIn("'ascii' codec can't encode "
4929             self.assertEqual(data, r'')
4932     def _test_log(self, method, level=None):
4941             self.addCleanup(logging.root.setLevel, old_level)
4944         support.patch(self, logging, 'basicConfig', my_basic_config)
4953         self.assertEqual(called, [((), {})])
4955     def test_log(self):
4956         self._test_log('log', logging.WARNING)
4958     def test_debug(self):
4959         self._test_log('debug')
4961     def test_info(self):
4962         self._test_log('info')
4964     def test_warning(self):
4965         self._test_log('warning')
4967     def test_error(self):
4968         self._test_log('error')
4970     def test_critical(self):
4971         self._test_log('critical')
4975     def setUp(self):
4976         super(LoggerAdapterTest, self).setUp()
4979         self.recording = RecordingHandler()
4980         self.logger = logging.root
4981         self.logger.addHandler(self.recording)
4982         self.addCleanup(self.logger.removeHandler, self.recording)
4983         self.addCleanup(self.recording.close)
4988         self.addCleanup(cleanup)
4989         self.addCleanup(logging.shutdown)
4990         self.adapter = logging.LoggerAdapter(logger=self.logger, extra=None)
4992     def test_exception(self):
4999             self.adapter.exception(msg, self.recording)
5001         self.assertEqual(len(self.recording.records), 1)
5002         record = self.recording.records[0]
5003         self.assertEqual(record.levelno, logging.ERROR)
5004         self.assertEqual(record.msg, msg)
5005         self.assertEqual(record.args, (self.recording,))
5006         self.assertEqual(record.exc_info,
5009     def test_exception_excinfo(self):
5015         self.adapter.exception('exc_info test', exc_info=exc)
5017         self.assertEqual(len(self.recording.records), 1)
5018         record = self.recording.records[0]
5019         self.assertEqual(record.exc_info,
5022     def test_critical(self):
5024         self.adapter.critical(msg, self.recording)
5026         self.assertEqual(len(self.recording.records), 1)
5027         record = self.recording.records[0]
5028         self.assertEqual(record.levelno, logging.CRITICAL)
5029         self.assertEqual(record.msg, msg)
5030         self.assertEqual(record.args, (self.recording,))
5032     def test_is_enabled_for(self):
5033         old_disable = self.adapter.logger.manager.disable
5034         self.adapter.logger.manager.disable = 33
5035         self.addCleanup(setattr, self.adapter.logger.manager, 'disable',
5037         self.assertFalse(self.adapter.isEnabledFor(32))
5039     def test_has_handlers(self):
5040         self.assertTrue(self.adapter.hasHandlers())
5042         for handler in self.logger.handlers:
5043             self.logger.removeHandler(handler)
5045         self.assertFalse(self.logger.hasHandlers())
5046         self.assertFalse(self.adapter.hasHandlers())
5048     def test_nested(self):
5052             def process(self, msg, kwargs):
5053                 return f"{self.prefix} {msg}", kwargs
5056         adapter = Adapter(logger=self.logger, extra=None)
5059         self.assertEqual(repr(adapter), repr(adapter_adapter))
5060         adapter_adapter.log(logging.CRITICAL, msg, self.recording)
5061         self.assertEqual(len(self.recording.records), 1)
5062         record = self.recording.records[0]
5063         self.assertEqual(record.levelno, logging.CRITICAL)
5064         self.assertEqual(record.msg, f"Adapter AdapterAdapter {msg}")
5065         self.assertEqual(record.args, (self.recording,))
5067         self.assertIs(adapter.manager, orig_manager)
5068         self.assertIs(self.logger.manager, orig_manager)
5072             self.assertIs(adapter_adapter.manager, temp_manager)
5073             self.assertIs(adapter.manager, temp_manager)
5074             self.assertIs(self.logger.manager, temp_manager)
5077         self.assertIs(adapter_adapter.manager, orig_manager)
5078         self.assertIs(adapter.manager, orig_manager)
5079         self.assertIs(self.logger.manager, orig_manager)
5084     def setUp(self):
5085         super(LoggerTest, self).setUp()
5086         self.recording = RecordingHandler()
5087         self.logger = logging.Logger(name='blah')
5088         self.logger.addHandler(self.recording)
5089         self.addCleanup(self.logger.removeHandler, self.recording)
5090         self.addCleanup(self.recording.close)
5091         self.addCleanup(logging.shutdown)
5093     def test_set_invalid_level(self):
5094         self.assert_error_message(
5096             self.logger.setLevel, None)
5097         self.assert_error_message(
5099             self.logger.setLevel, (0, 0))
5101     def test_exception(self):
5108             self.logger.exception(msg, self.recording)
5110         self.assertEqual(len(self.recording.records), 1)
5111         record = self.recording.records[0]
5112         self.assertEqual(record.levelno, logging.ERROR)
5113         self.assertEqual(record.msg, msg)
5114         self.assertEqual(record.args, (self.recording,))
5115         self.assertEqual(record.exc_info,
5118     def test_log_invalid_level_with_raise(self):
5120             self.assertRaises(TypeError, self.logger.log, '10', 'test message')
5122     def test_log_invalid_level_no_raise(self):
5124             self.logger.log('10', 'test message')  # no exception happens
5126     def test_find_caller_with_stack_info(self):
5128         support.patch(self, logging.traceback, 'print_stack',
5131         self.logger.findCaller(stack_info=True)
5133         self.assertEqual(len(called), 1)
5134         self.assertEqual('Stack (most recent call last):\n', called[0])
5136     def test_find_caller_with_stacklevel(self):
5138         trigger = self.logger.warning
5149         records = self.recording.records
5151         self.assertEqual(records[-1].funcName, 'innermost')
5155         self.assertEqual(records[-1].funcName, 'inner')
5156         self.assertGreater(records[-1].lineno, lineno)
5160         self.assertEqual(records[-1].funcName, 'outer')
5161         self.assertGreater(records[-1].lineno, lineno)
5164         root_logger.addHandler(self.recording)
5167         self.assertEqual(records[-1].funcName, 'outer')
5168         root_logger.removeHandler(self.recording)
5169         trigger = self.logger.warning
5172         self.assertEqual(records[-1].funcName, 'test_find_caller_with_stacklevel')
5173         self.assertGreater(records[-1].lineno, lineno)
5175     def test_make_record_with_extra_overwrite(self):
5184             self.assertRaises(KeyError, self.logger.makeRecord, name, level,
5188     def test_make_record_with_extra_no_overwrite(self):
5193         result = self.logger.makeRecord(name, level, fn, lno, msg, args,
5195         self.assertIn('valid_key', result.__dict__)
5197     def test_has_handlers(self):
5198         self.assertTrue(self.logger.hasHandlers())
5200         for handler in self.logger.handlers:
5201             self.logger.removeHandler(handler)
5202         self.assertFalse(self.logger.hasHandlers())
5204     def test_has_handlers_no_propagate(self):
5207         self.assertFalse(child_logger.hasHandlers())
5209     def test_is_enabled_for(self):
5210         old_disable = self.logger.manager.disable
5211         self.logger.manager.disable = 23
5212         self.addCleanup(setattr, self.logger.manager, 'disable', old_disable)
5213         self.assertFalse(self.logger.isEnabledFor(22))
5215     def test_is_enabled_for_disabled_logger(self):
5216         old_disabled = self.logger.disabled
5217         old_disable = self.logger.manager.disable
5219         self.logger.disabled = True
5220         self.logger.manager.disable = 21
5222         self.addCleanup(setattr, self.logger, 'disabled', old_disabled)
5223         self.addCleanup(setattr, self.logger.manager, 'disable', old_disable)
5225         self.assertFalse(self.logger.isEnabledFor(22))
5227     def test_root_logger_aliases(self):
5229         self.assertIs(root, logging.root)
5230         self.assertIs(root, logging.getLogger(None))
5231         self.assertIs(root, logging.getLogger(''))
5232         self.assertIs(root, logging.getLogger('root'))
5233         self.assertIs(root, logging.getLogger('foo').root)
5234         self.assertIs(root, logging.getLogger('foo.bar').root)
5235         self.assertIs(root, logging.getLogger('foo').parent)
5237         self.assertIsNot(root, logging.getLogger('\0'))
5238         self.assertIsNot(root, logging.getLogger('foo.bar').parent)
5240     def test_invalid_names(self):
5241         self.assertRaises(TypeError, logging.getLogger, any)
5242         self.assertRaises(TypeError, logging.getLogger, b'foo')
5244     def test_pickling(self):
5250                 self.assertIs(unpickled, logger)
5252     def test_caching(self):
5253         root = self.root_logger
5259         self.assertEqual(logger2.getEffectiveLevel(), logging.ERROR)
5260         self.assertEqual(logger2._cache, {})
5263         self.assertTrue(logger2.isEnabledFor(logging.ERROR))
5264         self.assertFalse(logger2.isEnabledFor(logging.DEBUG))
5265         self.assertEqual(logger2._cache, {logging.ERROR: True, logging.DEBUG: False})
5266         self.assertEqual(root._cache, {})
5267         self.assertTrue(logger2.isEnabledFor(logging.ERROR))
5270         self.assertEqual(root._cache, {})
5271         self.assertTrue(root.isEnabledFor(logging.ERROR))
5272         self.assertEqual(root._cache, {logging.ERROR: True})
5276         self.assertEqual(logger2.getEffectiveLevel(), logging.CRITICAL)
5277         self.assertEqual(logger2._cache, {})
5280         self.assertFalse(logger2.isEnabledFor(logging.ERROR))
5284         self.assertEqual(logger2.getEffectiveLevel(), logging.CRITICAL)
5285         self.assertEqual(logger2._cache, {})
5286         self.assertEqual(logger1._cache, {})
5287         self.assertEqual(root._cache, {})
5290         self.assertFalse(logger2.isEnabledFor(logging.ERROR))
5291         self.assertTrue(logger2.isEnabledFor(logging.CRITICAL))
5292         self.assertFalse(logger1.isEnabledFor(logging.ERROR))
5293         self.assertTrue(logger1.isEnabledFor(logging.CRITICAL))
5294         self.assertTrue(root.isEnabledFor(logging.ERROR))
5298         self.assertEqual(logger2.getEffectiveLevel(), logging.CRITICAL)
5299         self.assertEqual(logger2._cache, {})
5300         self.assertEqual(logger1._cache, {})
5301         self.assertEqual(root._cache, {})
5304         self.assertFalse(logger1.isEnabledFor(logging.CRITICAL))
5305         self.assertFalse(logger2.isEnabledFor(logging.CRITICAL))
5306         self.assertFalse(root.isEnabledFor(logging.CRITICAL))
5312     def setUp(self):
5313         BaseTest.setUp(self)
5314         fd, self.fn = tempfile.mkstemp(".log", "test_logging-2-")
5316         self.rmfiles = []
5318     def tearDown(self):
5319         for fn in self.rmfiles:
5321         if os.path.exists(self.fn):
5322             os.unlink(self.fn)
5323         BaseTest.tearDown(self)
5325     def assertLogFile(self, filename):
5327         self.assertTrue(os.path.exists(filename),
5329         self.rmfiles.append(filename)
5331     def next_rec(self):
5333                                  self.next_message(), None, None, None)
5336     def test_delay(self):
5337         os.unlink(self.fn)
5338         fh = logging.FileHandler(self.fn, encoding='utf-8', delay=True)
5339         self.assertIsNone(fh.stream)
5340         self.assertFalse(os.path.exists(self.fn))
5342         self.assertIsNotNone(fh.stream)
5343         self.assertTrue(os.path.exists(self.fn))
5346     def test_emit_after_closing_in_write_mode(self):
5348         os.unlink(self.fn)
5349         fh = logging.FileHandler(self.fn, encoding='utf-8', mode='w')
5351         fh.emit(self.next_rec())    # '1'
5353         fh.emit(self.next_rec())    # '2'
5354         with open(self.fn) as fp:
5355             self.assertEqual(fp.read().strip(), '1')
5359     def test_should_not_rollover(self):
5362                 self.fn, encoding="utf-8", maxBytes=0)
5363         self.assertFalse(rh.shouldRollover(None))
5370         self.assertFalse(rh.shouldRollover(self.next_rec()))
5373     def test_should_rollover(self):
5374         rh = logging.handlers.RotatingFileHandler(self.fn, encoding="utf-8", maxBytes=1)
5375         self.assertTrue(rh.shouldRollover(self.next_rec()))
5378     def test_file_created(self):
5381         rh = logging.handlers.RotatingFileHandler(self.fn, encoding="utf-8")
5382         rh.emit(self.next_rec())
5383         self.assertLogFile(self.fn)
5386     def test_rollover_filenames(self):
5390             self.fn, encoding="utf-8", backupCount=2, maxBytes=1)
5392         rh.emit(self.next_rec())
5393         self.assertLogFile(self.fn)
5394         rh.emit(self.next_rec())
5395         self.assertLogFile(namer(self.fn + ".1"))
5396         rh.emit(self.next_rec())
5397         self.assertLogFile(namer(self.fn + ".2"))
5398         self.assertFalse(os.path.exists(namer(self.fn + ".3")))
5401     def test_namer_rotator_inheritance(self):
5403             def namer(self, name):
5406             def rotator(self, source, dest):
5411             self.fn, encoding="utf-8", backupCount=2, maxBytes=1)
5412         self.assertEqual(rh.namer(self.fn), self.fn + ".test")
5413         rh.emit(self.next_rec())
5414         self.assertLogFile(self.fn)
5415         rh.emit(self.next_rec())
5416         self.assertLogFile(rh.namer(self.fn + ".1") + ".rotated")
5417         self.assertFalse(os.path.exists(rh.namer(self.fn + ".1")))
5421     def test_rotator(self):
5434             self.fn, encoding="utf-8", backupCount=2, maxBytes=1)
5437         m1 = self.next_rec()
5439         self.assertLogFile(self.fn)
5440         m2 = self.next_rec()
5442         fn = namer(self.fn + ".1")
5443         self.assertLogFile(fn)
5448             self.assertEqual(data.decode("ascii"), m1.msg + newline)
5449         rh.emit(self.next_rec())
5450         fn = namer(self.fn + ".2")
5451         self.assertLogFile(fn)
5455             self.assertEqual(data.decode("ascii"), m1.msg + newline)
5456         rh.emit(self.next_rec())
5457         fn = namer(self.fn + ".2")
5461             self.assertEqual(data.decode("ascii"), m2.msg + newline)
5462         self.assertFalse(os.path.exists(namer(self.fn + ".3")))
5467     def test_should_not_rollover(self):
5473         self.assertFalse(fh.shouldRollover(r))
5477     def test_rollover(self):
5479                 self.fn, 'S', encoding="utf-8", backupCount=1)
5484         self.assertLogFile(self.fn)
5500             fn = self.fn + prev.strftime(".%Y-%m-%d_%H-%M-%S")
5503                 self.rmfiles.append(fn)
5508             dn, fn = os.path.split(self.fn)
5517         self.assertTrue(found, msg=msg)
5519     def test_invalid(self):
5520         assertRaises = self.assertRaises
5522                      self.fn, 'X', encoding="utf-8", delay=True)
5524                      self.fn, 'W', encoding="utf-8", delay=True)
5526                      self.fn, 'W7', encoding="utf-8", delay=True)
5528     def test_compute_rollover_daily_attime(self):
5532             self.fn, encoding="utf-8", when='MIDNIGHT', interval=1, backupCount=0,
5536             self.assertEqual(actual, currentTime + 12 * 60 * 60)
5539             self.assertEqual(actual, currentTime + 36 * 60 * 60)
5544     def test_compute_rollover_weekly_attime(self):
5553                 self.fn, encoding="utf-8", when='W%d' % day, interval=1, backupCount=0,
5572                 self.assertEqual(actual, expected)
5580                 self.assertEqual(actual, expected)
5584     def test_compute_files_to_delete(self):
5587         self.addCleanup(shutil.rmtree, wd)
5619             self.assertEqual(len(candidates), 3)
5624                     self.assertTrue(fn.startswith(p))
5628                     self.assertTrue(fn.endswith('.log'))
5629                     self.assertTrue(fn.startswith(prefix + '.') and
5644     def test_compute_rollover(self, when=when, exp=exp):
5646             self.fn, encoding="utf-8", when=when, interval=1, backupCount=0, utc=True)
5675         self.assertEqual(exp, actual)
5682     def test_basic(self):
5698         self.assertLess(num_recs, win32evtlog.GetNumberOfEventLogRecords(elh))
5713         self.assertTrue(found, msg=msg)
5717     def test__all__(self):
5723         support.check__all__(self, logging, not_exported=not_exported)

Indexes created Thu Nov 07 10:32:03 CST 2024