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

gh-132106: Allow logging.handlers.QueueListener to be used as a context manager (#132107)

Browse source
This commit is contained in:
Charles Machalow 2025-04-12 05:00:04 -07:00 • committed by GitHub
parent ad3bbe8fbc
commit 517e96b9ed
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 54 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

17
Lib/test/test_logging.py
View file

@ -4311,8 +4311,6 @@ class QueueHandlerTest(BaseTest):
self.assertEqual(formatted_msg, log_record.msg)
self.assertEqual(formatted_msg, log_record.message)
@unittest.skipUnless(hasattr(logging.handlers, 'QueueListener'),
'logging.handlers.QueueListener required for this test')
def test_queue_listener(self):
handler = TestHandler(support.Matcher())
listener = logging.handlers.QueueListener(self.queue, handler)
@ -4347,8 +4345,17 @@ class QueueHandlerTest(BaseTest):
self.assertTrue(handler.matches(levelno=logging.CRITICAL, message='6'))
handler.close()
@unittest.skipUnless(hasattr(logging.handlers, 'QueueListener'),
'logging.handlers.QueueListener required for this test')
def test_queue_listener_context_manager(self):
handler = TestHandler(support.Matcher())
with logging.handlers.QueueListener(self.queue, handler) as listener:
self.assertIsInstance(listener, logging.handlers.QueueListener)
self.assertIsNotNone(listener._thread)
self.assertIsNone(listener._thread)
# doesn't hurt to call stop() more than once.
listener.stop()
self.assertIsNone(listener._thread)
def test_queue_listener_with_StreamHandler(self):
# Test that traceback and stack-info only appends once (bpo-34334, bpo-46755).
listener = logging.handlers.QueueListener(self.queue, self.root_hdlr)
@ -4363,8 +4370,6 @@ class QueueHandlerTest(BaseTest):
self.assertEqual(self.stream.getvalue().strip().count('Traceback'), 1)
self.assertEqual(self.stream.getvalue().strip().count('Stack'), 1)
@unittest.skipUnless(hasattr(logging.handlers, 'QueueListener'),
'logging.handlers.QueueListener required for this test')
def test_queue_listener_with_multiple_handlers(self):
# Test that queue handler format doesn't affect other handler formats (bpo-35726).
self.que_hdlr.setFormatter(self.root_formatter)