mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-19 05:08:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

Closes #29177: Merged fix from 3.6.

Browse source
This commit is contained in:
Vinay Sajip 2017-01-09 16:55:24 +00:00
commit 9ebb245db8
1 changed files with 62 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

86
Lib/test/test_logging.py
View file

@ -1,4 +1,4 @@
# Copyright 2001-2016 by Vinay Sajip. All Rights Reserved. # Copyright 2001-2017 by Vinay Sajip. All Rights Reserved.
# #
# Permission to use, copy, modify, and distribute this software and its # Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose and without fee is hereby granted, # documentation for any purpose and without fee is hereby granted,
@ -16,7 +16,7 @@
"""Test harness for the logging module. Run all tests. """Test harness for the logging module. Run all tests.
Copyright (C) 2001-2016 Vinay Sajip. All Rights Reserved. Copyright (C) 2001-2017 Vinay Sajip. All Rights Reserved.
""" """
import logging import logging
@ -1440,9 +1440,17 @@ class SocketHandlerTest(BaseTest):
"""Set up a TCP server to receive log messages, and a SocketHandler """Set up a TCP server to receive log messages, and a SocketHandler
pointing to that server's address and port.""" pointing to that server's address and port."""
BaseTest.setUp(self) BaseTest.setUp(self)
self.server = server = self.server_class(self.address, # Issue #29177: deal with errors that happen during setup
self.handle_socket, 0.01) self.server = self.sock_hdlr = self.server_exception = None
server.start() try:
self.server = server = self.server_class(self.address,
self.handle_socket, 0.01)
server.start()
# Uncomment next line to test error recovery in setUp()
# raise OSError('dummy error raised')
except OSError as e:
self.server_exception = e
return
server.ready.wait() server.ready.wait()
hcls = logging.handlers.SocketHandler hcls = logging.handlers.SocketHandler
if isinstance(server.server_address, tuple): if isinstance(server.server_address, tuple):
@ -1457,9 +1465,11 @@ class SocketHandlerTest(BaseTest):
def tearDown(self): def tearDown(self):
"""Shutdown the TCP server.""" """Shutdown the TCP server."""
try: try:
self.server.stop(2.0) if self.server:
self.root_logger.removeHandler(self.sock_hdlr) self.server.stop(2.0)
self.sock_hdlr.close() if self.sock_hdlr:
self.root_logger.removeHandler(self.sock_hdlr)
self.sock_hdlr.close()
finally: finally:
BaseTest.tearDown(self) BaseTest.tearDown(self)
@ -1480,6 +1490,8 @@ class SocketHandlerTest(BaseTest):
def test_output(self): def test_output(self):
# The log message sent to the SocketHandler is properly received. # The log message sent to the SocketHandler is properly received.
if self.server_exception:
self.skipTest(self.server_exception)
logger = logging.getLogger("tcp") logger = logging.getLogger("tcp")
logger.error("spam") logger.error("spam")
self.handled.acquire() self.handled.acquire()
@ -1488,6 +1500,8 @@ class SocketHandlerTest(BaseTest):
self.assertEqual(self.log_output, "spam\neggs\n") self.assertEqual(self.log_output, "spam\neggs\n")
def test_noserver(self): def test_noserver(self):
if self.server_exception:
self.skipTest(self.server_exception)
# Avoid timing-related failures due to SocketHandler's own hard-wired # Avoid timing-related failures due to SocketHandler's own hard-wired
# one-second timeout on socket.create_connection() (issue #16264). # one-second timeout on socket.create_connection() (issue #16264).
self.sock_hdlr.retryStart = 2.5 self.sock_hdlr.retryStart = 2.5
@ -1528,7 +1542,7 @@ class UnixSocketHandlerTest(SocketHandlerTest):
def tearDown(self): def tearDown(self):
SocketHandlerTest.tearDown(self) SocketHandlerTest.tearDown(self)
os.remove(self.address) support.unlink(self.address)
@unittest.skipUnless(threading, 'Threading required for this test.') @unittest.skipUnless(threading, 'Threading required for this test.')
class DatagramHandlerTest(BaseTest): class DatagramHandlerTest(BaseTest):
@ -1543,9 +1557,17 @@ class DatagramHandlerTest(BaseTest):
"""Set up a UDP server to receive log messages, and a DatagramHandler """Set up a UDP server to receive log messages, and a DatagramHandler
pointing to that server's address and port.""" pointing to that server's address and port."""
BaseTest.setUp(self) BaseTest.setUp(self)
self.server = server = self.server_class(self.address, # Issue #29177: deal with errors that happen during setup
self.handle_datagram, 0.01) self.server = self.sock_hdlr = self.server_exception = None
server.start() try:
self.server = server = self.server_class(self.address,
self.handle_datagram, 0.01)
server.start()
# Uncomment next line to test error recovery in setUp()
# raise OSError('dummy error raised')
except OSError as e:
self.server_exception = e
return
server.ready.wait() server.ready.wait()
hcls = logging.handlers.DatagramHandler hcls = logging.handlers.DatagramHandler
if isinstance(server.server_address, tuple): if isinstance(server.server_address, tuple):
@ -1560,9 +1582,11 @@ class DatagramHandlerTest(BaseTest):
def tearDown(self): def tearDown(self):
"""Shutdown the UDP server.""" """Shutdown the UDP server."""
try: try:
self.server.stop(2.0) if self.server:
self.root_logger.removeHandler(self.sock_hdlr) self.server.stop(2.0)
self.sock_hdlr.close() if self.sock_hdlr:
self.root_logger.removeHandler(self.sock_hdlr)
self.sock_hdlr.close()
finally: finally:
BaseTest.tearDown(self) BaseTest.tearDown(self)
@ -1576,6 +1600,8 @@ class DatagramHandlerTest(BaseTest):
def test_output(self): def test_output(self):
# The log message sent to the DatagramHandler is properly received. # The log message sent to the DatagramHandler is properly received.
if self.server_exception:
self.skipTest(self.server_exception)
logger = logging.getLogger("udp") logger = logging.getLogger("udp")
logger.error("spam") logger.error("spam")
self.handled.wait() self.handled.wait()
@ -1600,7 +1626,7 @@ class UnixDatagramHandlerTest(DatagramHandlerTest):
def tearDown(self): def tearDown(self):
DatagramHandlerTest.tearDown(self) DatagramHandlerTest.tearDown(self)
os.remove(self.address) support.unlink(self.address)
@unittest.skipUnless(threading, 'Threading required for this test.') @unittest.skipUnless(threading, 'Threading required for this test.')
class SysLogHandlerTest(BaseTest): class SysLogHandlerTest(BaseTest):
@ -1615,9 +1641,17 @@ class SysLogHandlerTest(BaseTest):
"""Set up a UDP server to receive log messages, and a SysLogHandler """Set up a UDP server to receive log messages, and a SysLogHandler
pointing to that server's address and port.""" pointing to that server's address and port."""
BaseTest.setUp(self) BaseTest.setUp(self)
self.server = server = self.server_class(self.address, # Issue #29177: deal with errors that happen during setup
self.handle_datagram, 0.01) self.server = self.sl_hdlr = self.server_exception = None
server.start() try:
self.server = server = self.server_class(self.address,
self.handle_datagram, 0.01)
server.start()
# Uncomment next line to test error recovery in setUp()
# raise OSError('dummy error raised')
except OSError as e:
self.server_exception = e
return
server.ready.wait() server.ready.wait()
hcls = logging.handlers.SysLogHandler hcls = logging.handlers.SysLogHandler
if isinstance(server.server_address, tuple): if isinstance(server.server_address, tuple):
@ -1630,11 +1664,13 @@ class SysLogHandlerTest(BaseTest):
self.handled = threading.Event() self.handled = threading.Event()
def tearDown(self): def tearDown(self):
"""Shutdown the UDP server.""" """Shutdown the server."""
try: try:
self.server.stop(2.0) if self.server:
self.root_logger.removeHandler(self.sl_hdlr) self.server.stop(2.0)
self.sl_hdlr.close() if self.sl_hdlr:
self.root_logger.removeHandler(self.sl_hdlr)
self.sl_hdlr.close()
finally: finally:
BaseTest.tearDown(self) BaseTest.tearDown(self)
@ -1643,6 +1679,8 @@ class SysLogHandlerTest(BaseTest):
self.handled.set() self.handled.set()
def test_output(self): def test_output(self):
if self.server_exception:
self.skipTest(self.server_exception)
# The log message sent to the SysLogHandler is properly received. # The log message sent to the SysLogHandler is properly received.
logger = logging.getLogger("slh") logger = logging.getLogger("slh")
logger.error("sp\xe4m") logger.error("sp\xe4m")
@ -1675,7 +1713,7 @@ class UnixSysLogHandlerTest(SysLogHandlerTest):
def tearDown(self): def tearDown(self):
SysLogHandlerTest.tearDown(self) SysLogHandlerTest.tearDown(self)
os.remove(self.address) support.unlink(self.address)
@unittest.skipUnless(threading, 'Threading required for this test.') @unittest.skipUnless(threading, 'Threading required for this test.')
class HTTPHandlerTest(BaseTest): class HTTPHandlerTest(BaseTest):