Using appropriate checks in tests

Lib/test/test_logging.py

-#!/usr/bin/env python
-#
 # Copyright 2001-2013 by Vinay Sajip. All Rights Reserved.
 #
 # Permission to use, copy, modify, and distribute this software and its
 # documentation for any purpose and without fee is hereby granted,
 # provided that the above copyright notice appear in all copies and that
 # both that copyright notice and this permission notice appear in
 # supporting documentation, and that the name of Vinay Sajip
 # not be used in advertising or publicity pertaining to distribution
 # of the software without specific, written prior permission.
 # VINAY SAJIP DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING
@@ skipping 21 matching lines @@
 import json
 import os
 import queue
 import random
 import re
 import select
 import socket
 import struct
 import sys
 import tempfile
+from test.script_helper import assert_python_ok
 from test.support import (captured_stdout, run_with_locale, run_unittest,
                           patch, requires_zlib, TestHandler, Matcher)
 import textwrap
 import time
 import unittest
 import warnings
 import weakref
 try:
     import threading
     # The following imports are needed only for tests which
@@ skipping 529 matching lines @@
         h.close()
 
     @unittest.skipIf(os.name == 'nt', 'WatchedFileHandler not appropriate for Windows.')
     @unittest.skipUnless(threading, 'Threading required for this test.')
     def test_race(self):
         # Issue #14632 refers.
         def remove_loop(fname, tries):
             for _ in range(tries):
                 try:
                     os.unlink(fname)
+                    self.deletion_time = time.time()
                 except OSError:
                     pass
                 time.sleep(0.004 * random.randint(0, 4))
 
         del_count = 500
         log_count = 500
+
+        self.handle_time = None
+        self.deletion_time = None
 
         for delay in (False, True):
             fd, fn = tempfile.mkstemp('.log', 'test_logging-3-')
             os.close(fd)
             remover = threading.Thread(target=remove_loop, args=(fn, del_count))
             remover.daemon = True
             remover.start()
             h = logging.handlers.WatchedFileHandler(fn, delay=delay)
             f = logging.Formatter('%(asctime)s: %(levelname)s: %(message)s')
             h.setFormatter(f)
             try:
                 for _ in range(log_count):
                     time.sleep(0.005)
                     r = logging.makeLogRecord({'msg': 'testing' })
-                    h.handle(r)
+                    try:
+                        self.handle_time = time.time()
+                        h.handle(r)
+                    except Exception:
+                        print('Deleted at %s, '
+                              'opened at %s' % (self.deletion_time,
+                                                self.handle_time))
+                        raise
             finally:
                 remover.join()
                 h.close()
                 if os.path.exists(fn):
                     os.unlink(fn)
 
 
 class BadStream(object):
     def write(self, data):
         raise RuntimeError('deliberate mistake')
@@ skipping 281 matching lines @@
             super(TestUDPServer, self).server_close()
             self._closed = True
 
     class TestUnixDatagramServer(TestUDPServer):
         address_family = socket.AF_UNIX
 
 # - end of server_helper section
 
 @unittest.skipUnless(threading, 'Threading required for this test.')
 class SMTPHandlerTest(BaseTest):
+    TIMEOUT = 8.0
     def test_basic(self):
         sockmap = {}
         server = TestSMTPServer(('localhost', 0), self.process_message, 0.001,
                                 sockmap)
         server.start()
         addr = ('localhost', server.port)
-        h = logging.handlers.SMTPHandler(addr, 'me', 'you', 'Log', timeout=5.0)
+        h = logging.handlers.SMTPHandler(addr, 'me', 'you', 'Log',
+                                         timeout=self.TIMEOUT)
         self.assertEqual(h.toaddrs, ['you'])
         self.messages = []
         r = logging.makeLogRecord({'msg': 'Hello'})
         self.handled = threading.Event()
         h.handle(r)
-        self.handled.wait(5.0)  # 14314: don't wait forever
+        self.handled.wait(self.TIMEOUT)  # 14314: don't wait forever
         server.stop()
         self.assertTrue(self.handled.is_set())
         self.assertEqual(len(self.messages), 1)
         peer, mailfrom, rcpttos, data = self.messages[0]
         self.assertEqual(mailfrom, 'me')
         self.assertEqual(rcpttos, ['you'])
         self.assertIn('\nSubject: Log\n', data)
         self.assertTrue(data.endswith('\n\nHello'))
         h.close()
 
@@ skipping 2439 matching lines @@
 
         class MyLogger(logging.Logger):
             pass
 
         logging.setLoggerClass(MyLogger)
         self.assertEqual(logging.getLoggerClass(), MyLogger)
 
         logging.setLoggerClass(logging.Logger)
         self.assertEqual(logging.getLoggerClass(), logging.Logger)
 
+    def test_logging_at_shutdown(self):
+        # Issue #20037
+        code = """if 1:
+            import logging
+
+            class A:
+                def __del__(self):
+                    try:
+                        raise ValueError("some error")
+                    except Exception:
+                        logging.exception("exception in __del__")
+
+            a = A()"""
+        rc, out, err = assert_python_ok("-c", code)
+        err = err.decode()
+        self.assertIn("exception in __del__", err)
+        self.assertIn("ValueError: some error", err)
+
+
 class LogRecordTest(BaseTest):
     def test_str_rep(self):
         r = logging.makeLogRecord({})
         s = str(r)
         self.assertTrue(s.startswith('<LogRecord: '))
         self.assertTrue(s.endswith('>'))
 
     def test_dict_arg(self):
         h = RecordingHandler()
         r = logging.getLogger()
@@ skipping 79 matching lines @@
         formatter = handler.formatter
         # format defaults to logging.BASIC_FORMAT
         self.assertEqual(formatter._style._fmt, logging.BASIC_FORMAT)
         # datefmt defaults to None
         self.assertIsNone(formatter.datefmt)
         # style defaults to %
         self.assertIsInstance(formatter._style, logging.PercentStyle)
 
         # level is not explicitly set
         self.assertEqual(logging.root.level, self.original_logging_level)
+
+    def test_strformatstyle(self):
+        with captured_stdout() as output:
+            logging.basicConfig(stream=sys.stdout, style="{")
+            logging.error("Log an error")
+            sys.stdout.seek(0)
+            self.assertEqual(output.getvalue().strip(),
+                "ERROR:root:Log an error")
+
+    def test_stringtemplatestyle(self):
+        with captured_stdout() as output:
+            logging.basicConfig(stream=sys.stdout, style="$")
+            logging.error("Log an error")
+            sys.stdout.seek(0)
+            self.assertEqual(output.getvalue().strip(),
+                "ERROR:root:Log an error")
 
     def test_filename(self):
 
         def cleanup(h1, h2, fn):
             h1.close()
             h2.close()
             os.remove(fn)
 
         logging.basicConfig(filename='test.log')
 
@@ skipping 621 matching lines @@
                  BasicConfigTest, LoggerAdapterTest, LoggerTest,
                  SMTPHandlerTest, FileHandlerTest, RotatingFileHandlerTest,
                  LastResortTest, LogRecordTest, ExceptionTest,
                  SysLogHandlerTest, HTTPHandlerTest, NTEventLogHandlerTest,
                  TimedRotatingFileHandlerTest, UnixSocketHandlerTest,
                  UnixDatagramHandlerTest, UnixSysLogHandlerTest
                 )
 
 if __name__ == "__main__":
     test_main()