[Python-checkins] cpython (3.6): Fixes #29177: Improved resilience of logging tests which use socket servers.
vinay.sajip
python-checkins at python.org
Mon Jan 9 11:55:32 EST 2017
https://hg.python.org/cpython/rev/6bf563472ccd
changeset: 106072:6bf563472ccd
branch: 3.6
parent: 106070:af5b34b2d169
user: Vinay Sajip <vinay_sajip at yahoo.co.uk>
date: Mon Jan 09 16:54:12 2017 +0000
summary:
Fixes #29177: Improved resilience of logging tests which use socket servers.
Thanks to Xavier de Gaye for the report and patch improvements.
files:
Lib/test/test_logging.py | 86 ++++++++++++++++++++-------
1 files changed, 62 insertions(+), 24 deletions(-)
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
--- a/Lib/test/test_logging.py
+++ b/Lib/test/test_logging.py
@@ -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
# documentation for any purpose and without fee is hereby granted,
@@ -16,7 +16,7 @@
"""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
@@ -1440,9 +1440,17 @@
"""Set up a TCP server to receive log messages, and a SocketHandler
pointing to that server's address and port."""
BaseTest.setUp(self)
- self.server = server = self.server_class(self.address,
- self.handle_socket, 0.01)
- server.start()
+ # Issue #29177: deal with errors that happen during setup
+ self.server = self.sock_hdlr = self.server_exception = None
+ 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()
hcls = logging.handlers.SocketHandler
if isinstance(server.server_address, tuple):
@@ -1457,9 +1465,11 @@
def tearDown(self):
"""Shutdown the TCP server."""
try:
- self.server.stop(2.0)
- self.root_logger.removeHandler(self.sock_hdlr)
- self.sock_hdlr.close()
+ if self.server:
+ self.server.stop(2.0)
+ if self.sock_hdlr:
+ self.root_logger.removeHandler(self.sock_hdlr)
+ self.sock_hdlr.close()
finally:
BaseTest.tearDown(self)
@@ -1480,6 +1490,8 @@
def test_output(self):
# The log message sent to the SocketHandler is properly received.
+ if self.server_exception:
+ self.skipTest(self.server_exception)
logger = logging.getLogger("tcp")
logger.error("spam")
self.handled.acquire()
@@ -1488,6 +1500,8 @@
self.assertEqual(self.log_output, "spam\neggs\n")
def test_noserver(self):
+ if self.server_exception:
+ self.skipTest(self.server_exception)
# Avoid timing-related failures due to SocketHandler's own hard-wired
# one-second timeout on socket.create_connection() (issue #16264).
self.sock_hdlr.retryStart = 2.5
@@ -1528,7 +1542,7 @@
def tearDown(self):
SocketHandlerTest.tearDown(self)
- os.remove(self.address)
+ support.unlink(self.address)
@unittest.skipUnless(threading, 'Threading required for this test.')
class DatagramHandlerTest(BaseTest):
@@ -1543,9 +1557,17 @@
"""Set up a UDP server to receive log messages, and a DatagramHandler
pointing to that server's address and port."""
BaseTest.setUp(self)
- self.server = server = self.server_class(self.address,
- self.handle_datagram, 0.01)
- server.start()
+ # Issue #29177: deal with errors that happen during setup
+ self.server = self.sock_hdlr = self.server_exception = None
+ 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()
hcls = logging.handlers.DatagramHandler
if isinstance(server.server_address, tuple):
@@ -1560,9 +1582,11 @@
def tearDown(self):
"""Shutdown the UDP server."""
try:
- self.server.stop(2.0)
- self.root_logger.removeHandler(self.sock_hdlr)
- self.sock_hdlr.close()
+ if self.server:
+ self.server.stop(2.0)
+ if self.sock_hdlr:
+ self.root_logger.removeHandler(self.sock_hdlr)
+ self.sock_hdlr.close()
finally:
BaseTest.tearDown(self)
@@ -1576,6 +1600,8 @@
def test_output(self):
# The log message sent to the DatagramHandler is properly received.
+ if self.server_exception:
+ self.skipTest(self.server_exception)
logger = logging.getLogger("udp")
logger.error("spam")
self.handled.wait()
@@ -1600,7 +1626,7 @@
def tearDown(self):
DatagramHandlerTest.tearDown(self)
- os.remove(self.address)
+ support.unlink(self.address)
@unittest.skipUnless(threading, 'Threading required for this test.')
class SysLogHandlerTest(BaseTest):
@@ -1615,9 +1641,17 @@
"""Set up a UDP server to receive log messages, and a SysLogHandler
pointing to that server's address and port."""
BaseTest.setUp(self)
- self.server = server = self.server_class(self.address,
- self.handle_datagram, 0.01)
- server.start()
+ # Issue #29177: deal with errors that happen during setup
+ self.server = self.sl_hdlr = self.server_exception = None
+ 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()
hcls = logging.handlers.SysLogHandler
if isinstance(server.server_address, tuple):
@@ -1630,11 +1664,13 @@
self.handled = threading.Event()
def tearDown(self):
- """Shutdown the UDP server."""
+ """Shutdown the server."""
try:
- self.server.stop(2.0)
- self.root_logger.removeHandler(self.sl_hdlr)
- self.sl_hdlr.close()
+ if self.server:
+ self.server.stop(2.0)
+ if self.sl_hdlr:
+ self.root_logger.removeHandler(self.sl_hdlr)
+ self.sl_hdlr.close()
finally:
BaseTest.tearDown(self)
@@ -1643,6 +1679,8 @@
self.handled.set()
def test_output(self):
+ if self.server_exception:
+ self.skipTest(self.server_exception)
# The log message sent to the SysLogHandler is properly received.
logger = logging.getLogger("slh")
logger.error("sp\xe4m")
@@ -1675,7 +1713,7 @@
def tearDown(self):
SysLogHandlerTest.tearDown(self)
- os.remove(self.address)
+ support.unlink(self.address)
@unittest.skipUnless(threading, 'Threading required for this test.')
class HTTPHandlerTest(BaseTest):
--
Repository URL: https://hg.python.org/cpython
More information about the Python-checkins
mailing list