def _configure_queue_handler(self, klass, **kwargs):
if 'queue' in kwargs:
- q = kwargs['queue']
+ q = kwargs.pop('queue')
else:
q = queue.Queue() # unbounded
- rhl = kwargs.get('respect_handler_level', False)
- if 'listener' in kwargs:
- lklass = kwargs['listener']
- else:
- lklass = logging.handlers.QueueListener
- listener = lklass(q, *kwargs.get('handlers', []), respect_handler_level=rhl)
- handler = klass(q)
+
+ rhl = kwargs.pop('respect_handler_level', False)
+ lklass = kwargs.pop('listener', logging.handlers.QueueListener)
+ handlers = kwargs.pop('handlers', [])
+
+ listener = lklass(q, *handlers, respect_handler_level=rhl)
+ handler = klass(q, **kwargs)
handler.listener = listener
return handler
}
logging.config.dictConfig(config)
+ # gh-118868: check if kwargs are passed to logging QueueHandler
+ def test_kwargs_passing(self):
+ class CustomQueueHandler(logging.handlers.QueueHandler):
+ def __init__(self, *args, **kwargs):
+ super().__init__(queue.Queue())
+ self.custom_kwargs = kwargs
+
+ custom_kwargs = {'foo': 'bar'}
+
+ config = {
+ 'version': 1,
+ 'handlers': {
+ 'custom': {
+ 'class': CustomQueueHandler,
+ **custom_kwargs
+ },
+ },
+ 'root': {
+ 'level': 'DEBUG',
+ 'handlers': ['custom']
+ }
+ }
+
+ logging.config.dictConfig(config)
+
+ handler = logging.getHandlerByName('custom')
+ self.assertEqual(handler.custom_kwargs, custom_kwargs)
+
+
class ManagerTest(BaseTest):
def test_manager_loggerclass(self):
logged = []