mirror of https://github.com/python/cpython
gh-120868: Fix breaking change in `logging.config` when using `QueueHandler` (GH-120872)
This commit is contained in:
parent
4be1f37b20
commit
7d9c68513d
|
@ -780,25 +780,44 @@ class DictConfigurator(BaseConfigurator):
|
|||
# if 'handlers' not in config:
|
||||
# raise ValueError('No handlers specified for a QueueHandler')
|
||||
if 'queue' in config:
|
||||
from multiprocessing.queues import Queue as MPQueue
|
||||
from multiprocessing import Manager as MM
|
||||
proxy_queue = MM().Queue()
|
||||
proxy_joinable_queue = MM().JoinableQueue()
|
||||
qspec = config['queue']
|
||||
if not isinstance(qspec, (queue.Queue, MPQueue,
|
||||
type(proxy_queue), type(proxy_joinable_queue))):
|
||||
if isinstance(qspec, str):
|
||||
q = self.resolve(qspec)
|
||||
if not callable(q):
|
||||
raise TypeError('Invalid queue specifier %r' % qspec)
|
||||
q = q()
|
||||
elif isinstance(qspec, dict):
|
||||
if '()' not in qspec:
|
||||
raise TypeError('Invalid queue specifier %r' % qspec)
|
||||
q = self.configure_custom(dict(qspec))
|
||||
else:
|
||||
|
||||
if isinstance(qspec, str):
|
||||
q = self.resolve(qspec)
|
||||
if not callable(q):
|
||||
raise TypeError('Invalid queue specifier %r' % qspec)
|
||||
config['queue'] = q
|
||||
config['queue'] = q()
|
||||
elif isinstance(qspec, dict):
|
||||
if '()' not in qspec:
|
||||
raise TypeError('Invalid queue specifier %r' % qspec)
|
||||
config['queue'] = self.configure_custom(dict(qspec))
|
||||
else:
|
||||
from multiprocessing.queues import Queue as MPQueue
|
||||
|
||||
if not isinstance(qspec, (queue.Queue, MPQueue)):
|
||||
# Safely check if 'qspec' is an instance of Manager.Queue
|
||||
# / Manager.JoinableQueue
|
||||
|
||||
from multiprocessing import Manager as MM
|
||||
from multiprocessing.managers import BaseProxy
|
||||
|
||||
# if it's not an instance of BaseProxy, it also can't be
|
||||
# an instance of Manager.Queue / Manager.JoinableQueue
|
||||
if isinstance(qspec, BaseProxy):
|
||||
# Sometimes manager or queue creation might fail
|
||||
# (e.g. see issue gh-120868). In that case, any
|
||||
# exception during the creation of these queues will
|
||||
# propagate up to the caller and be wrapped in a
|
||||
# `ValueError`, whose cause will indicate the details of
|
||||
# the failure.
|
||||
mm = MM()
|
||||
proxy_queue = mm.Queue()
|
||||
proxy_joinable_queue = mm.JoinableQueue()
|
||||
if not isinstance(qspec, (type(proxy_queue), type(proxy_joinable_queue))):
|
||||
raise TypeError('Invalid queue specifier %r' % qspec)
|
||||
else:
|
||||
raise TypeError('Invalid queue specifier %r' % qspec)
|
||||
|
||||
if 'listener' in config:
|
||||
lspec = config['listener']
|
||||
if isinstance(lspec, type):
|
||||
|
|
|
@ -3928,6 +3928,50 @@ class ConfigDictTest(BaseTest):
|
|||
msg = str(ctx.exception)
|
||||
self.assertEqual(msg, "Unable to configure handler 'ah'")
|
||||
|
||||
@threading_helper.requires_working_threading()
|
||||
@support.requires_subprocess()
|
||||
@patch("multiprocessing.Manager")
|
||||
def test_config_queue_handler_does_not_create_multiprocessing_manager(self, manager):
|
||||
# gh-120868
|
||||
|
||||
from multiprocessing import Queue as MQ
|
||||
|
||||
q1 = {"()": "queue.Queue", "maxsize": -1}
|
||||
q2 = MQ()
|
||||
q3 = queue.Queue()
|
||||
|
||||
for qspec in (q1, q2, q3):
|
||||
self.apply_config(
|
||||
{
|
||||
"version": 1,
|
||||
"handlers": {
|
||||
"queue_listener": {
|
||||
"class": "logging.handlers.QueueHandler",
|
||||
"queue": qspec,
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
manager.assert_not_called()
|
||||
|
||||
@patch("multiprocessing.Manager")
|
||||
def test_config_queue_handler_invalid_config_does_not_create_multiprocessing_manager(self, manager):
|
||||
# gh-120868
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
self.apply_config(
|
||||
{
|
||||
"version": 1,
|
||||
"handlers": {
|
||||
"queue_listener": {
|
||||
"class": "logging.handlers.QueueHandler",
|
||||
"queue": object(),
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
manager.assert_not_called()
|
||||
|
||||
@support.requires_subprocess()
|
||||
def test_multiprocessing_queues(self):
|
||||
# See gh-119819
|
||||
|
|
Loading…
Reference in New Issue