# Original logger output is empty.
self.assert_log_lines([])
-class LogRecordStreamHandler(StreamRequestHandler):
+ def test_logger_disabling(self):
+ self.apply_config(self.disable_test)
+ logger = logging.getLogger('foo')
+ self.assertFalse(logger.disabled)
+ self.apply_config(self.disable_test)
+ self.assertTrue(logger.disabled)
+ self.apply_config(self.disable_test, disable_existing_loggers=False)
+ self.assertFalse(logger.disabled)
+
- """Handler for a streaming logging request. It saves the log message in the
- TCP server's 'log_output' attribute."""
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class SocketHandlerTest(BaseTest):
+
+ """Test for SocketHandler objects."""
+
+ def setUp(self):
+ """Set up a TCP server to receive log messages, and a SocketHandler
+ pointing to that server's address and port."""
+ BaseTest.setUp(self)
+ addr = ('localhost', 0)
+ self.server = server = TestTCPServer(addr, self.handle_socket,
+ 0.01)
+ server.start()
+ server.ready.wait()
+ self.sock_hdlr = logging.handlers.SocketHandler('localhost',
+ server.port)
+ self.log_output = ''
+ self.root_logger.removeHandler(self.root_logger.handlers[0])
+ self.root_logger.addHandler(self.sock_hdlr)
+ self.handled = threading.Semaphore(0)
- TCP_LOG_END = "!!!END!!!"
+ def tearDown(self):
+ """Shutdown the TCP server."""
+ try:
+ self.server.stop(2.0)
+ self.root_logger.removeHandler(self.sock_hdlr)
+ self.sock_hdlr.close()
+ finally:
+ BaseTest.tearDown(self)
- def handle(self):
- """Handle multiple requests - each expected to be of 4-byte length,
- followed by the LogRecord in pickle format. Logs the record
- according to whatever policy is configured locally."""
+ def handle_socket(self, request):
+ conn = request.connection
while True:
- chunk = self.connection.recv(4)
+ chunk = conn.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack(">L", chunk)[0]