celery-celery-9815-9853

from celery import signals
from celery.utils.nodenames import anon_nodename
from celery.utils.saferepr import saferepr
from celery.utils.text import indent as textindent
from celery.utils.time import maybe_make_aware

from . import routes as _routes

__all__ = ('AMQP', 'Queues', 'task_message')

#: earliest date supported by time.mktime.
INT_MIN = -2147483648

#: Human readable queue declaration.
QUEUE_FORMAT = """
.> {0.name:<16} exchange={0.exchange.name}({0.exchange.type}) \
…[8012 chars omitted]…
              queue_arguments = {'x-queue-type': 'quorum'}
            queues = (Queue(conf.task_default_queue,
                            exchange=self.default_exchange,
                            routing_key=default_routing_key,
                            queue_arguments=queue_arguments),)
        autoexchange = (self.autoexchange if autoexchange is None
                        else autoexchange)

celery/worker/consumer/consumer.py:723:    def max_prefetch_count(self):
celery/worker/worker.py:54:enable the `task_create_missing_queues` setting.

        save_meta_as_text=Option(True, type='bool'),
    ),
    security=Namespace(
        __old__=old_ns('celery_security'),

        certificate=Option(type='string'),
        cert_store=Option(type='string'),
        key=Option(type='string'),
        key_password=Option(type='bytes'),
        digest=Option(DEFAULT_SECURITY_DIGEST, type='string'),
    ),
    databa
…[2165 chars omitted]…
),
        remote_tracebacks=Option(False, type='bool'),
        routes=Option(type='any'),
        send_sent_event=Option(
            False, type='bool', old={'celery_send_task_sent_event'},
        ),
        serializer=Option('json', old={'celery_task_serializer'}),
        soft_time_limit=Option(
            type='float', old={'celeryd_task_soft_time_limit'},
        ),
        time_limit=Option(

            False, type='bool'
        ),
        soft_shutdown_timeout=Option(0.0, type='float'),
        enable_soft_shutdown_on_idle=Option(False, type='bool'),
        concurrency=Option(None, type='int'),
        consumer=Option('celery.worker.consumer:Consumer', type='string'),
        direct=Option(False, type='bool', old={'celery_worker_direct'}),
        disable_rate_limits=Opt
…[3162 chars omitted]…
mbu.Connection): Optional specific connection
                instance to use.  If not provided a connection will
215:        if self.hub or getattr(self.pool, 'is_green', False):
256:        limit = rate(getattr(type, 'rate_limit', None))
320:    def _limit_post_eta(self, request, bucket, tokens):
489:            if getattr(conn, 'alt', None) and interval == 0:
572:    def apply_eta_task(self, task):

from billiard.common import restart_state
from billiard.exceptions import RestartFreqExceeded
from kombu.asynchronous.semaphore import DummyLock
from kombu.exceptions import ContentDisallowed, DecodeError
from kombu.utils.compat import _detect_environment
from kombu.utils.encoding import safe_repr
from kombu.utils.limits import TokenBucket
from vine import ppartial, promise

from celery import bootsteps, signals
from celery.app.trace import build_tracer
from celery.exceptions import (CPendingDeprecationWarning, InvalidTaskError, NotRegistered, WorkerShutdown,
                   
…[8629 chars omitted]…
e acknowledged, but it has already been acknowledged.
            return False  # Do not cancel and allow it to gracefully finish as it has already been acknowledged.

        requests_to_cancel = tuple(filter(should_cancel, active_requests))

        if requests_to_cancel:
            for request in requests_to_cancel:
                request.cancel(self.pool)


class Evloop(bootsteps.StartStopStep):

test_loaders.py
test_log.py
test_preload_cli.py
test_registry.py
test_routes.py
test_schedules.py
test_trace.py
test_utils.py

t/unit/events:
__init__.py
test_cursesmon.py
test_dumper.py
test_events.py
test_snapshot.py
test_state.py

t/unit/worker:
__init__.py
test_autoscale.py
test_bootsteps.py
test_components.py
test_consumer.py
test_control.py
test_heartbeat.py
test_loops.py
test_native_delayed_delivery.py
test_request.py
test_revoke.py
test_state.py
test_strategy.py
test_worker.py
75:        assert Queues(None) == {}
78:        q = Queues()
85:        q = Queues(default_exchange=Exchange('bar'))
93:        q = Queues()
99:        q = Queues()
106:        q = Queues(default_exchange=ex)
111:        q = Queues()
133:        queues = Queues(**queues_kwargs)
240:        x = self.app.amqp.Queues({}, max_priority=23)

565:        c = control.Control(self.app)
566:        assert c.mailbox.namespace == 'test_exchange'
193:class test_EventReceiver:
205:        r = self.app
…[3438 chars omitted]…
:    def test_qos_global_worker_detect_quorum_queues_true_no_quorum_queues(self):
683:        tasks = Tasks(c)
684:        assert tasks.qos_global(c) is True
686:    def test_qos_global_worker_detect_quorum_queues_true_with_quorum_queues(self):
690:        tasks = Tasks(c)
691:        assert tasks.qos_global(c) is False
693:    def test_log_when_qos_is_false(self, caplog):
698:        tasks = Tasks(c)

        amqp.producer_pool = closed_pool
        assert closed_pool is self.app.control.mailbox.producer_pool
        self.app.control._after_fork()
        new_pool = Mock(name='new pool')
        amqp.producer_pool = new_pool
        assert new_pool is self.app.control.mailbox.producer_pool

    def test_control_exchange__default(self):
        c = control.Control(self.app)
        assert c.mailbox.namespace == 'celery'

    def test_control_exch
…[7846 chars omitted]…
 assert record.levelname == "INFO"
        assert record.msg == "Global QoS is disabled. Prefetch count in now static."


class test_Agent:

    def test_start(self):
        c = Mock()
        agent = Agent(c)
        agent.instantiate = Mock()
        agent.agent_cls = 'foo:Agent'
        assert agent.create(c) is not None
        agent.instantiate.assert_called_with(agent.agent_cls, c.connection)

DEFAULT_POOL = 'prefork'

DEFAULT_ACCEPT_CONTENT = ('json',)
DEFAULT_PROCESS_LOG_FMT = """
    [%(asctime)s: %(levelname)s/%(processName)s] %(message)s
""".strip()
DEFAULT_TASK_LOG_FMT = """[%(asctime)s: %(levelname)s/%(processName)s] \
%(task_name)s[%(task_id)s]: %(message)s"""

DEFAULT_SECURITY_DIGEST = 'sha256'


OLD_NS = {'celery_{0}'}
OLD_NS_BEAT = {'celerybeat_{0}'}
OLD_NS_WORKER = {'celeryd_{0}'}

searchresult = namedtuple('searchresult', ('namespace', 'key', 'type'))


def Namespace(__old__=None, **options
…[6808 chars omitted]…
ween_retries_ms=Option(10000, type='int'),
        backend_max_retries=Option(float("inf"), type='float'),
        backend_base_sleep_between_retries_ms=Option(10, type='int'),
        backend_always_retry=Option(False, type='bool'),
    ),
    elasticsearch=Namespace(
        __old__=old_ns('celery_elasticsearch'),

        retry_on_timeout=Option(type='bool'),
        max_retries=Option(type='int'),

/worker/loops.py:94:                update_qos()
celery/worker/loops.py:108:def synloop(obj, connection, consumer, blueprint, hub, qos,
celery/worker/loops.py:128:        if qos.prev != qos.value:
celery/worker/loops.py:129:            qos.update()
celery/worker/strategy.py:194:            consumer.qos.increment_eventually()
celery/worker/strategy.py:198:            consumer.qos.increment_eventually()

        if (req.expires or req.id in revoked_tasks) and req.revoked():
            return

        signals.task_received.send(sender=consumer, request=req)

        if task_sends_events:
            send_event(
                'task-received',
                uuid=req.id, name=req.name,
                args=req.argsrepr, kwargs=req.kwargsrepr,
                root_id=req.root_id, parent_id=req.parent_id,
                retries=req.reques
…[1134 chars omitted]…
ket, 1)

        task_reserved(req)
        if callbacks:
            [callback(req) for callback in callbacks]
        handle(req)
    return task_message_handler
celery/worker/consumer/consumer.py:572:    def apply_eta_task(self, task):
celery/worker/strategy.py:121:    apply_eta_task = consumer.apply_eta_task
celery/worker/strategy.py:199:            call_at(eta, apply_eta_task, (req,), priority=6)

    @conf.setter
    def conf(self, d):
        self._conf = d

    @cached_property
    def control(self):
        """Remote control: :class:`~@control`."""
        return instantiate(self.control_cls, app=self)

    @cached_property
    def events(self):
        """Consuming and sending events: :class:`~@events`."""
        return instantiate(self.events_cls, app=self)

    @cached_property
    def loader(self):
        """Current loader instance."""
        return get_loader_cls(self.loader_cls)(app=self)

    @cached_property
    def log(self):
        """Loggi
…[220 chars omitted]…
son = reason
        self.requeue = requeue
        super().__init__(reason, requeue)

    def __repr__(self):
        return f'reject requeue={self.requeue}: {self.reason}'


class ImproperlyConfigured(CeleryError):
    """Celery is somehow improperly configured."""


class SecurityError(CeleryError):
    """Security related exception."""


class TaskError(CeleryError):
    """Task related errors."""

docs/userguide/configuration.rst:2613:``task_create_missing_queues``
docs/userguide/configuration.rst:2641:.. setting:: task_default_queue_type
docs/userguide/configuration.rst:2643:``task_default_queue_type``
docs/userguide/routing.rst:28::setting:`task_create_missing_queues` setting (on by default).
docs/userguide/workers.rst:877::setting:`task_create_missing_queues` option).

guration setting will no longer determine
  whether broker connection retries are made during startup in Celery 6.0 and above.
  If you wish to refrain from retrying connections on startup,
  you should set broker_connection_retry_on_startup to False instead.
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
241 passed, 18 warnings, 37 subtests passed in 3.79s

from importlib import import_module, reload
from unittest.mock import MagicMock, Mock, patch

import pytest
from kombu import Queue

from celery.backends.cache import CacheBackend, DummyClient
# we have to import the pytest plugin fixtures here,
# in case user did not do the `python setup.py develop` yet,
# that installs the pytest plugin into the setuptools registry.
from celery.contrib.pytest import celery_app, celery_enable_logging, celery_parameters, depends_on_current_app
from celery.contrib.testing.app import TestApp, Trap
from celery.contrib.testing.mocks import TaskMessage, TaskMessage1, task_message_from_sig

# Tricks flake8 into silencing redefining fixtures warnings.
__all
…[4063 chars omitted]…

    _state.task_join_will_block = prev_state_join_block
    _state._set_task_join_will_block(False)


@pytest.fixture(scope='session', autouse=True)
def record_threads_at_startup(request):
    try:
        request.session._threads_at_startup
    except AttributeError:
        request.session._threads_at_startup = alive_threads()


./t/integration/conftest.py
./t/smoke/conftest.py
./t/unit/conftest.py

    """Redefine this fixture to change the init parameters of test Celery app.

    The dict returned by your fixture will then be used
    as parameters when instantiating :class:`~celery.Celery`.
    """
    return {}


@pytest.fixture(scope='session')
def celery_worker_parameters():
    # type: () -> Mapping[str, Any]
    """Redefine this fixture to change the init parameters of Celery workers.

    This can be used e. g. to define queues the worker will cons
…[1308 chars omitted]…
dule in celery_includes:
            celery_app.loader.import_task_module(module)
        with worker.start_worker(celery_app,
                                 pool=celery_worker_pool,
                                 **celery_worker_parameters) as w:
            yield w


@pytest.fixture()
def depends_on_current_app(celery_app):
    """Fixture that sets app as current."""
    celery_app.set_current()

    from .certificate import Certificate, FSCertStore
celery/security/certificate.py:9: in <module>
    from cryptography.hazmat.backends import default_backend
E   ModuleNotFoundError: No module named 
…[3020 chars omitted]…
nit/backends/test_s3.py
ERROR t/unit/security/test_certificate.py
ERROR t/unit/security/test_key.py
ERROR t/unit/security/test_security.py
ERROR t/unit/security/test_serialization.py
ERROR t/unit/tasks/test_canvas.py
ERROR t/unit/utils/test_annotations.py
ERROR t/unit/utils/test_functional.py
!!!!!!!!!!!!!!!!!!! Interrupted: 12 errors during collection !!!!!!!!!!!!!!!!!!!
9 skipped, 12 errors in 8.00s

    File "/repo/celery/app/trace.py", line 606, in trace_task
      I, _, _, _ = on_error(task_request, exc)
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^
    File "/repo/celery/app/trace.py", line 407, in on_error
      R = I.handle_error_state(
          ^^^^^^^^^^^^^^^^^^^^^
    File "/repo/celery/app/trace.py", line 175, in handle_error_state
      return {
             ^
 
…[600 chars omitted]…
ning(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
ERROR t/unit/app/test_preload_cli.py::test_preload_options[subcommand_with_params0]
ERROR t/unit/app/test_preload_cli.py::test_preload_options[subcommand_with_params1]
1082 passed, 12 skipped, 25 warnings, 2 errors, 37 subtests passed in 27.93s

     """
 
@@ -55,7 +58,9 @@ class Queues(dict):
 
     def __init__(self, queues=None, default_exchange=None,
                  create_missing=True, autoexchange=None,
-                 max_priority=None, default_routing_key=None):
+         
…[8363 chars omitted]…
eps.StartStopStep):
                 prefetch_count=prefetch_count,
                 apply_global=qos_global,
             )
-        c.qos = QoS(set_prefetch_count, c.initial_prefetch_count)
+        c.qos = QoS(
+            set_prefetch_count, c.initial_prefetch_count,
+            max_prefetch=c.app.conf.worker_eta_task_limit,
+        )
 
     def stop(self, c):
         """Stop task consumer."""

/root/.cache/pip/http-v2/1/e/e/8/a/1ee8a229287
…[908 chars omitted]…
che/pip/http-v2/b/1/1/9/0/b11909f451f61fb29636217082b1c638b177c59eac335e0200a136e6.body
/root/.cache/pip/http-v2/b/1/1/9/0/b11909f451f61fb29636217082b1c638b177c59eac335e0200a136e6
/root/.cache/pip/http-v2/b/0/2/1/8/b02187b079c9cdd7c6026b274bfbf83fd3d25bd63fc7ab60b94a568f
__editable__.celery-5.5.3.pth
__editable___celery_5_5_3_finder.py
celery-5.5.3.dist-info
pytest_celery
pytest_celery-1.2.1.dist-info

    def teardown_method(self):
        if self._prev:
            sys.modules['celery.app.defaults'] = self._prev

    def test_option_repr(self):
        assert repr(NAMESPACES['broker']['url'])

    def test_any(self):
        val = object()
        assert self.defaults.Option.typemap['any'](val) is val

    def test_compat_indices(self):
        assert not any(key.isupper() for key in DEFAULTS)
        assert not any(key.islower() for key in _OLD_DEFAULTS)
        assert not any(key.isupper() for key in _TO_OLD_KEY)
        assert not any(key.islower() for ke
…[306 chars omitted]…
r key in _TO_NEW_KEY:
            assert key in _OLD_SETTING_KEYS
        for key in _TO_OLD_KEY:
            assert key in SETTING_KEYS

    def test_find(self):
        find = self.defaults.find

        assert find('default_queue')[2].default == 'celery'
        assert find('task_default_exchange')[2] is None

    @property
    def defaults(self):
        return import_module('celery.app.defaults')

        # allows Queue('name', [binding(...), binding(...), ...])
        if isinstance(exchange, (list, tuple, set)):
            self.bindings |= set(exchange)
        if self.bindings:
            self.exchange = None

        # exclusive implies auto-delete.
        if self.exclusive:
            self.auto_delete = True
        self.maybe_bind(channel)

…[1468 chars omitted]…
binding_key,
                     durable=q_durable,
                     exclusive=options.get('exclusive'),
                     auto_delete=q_auto_delete,
                     no_ack=options.get('no_ack'),
                     queue_arguments=q_arguments,
                     binding_arguments=b_arguments,
                     consumer_arguments=c_arguments,
                     bindings=bindings)

t/unit/events/test_events.py:314:            r = self.app.events.Receiver(

__all__ = (
    'Event', 'EventDispatcher', 'EventReceiver',
    'event_exchange', 'get_exchange', 'group_from',
)

ERROR t/unit/app/test_app.py
!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
1 error in 0.40s

112             exclusive=self.queue_exclusive,
113             auto_delete=not self.queue_durable,

celery.egg-info
docker
docs
examples
extra
helm-chart
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini

t/smoke/tests/test_consumer.py:107:    def test_max_prefetch_not_passed_on_broker_restart(self, celery_setup: CeleryTestSetup):
t/unit/app/test_app.py:292:        with self.Celery(task_create_missing_queues=False) as app:
t/unit/app/test_app.py:293:            assert app.conf.task_create_missing_queues is False
t/unit/tasks/test_tasks.py:61:class TasksCase:

from celery import signals
from celery.utils.nodenames import anon_nodename
from celery.utils.saferepr import saferepr
from celery.utils.text import indent as textindent
from celery.utils.time import maybe_make_aware

from . import routes as _routes

__all__ = ('AMQP', 'Queues', 'task_message')

#: earliest date supported by time.mktime.
INT_MIN = -2147483648

#: Human readable queue declaration.
QUEUE_FORMAT = """
.> {0.name:<16} exchange={0.exchange.name}({0.exchange.type}) \
…[8630 chars omitted]…
tance=isinstance):
        if isinstance(body, list):  # celery 4.0+: List of events
            process, from_message = self.process, self.event_from_message
            [process(*from_message(event)) for event in body]
        else:
            self.process(*self.event_from_message(body))

    @property
    def connection(self):
        return self.channel.connection.client if self.channel else None

    # Exchange class/function used when defining automatic queues.
    # For example, you can use ``autoexchange = lambda n: None`` to use the
    # AMQP default exchange: a shortcut to bypass routing
    # and instead send directly to the queue named in the routing key.
    autoexchange = None

    #: Max size of positional argument representation used for
    #: logging purposes.
    argsrepr_maxsize = 1024

    #: Max size of keyword argument representation used for logging purposes.
    kwargsrepr_maxsize = 1024

    def __init__(self, app):
        self.app = app
        self.task_protocols = {
           
…[8630 chars omitted]…
tion(
            timedelta(days=1),
            type='float', old={'celery_task_result_expires'},
        ),
        persistent=Option(None, type='bool'),
        extended=Option(False, type='bool'),
        serializer=Option('json'),
        backend_transport_options=Option({}, type='dict'),
        chord_retry_interval=Option(1.0, type='float'),
        chord_join_timeout=Option(3.0, type='float'),

        if self.max_priority is not None:
            if queue.queue_arguments is None:
                queue.queue_arguments = {}
            self._set_max_priority(queue.queue_arguments)
        self[queue.name] = queue
        return queue

    def _set_max_priority(self, args):
        if 'x-max-priority' not in args and self.max_priority is not
…[8629 chars omitted]…
eject_on_worker_lost=Option(type='bool'),
318:        direct=Option(False, type='bool', old={'celery_worker_direct'}),
331:        lost_wait=Option(10.0, type='float', old={'celeryd_worker_lost_wait'}),
celery/worker/consumer/tasks.py:49:        c.qos = QoS(set_prefetch_count, c.initial_prefetch_count)
celery/worker/consumer/tasks.py:49:        c.qos = QoS(set_prefetch_count, c.initial_prefetch_count)

requirements/test-ci-default.txt:21:git+https://github.com/celery/kombu.git
requirements/test-integration.txt:6:git+https://github.com/celery/kombu.git
re
…[4120 chars omitted]…
r',
                               PREFETCH_COUNT_MAX)
                new_value = 0
            logger.debug('basic.qos: prefetch_count->%s', new_value)
            self.callback(prefetch_count=new_value)
            self.prev = pcount
        return pcount

    def update(self):
        """Update prefetch count with current value."""
        with self._mutex:
            return self.set(self.value)

    def on_ready(self):
        callback, self.init_callback = self.init_callback, None
        if callback:
            callback(self)

    def loop_args(self):
        return (self, self.connection, self.task_consumer,
                self.blueprint, self.hub, self.qos, self.amqheartbeat,
                self.app.clock, self.amqheartbeat_rate)
180:    def __init__(self, on_task_request,

    dispatcher_cls = 'celery.events.dispatcher:EventDispatcher'
    state_cls = 'celery.events.state:State'

    def __init__(self, app=None):
        self.app = app

    @cached_property
    def Receiver(self):
        return self.app.subclass_with_self(
            self.receiver_cls, reverse='events.Receiver')

    @cached_property
    def Dispatcher(self):
        return self.app.subclass_with_self(
            self.dispatcher_cls, reverse='events.Dispatcher')

    @cached_property
    def State(self):
        return self.app.subclass_with_self(
           
…[8576 chars omitted]…
time_limit=Option(
298-            type='float', old={'celeryd_task_soft_time_limit'},
299-        ),
300-        time_limit=Option(
301-            type='float', old={'celeryd_task_time_limit'},
302-        ),
303-        store_errors_even_if_ignored=Option(False, type='bool'),
304-        track_started=Option(False, type='bool'),
305-        allow_error_cb_on_chord_header=Option(False, type='bool'),

:`task_create_missing_queues` is enabled.
celery/worker/worker.py:54:enable the `task_create_missing_queues` setting.
t/unit/app/test_app.py:292:        with self.Celery(task_create_missing_queues=False) as app:
t/unit/app/test_app.py:293:            assert app.conf.task_create_missing_queues is False
celery/app/amqp.py:203:    queues_cls = Queues
celery/app/amqp.py:261:        return self.queues_cls(

 getattr(self.pool, 'is_green', False):
celery/worker/consumer/consumer.py:256:        limit = rate(getattr(type, 'rate_limit', None))
celery/worker/consumer/consumer.py:320:    def _limit_post_eta(self, request, bucket, tokens):
celery/worker/consumer/consumer.py:489:            if getattr(conn, 'alt', None) and interval == 0:
celery/worker/consumer/consumer.py:572:    def apply_eta_task(self, task):

        self.on_declared = on_declared

        # allows Queue('name', [binding(...), binding(...), ...])
        if isinstance(exchange, (list, tuple, set)):
            self.bindings |= set(exchange)
        if self.bindings:
            self.exchange = None

        # exclusive implies auto-delete.
        if self.exclusive:
            self.auto_delete = True
        self.maybe_bind(channel)

Exchange ''(direct)

        from celery.beat import EmbeddedService
        if w.pool_cls.__module__.endswith(('gevent', 'eventlet')):
            raise ImproperlyConfigured(ERR_B_GREEN)
        b = w.beat = EmbeddedService(w.app,
                                     schedule_filename=w.schedule_filename,
                                     scheduler_cls=w.scheduler)
        return b


class StateDB(bootsteps.Step):
    """Bootstep that sets up between-restart state database file."""

    def __init__(self, w, **kwargs):
        self.enabled = w.statedb
        w._persistence = None
        super().__init__(w, **kwargs)

    
…[8628 chars omitted]…
 soft shutdown will not be initiated
173:    during the cold shutdown. When using ETA tasks, it is recommended to enable the soft shutdown on idle.
1059:Dump of scheduled (ETA) tasks
1062:You can get a list of tasks waiting to be scheduled by using
1063::meth:`~celery.app.control.Inspect.scheduled`:
1067:    >>> i.scheduled()
1084:    These are tasks with an ETA/countdown argument, not periodic tasks.

t/unit/tasks/test_tasks.py:1428:class test_apply_task(TasksCase):
t/unit/tasks/test_tasks.py:1616:class test_apply_async(TasksCase):
t/unit/worker/test_consumer.p
…[3461 chars omitted]…
onsumer.qos, 10)
t/unit/worker/test_worker.py:362:        assert c.task_consumer.consume.call_count
t/unit/worker/test_worker.py:363:        c.task_consumer.qos.assert_called_with(prefetch_count=10)
t/unit/worker/test_worker.py:369:        c.task_consumer.qos.assert_called_with(prefetch_count=9)
t/unit/worker/test_worker.py:401:        c.qos = QoS(c.task_consumer.qos, 1)
t/unit/worker/test_consumer.py

from celery.exceptions import WorkerShutdown, WorkerTerminate
from celery.utils.collections import LimitedSet
from celery.utils.quorum_queues import detect_quorum_queues
from celery.worker.consumer.agent import Agent
from celery.worker.consumer.consumer import CANCEL_TASKS_BY_DEFAULT, CLOSE, TERMINATE, Consumer
from celery.worker.consumer.gossip import Gossip
from celery.worker.consumer.heart import Heart
from celery.worker.consumer.mingle import Mingle
from celery.worker.consumer.tasks import Tasks
from celery.worker.state import active_requests


class ConsumerTestCase:
    def get
…[8629 chars omitted]…
ller(concurrency=1, loglevel=0, beat=True)
t/unit/worker/test_worker.py:911:        worker = self.app.WorkController(concurrency=1, loglevel=0)
t/unit/worker/test_worker.py:928:        worker = self.app.WorkController(concurrency=1, loglevel=0)
t/unit/worker/test_worker.py:938:        worker = self.app.WorkController(concurrency=1, loglevel=10)
208:class test_Control_broadcast:
232:class test_Control:

    def test_broadcast(self):
        self.app.control.broadcast('foobarbaz', arguments={'foo': 2})
        self.app.control.mailbox.assert_called()
        self.app.control.mailbox()._broadcast.assert_called_with(
            'foobarbaz', {'foo': 2}, None, False, 1.0, None, None,
            channel=None,
        )

    def test_broadcast_limit(self):
        self.app.control.broadcast(
            'foobarbaz1', arguments=None, limit=None, destination=[1, 2, 3],
        )
        self.app.control.mailbox.assert_called()
        self.app.control.mailbox()._broadca
…[5025 chars omitted]…
failover_strategy=failover_strategy,
            **kwargs
        )

    def _connection(self, url, userid=None, password=None,
                    virtual_host=None, port=None, ssl=None,
                    connect_timeout=None, transport=None,
                    transport_options=None, heartbeat=None,
                    login_method=None, failover_strategy=None, **kwargs):
        conf = self.conf

    def setup_method(self):
        self.c = Mock()
        self.c.app.conf.worker_detect_quorum_queues = True
        self.c.connection.qos_semantics_matches_spec = False

    def test_stop(self):
        c = self.c
        tasks = Tasks(c)
        assert c.task_consumer is None
        assert c.qos is None

        c.task_consumer = Mock()
        tasks.stop(c)

    def test_stop_already_stopped(self):
        c = self.c
        tasks = Tasks(c)
        tasks.stop(c)

    def test_detect_quorum_queues_positive(self):
        c = self.c
        self.c.connection.tran
…[5541 chars omitted]…
b19-413e-91d2-0b74f3844c4d',
            'args': '[1]',
            'kwargs': '{}'}},
         {'eta': '2010-06-07 09:07:53', 'priority': 0,
          'request': {
            'name': 'tasks.sleeptask',
            'id': '49661b9a-aa22-4120-94b7-9ee8031d219d',
            'args': '[2]',
            'kwargs': '{}'}}]}]

.. note::

    These are tasks with an ETA/countdown argument, not periodic tasks.

218-class QueueNotFound(KeyError, TaskError):
677:    def config_from_object(self, obj,
701:    def config_from_envvar(self, variable_name, silent=False, force=False):
719:    def config_from_cmdline(self, argv, namespace='celery'):
1456:    def conf(self):
1463:    def conf(self, d):
70:class Settings(ConfigurationView):

#: Note: Use :func:`get_exchange` instead, as the type of
#: exchange will vary depending on the broker connection.
event_exchange = Exchange(EVENT_EXCHANGE_NAME, type='topic')


def Event(type, _fields=None, __dict__=dict, __now__=time.time, **fields):
    """Create an event.

    Notes:
        An event is simply a dictionary: the only required field is ``type``.
        A ``timestamp`` field will be set to the current time if not provided.
    """
    event = __dict__(_fields, **fields) if _fields else fields
    if 'timestamp' not in event:
        event.update(timestamp=__now__(), type=type)
    else:
        event['type']
…[8628 chars omitted]…
eps.StartStopStep):
                 prefetch_count=prefetch_count,
                 apply_global=qos_global,
             )
-        c.qos = QoS(set_prefetch_count, c.initial_prefetch_count)
+        c.qos = QoS(
+            set_prefetch_count, c.initial_prefetch_count,
+            max_prefetch=c.app.conf.worker_eta_task_limit,
+        )
 
     def stop(self, c):
         """Stop task consumer."""

guration setting will no longer determine
  whether broker connection retries are made during startup in Celery 6.0 and above.
  If you wish to refrain from retrying connections on startup,
  you should set broker_connection_retry_on_startup to False instead.
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
237 passed, 18 warnings, 37 subtests passed in 3.95s

g:`worker_prefetch_multiplier`
docs/userguide/configuration.rst:3119:.. setting:: worker_prefetch_multiplier
docs/userguide/configuration.rst:3121:``worker_prefetch_multiplier``
docs/userguide/configuration.rst:3134:To disable prefetching, set :setting:`worker_prefetch_multiplier` to 1.
docs/userguide/workers.rst:301:tasks that are currently running multiplied by :setting:`worker_prefetch_multiplier`.

Or change all of the settings to use the new format :)

"""

FMT_REPLACE_SETTING = '{replace:<36} -> {with_}'


def appstr(app):
    """String used in __repr__ etc, to id app instances."""
    return f'{app.main or "__main__"} at {id(app):#x}'


class Settings(ConfigurationView):
    """Celery settings object.

    .. seealso:

        :ref:`configuration` for a full list of configuration keys.

    """

    def __init__(self, *args, deprecated_settings=None, **kwargs):
        super().__init__(*args, **kwargs)

        self.deprecated_settings = deprecated_settings

    @property
    def broker_read_url(self):
        return (
            os.environ.get('CELERY_BROKER_READ_URL') or
…[8656 chars omitted]…
/__pycache__/defaults.cpython-312.pyc: binary file matches
celery/app/defaults.py:58:    def to_python(self, value):
worker_eta_task_limit None <class 'NoneType'>
task_create_missing_queue_type 'classic' <class 'str'>
task_create_missing_queue_exchange_type 'direct' <class 'str'>
control_queue_durable False <class 'bool'>
event_queue_exclusive False <class 'bool'>
worker eta str 0 => '0' <class 'str'>

    def teardown_method(self):
        if self._prev:
            sys.modules['celery.app.defaults'] = self._prev

    def test_option_repr(self):
        assert repr(NAMESPACES['broker']['url'])

    def test_any(self):
        val = object()
        assert self.defaults.Option.typemap['any'](val) is val

    def test_compat_indices(self):
        assert not any(key.isupper() for key in DEFAULTS)
        assert not any(key.islower() for key in _OLD_DEFAULTS)
        assert not any(key.isupper() for key in _TO_OLD_KEY)
        assert not any(key.islower() for ke
…[4853 chars omitted]…
():
            if ns.lower() == name.lower():
                return searchresult(None, ns, opts)
            elif isinstance(opts, dict):
                try:
                    return searchresult(ns, name.lower(), opts[name.lower()])
                except KeyError:
                    pass
    # - See if name is a qualname last.
    return searchresult(None, name.lower(), DEFAULTS[name.lower()])

    @cached_property
    def router(self):
        return self.Router()

    @router.setter
    def router(self, value):
        return value

    @property
    def producer_pool(self):
        if self._producer_pool is None:
            self._producer_pool = pools.producers[
                self.app.connection_for_write()]

The number of concurrent worker processes/threads/green threads executing
tasks.

If you're doing mostly I/O you can have more processes,
but if mostly CPU-bound, try to keep it close to the
number of CPUs on your machine. If not set, the number of CPUs
…[6906 chars omitted]…
                  :setting:`cache_backend`
``CELERY_CACHE_BACKEND_OPTIONS``           :setting:`cache_backend_options`
``CASSANDRA_COLUMN_FAMILY``                :setting:`cassandra_table`
``CASSANDRA_ENTRY_TTL``                    :setting:`cassandra_entry_ttl`
``CASSANDRA_KEYSPACE``                     :setting:`cassandra_keyspace`
``CASSANDRA_PORT``                         :setting:`cassandra_port`

        return self._rtable

    @cached_property
    def router(self):
        return self.Router()

    @router.setter
    def router(self, value):
        return value

    @property
    def producer_pool(self):
        if self._producer_pool is None:
            self._producer_pool = pools.producers[
                self.app.connection_for_write()]
            self._producer_pool.limit = self.app.pool.limit
        return self._producer_pool
    publisher_pool = producer_pool  # compat alias

    @cached_property
382:class Queue(MaybeChannelBound):

     def __init__(self, queues=None, default_exchange=None,
                  create_missing=True, autoexchange=None,
-                 max_priority=None, default_routing_key=None):
+                 max_priority=None, default_routing_key=None,
+                 create_missing_queue_type='classic',
+                 create_missing_queue_exchange_type='direct'):
         super().__init__()
         self.aliases = WeakVal
…[7735 chars omitted]…
_count=prefetch_count,
                 apply_global=qos_global,
             )
-        c.qos = QoS(set_prefetch_count, c.initial_prefetch_count)
+        c.qos = QoS(
+            set_prefetch_count, c.initial_prefetch_count,
+            max_prefetch=c.app.conf.worker_eta_task_limit,
+        )
 
     def stop(self, c):
         """Stop task consumer."""
direct AMQP construction and ETA limits work

3362:Events
3737:.. admonition:: Eventlet/Gevent
163:``CELERY_ENABLE_REMOTE_CONTROL``           :setting:`worker_enable_remote_control`
1903:Allows to control the number of concurrent operations. For example::
3058:to import signal handlers and additional remote control commands, etc.
3175:be useful in scenarios where a fixed prefetch count is desired to co
…[631 chars omitted]…
es to remote control reply queues.
3542:.. setting:: control_queue_expires
3544:``control_queue_expires``
3549:Time in seconds, before an unused remote control command queue is deleted
3552:This setting also applies to remote control reply queues.
3554:.. setting:: control_exchange
3556:``control_exchange``
3561:Name of the control command exchange.
3752::control:`pool_restart` remote control command.

Send task-related events so that tasks can be monitored using tools like
`flower`. Sets the default value for the workers
:option:`-E <celery worker -E>` argument.

.. setting:: task_send_sent_event

``task_send_sent_event``
~~~~~~~~~~~~~~~~~~~~~~~~

.. versionadded:: 2.2

Default: Disabled by default.

If enabled, a :event:`task-sent` event will be sent for every task so tasks can be
tracked before they're consumed by a worker.

.. setting:: event_queue_ttl

``event_queue_ttl``
~~~~~~~~~~~~~~~~~~~
:transports supported: ``amqp``

Default: 5.0 seconds.

Message expiry time in seconds (int/float) for when messages sent to a monitor clients
event queue is deleted (``x-message-ttl``)

For example, if this value is set to 10 then a 
…[2096 chars omitted]…
ommand exchange.

.. warning::

    This option is in experimental stage, please use it with caution.

.. _conf-logging:

Logging
-------

.. setting:: worker_hijack_root_logger

``worker_hijack_root_logger``
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
136:``CELERY_CREATE_MISSING_QUEUES``           :setting:`task_create_missing_queues`
2611:.. setting:: task_create_missing_queues
2613:``task_create_missing_queues``

    # Missing docstring in public method
    D102,
    # Missing docstring in public package
    D104,
    # Missing docstring in magic method
    D105,
    # Missing docstring in __init__
    D107,
    # First line should be in imperative mood; try rephrasing
    D401,
    # No blank lines allowed between a section header and its content
    D412,
    # ambiguous variable name '...'
    E741,
    # ambiguous class definition '...'
    E742,
per-file-ignores =
   t/*,setup.py,examples/*,docs/*,extra/*:
       # docstrings
       D,

[bdist_rpm]
requires = backports.zoneinfo>=0.2.1;python_version<'3.9'
           tzdata>=2022.7
           billiard >=4.1.0,<5.0
           kombu >= 5.3.4,<6.0.0


[metadata]
license_files = LICENSE
tox.ini:9:    flake8
setup.cfg:6:[flake8]

ImportError while importing test module '/repo/t/unit/backends/test_azureblockblob.py'.
Hint: make sure your test modules/packages have valid Pyth
…[8495 chars omitted]…
nit/backends/test_s3.py
ERROR t/unit/security/test_certificate.py
ERROR t/unit/security/test_key.py
ERROR t/unit/security/test_security.py
ERROR t/unit/security/test_serialization.py
ERROR t/unit/tasks/test_canvas.py
ERROR t/unit/utils/test_annotations.py
ERROR t/unit/utils/test_functional.py
!!!!!!!!!!!!!!!!!!! Interrupted: 12 errors during collection !!!!!!!!!!!!!!!!!!!
9 skipped, 12 errors in 8.25s

 M celery/worker/consumer/tasks.py
 M docs/userguide/configuration.rst

from celery import signals
from celery.utils.nodenames import anon_nodename
from celery.utils.saferepr import saferepr
from celery.utils.text import indent as textindent
from celery.utils.time import maybe_make_aware
__all__ = ('AMQP', 'Queues', 'task_message')

#: earliest date supported by time.mktime.
INT_MIN = -2147483648

#: Human readable queue declaration.
QUEUE_FORMAT = """
.> {0.name:<16} exchange={0.exchange.name}({0.exchange.type}) \
key={0.rout
…[8629 chars omitted]…
aemonizing (defaults to :data:`sys.executable`).


.. _conf-control:

Remote Control Commands
-----------------------

.. note::

    To disable remote control commands see
    the :setting:`worker_enable_remote_control` setting.

.. setting:: control_queue_ttl

``control_queue_ttl``
~~~~~~~~~~~~~~~~~~~~~

Default: 300.0

Time in seconds, before a message in a remote control command queue
will expire.

        Arguments:
            exclude (Sequence[str], str): Names of queues to avoid
                consuming from.
        """
        if exclude:
            exclude = maybe_list(exclude)
            if self._consume_from is None:
                # using all queues
                return self.select(k for k in self if k not in exclude)
            # using selection
            for queue in exclude:
                self._consume_from.pop(queue, None)

 
…[4559 chars omitted]…

Make remote control mailbox queues exclusive. Exclusive mailbox queues are
non-durable and are automatically deleted.

.. setting:: control_queue_expires

``control_queue_expires``
~~~~~~~~~~~~~~~~~~~~~~~~~

Default: 10.0

Time in seconds, before an unused remote control command queue is deleted
from the broker.

This setting also applies to remote control reply queues.

.. setting:: control_exchange

``task_create_missing_queue_type``
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Default: ``"classic"``.

Queue type to use for automatically created task queues. For RabbitMQ,
set this to ``"quorum"`` to create quorum queues.

.. setting:: task_create_missing_queue_exchange_type

``task_create_missing_queue_exchange_type``
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Default: ``"direct"``.

Exchange type to use for automatically created task queues.

.. setting:: task_default_queue

``task_default_queue``
~~~~~~~~~~~~~~~~~~~~~~

Default: ``"celery"``.

The name of the default qu
…[2787 chars omitted]…
ble. A durable event queue cannot be
configured as exclusive.

.. setting:: event_queue_exclusive

``event_queue_exclusive``
~~~~~~~~~~~~~~~~~~~~~~~~

Default: Disabled.

"""Sending/Receiving Messages (Kombu integration)."""
import numbers
from copy import copy
from collections import namedtuple
from collections.abc import Mapping
from datetime import timedelta
from weakref import WeakValueDictionary

+
+Default: ``"classic"``.
+
+Queue type to use for automatically created task queues. For RabbitMQ,
+set this to ``"quorum"`` to create quorum queues.
+
+.. setting:: task_create_missing_queue_exchange_type
+
+``task_create_missing_queue_exchange_type``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Default: ``"direct"``.
+
+Exchange type to use for automatically created task queues.
+
 .. setting:: task_d
…[8629 chars omitted]…
eps.StartStopStep):
                 prefetch_count=prefetch_count,
                 apply_global=qos_global,
             )
-        c.qos = QoS(set_prefetch_count, c.initial_prefetch_count)
+        c.qos = QoS(
+            set_prefetch_count, c.initial_prefetch_count,
+            max_prefetch=c.app.conf.worker_eta_task_limit,
+        )
 
     def stop(self, c):
         """Stop task consumer."""

Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
  File "/usr/local/lib/python3.12/inspect.py", line 1285, in getsource
    lines, lnum = getsourcelines(object)
                  ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/inspect.py", line 1267, in getsourcelines
    lines, lnum = findsource(object)
                  ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/inspect.py", line 1078, in findsource
    file = getsour
…[35 chars omitted]…
^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/inspect.py", line 955, in getsourcefile
    filename = getfile(object)
               ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/inspect.py", line 935, in getfile
    raise TypeError('module, class, method, function, traceback, frame, or '
TypeError: module, class, method, function, traceback, frame, or code object was expected, got cached_property

from celery.exceptions import QueueNotFound
from celery.utils.collections import lpmerge
from celery.utils.functional import maybe_evaluate, mlazy
from celery.utils.imports import symbol_by_name

try:
    Pattern = re._pattern_type
"""Task implementation: request context and the task base class."""
import sys

from billiard.einfo import ExceptionInfo, ExceptionWithTraceback
from kombu import serialization
from kombu.exceptions import OperationalError
from kombu.utils.uuid import uuid

from celery import current_app, states
from celery._state import _task_stack
from celery.canvas 
…[69 chars omitted]…
 ImproperlyConfigured, MaxRetriesExceededError, Reject, Retry
from celery.local import class_property
from celery.result import EagerResult, denied_join_result
from celery.utils import abstract
from celery.utils.functional import mattrgetter, maybe_list
from celery.utils.imports import instantiate
from celery.utils.nodenames import gethostname
from celery.utils.serialization import raise_with_context

    """

    app = None

    def __init__(self, channel, handlers=None, routing_key='#',
                 node_id=None, app=None, queue_prefix=None,
                 accept=None, queue_ttl=None, queue_expires=None):
        self.app = app_or_default(app or self.app)
        self.channel = maybe_channel(channel)
        self.handlers = {} if handlers is None else handlers
        self.routing_key = routing_key
        self.node_id = node_id or uuid()
       
…[773 chars omitted]…
            exchange=self.exchange,
            routing_key=self.routing_key,
            auto_delete=not queue_durable,
            durable=queue_durable,
            exclusive=queue_exclusive,
            message_ttl=queue_ttl,
            expires=queue_expires,
        )
        self.clock = self.app.clock
        self.adjust_clock = self.clock.adjust
        self.forward_clock = self.clock.forward

2634 '``task_create_missing_queue_exchange_type``' 43 44
3456 '``event_queue_exclusive``' 25 24
3475 '``event_exchange``' 18 22
3627 '``control_exchange``' 20 22
3783 '``security_digest``' 19 24

            self._backend = self._get_backend()
        return self._backend

    @property
    def conf(self):
        """Current configuration."""
        if self._conf is None:
            self._conf = self._load_config()
        return self._conf

    @conf.setter
    def conf(self, d):
        self._conf = d

    @cached_property
    def control(self):
        """Remote control: :class:`~@control`."""
        return instantiate(self.control_cls, app=self)

    @cached_property
    def events(self):
        """Consuming and sending events: :cl
…[2089 chars omitted]…
       self.main = main
329-        self.amqp_cls = amqp or self.amqp_cls
330-        self.events_cls = events or self.events_cls
331-        self.loader_cls = loader or self._get_default_loader()
332-        self.log_cls = log or self.log_cls
333-        self.control_cls = control or self.control_cls
334-        self._custom_task_cls_used = (
custom autoexchange is preserved and exchange kind applied

    #: If set, this is a subset of queues to consume from.
    #: The rest of the queues are then used for routing only.
    _consume_from = None

    def __init__(self, queues=None, default_exchange=None,
                 create_missing=True, autoexchange=N
…[8304 chars omitted]…
   apply_global=qos_global,
            )
        c.qos = QoS(
            set_prefetch_count, c.initial_prefetch_count,
            max_prefetch=c.app.conf.worker_eta_task_limit,
        )

    def stop(self, c):
        """Stop task consumer."""
        if c.task_consumer:
            debug('Canceling task consumer...')
            ignore_errors(c, c.task_consumer.cancel)

    def shutdown(self, c):

        """`self(channel) -> self.bind(channel)`."""
        return self.bind(channel)

    def bind(
        self: _MaybeChannelBoundType, channel: (Channel | Connection)
    ) -> _MaybeChannelBoundType:
        """Create copy of the instance that is bound to a channel."""
        return copy(self).maybe_bind(channel)

    def maybe_bind(
        self: _MaybeChannelBoundType, channel: (Channel | Connection)
    ) -> _MaybeChannelBoundType:
        """Bind instance to channel if not already bound."""
 
…[1087 chars omitted]…
nnel(self) -> Channel:
        """Current channel if the object is bound."""
        channel = self._channel
        if channel is None:
            raise NotBoundError(
                "Can't call method on {} not bound to a channel".format(
                    type(self).__name__))
        if isinstance(channel, ChannelPromise):
            channel = self._channel = channel()
        return channel

            on_ack=ack, on_reject=reject, app=app, hostname=hostname,
            eventer=eventer, task=task, connection_errors=connection_errors,
            body=body, headers=headers, decoded=decoded, utc=utc,
        )
        if _does_info:
            # Similar to `app.trace.info()`, we pass the formatting args as the
            # `extra` kwarg for custom log handlers
            context = {
             
…[8629 chars omitted]…
r',
                               PREFETCH_COUNT_MAX)
                new_value = 0
            logger.debug('basic.qos: prefetch_count->%s', new_value)
            self.callback(prefetch_count=new_value)
            self.prev = pcount
        return pcount

    def update(self):
        """Update prefetch count with current value."""
        with self._mutex:
            return self.set(self.value)

        self: _MaybeChannelBoundType, channel: (Channel | Connection)
    ) -> _MaybeChannelBoundType:
        """`self(channel) -> self.bind(channel)`."""
        return self.bind(channel)

    def bind(
        self: _MaybeChannelBoundType, channel: (Channel | Connection)
    ) -> _MaybeChannelBoundType:
        """Create copy of the instance that is bound to a channel."""
        return copy(sel
…[1857 chars omitted]…
urse and isinstance(obj, Object):
                return obj.as_dict(recurse=True)
            return type(obj) if type and obj is not None else obj
        return {
            attr: f(getattr(self, attr), type) for attr, type in self.attrs
        }

{'name': 'x', 'type': 'fanout', 'arguments': None, 'durable': True, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}

    _rtable = None

    #: Underlying producer pool instance automatically
    #: set by the :attr:`producer_pool`.
    _producer_pool = None

    # Exchange class/function used when defining automatic queues.
    # For example, you can use ``autoexchange = lambda n: None`` to use the
    # AMQP default exchange: a shortcut to bypass routing
    # and instead send directly to the queue named in the routing key.
    autoexchange = None

    #: Max size of positional argume
…[1096 chars omitted]…
tcher(enabled=False)

    def _handle_conf_update(self, *args, **kwargs):
        if ('task_routes' in kwargs or 'task_routes' in args):
            self.flush_routes()
            self.router = self.Router()
        return
amqp autoexchange attr None <class 'NoneType'>
default_exchange Exchange celery(direct)
queues autoexchange <class 'kombu.entity.Exchange'>
autoexchange result Exchange foo(direct)

698:            if self.load
…[4347 chars omitted]…
 {} if not kwargs else kwargs
        sig = (sig.clone(args, kwargs)
               if isinstance(sig, abstract.CallableSignature)
               else self.signature(sig.name, args, kwargs))
        return name or repr(sig), {
            'schedule': schedule,
            'task': sig.name,
            'args': sig.args,
            'kwargs': sig.kwargs,
            'options': dict(sig.options, **opts),

            # ## find name-space.
            # .key=value|_key=value expands to default name-space.
            if key[0] == '_':
                ns, key = namespace, key[1:]
            else:
                # find name-space part of key
                ns, key = key.split('_', 1)

            ns_key = (ns and ns + '_' or '') + key

            # (type)value makes cast to custom type.
            cast = re_type.match(value)
            if cast
…[3115 chars omitted]…
pace,
        )
        if self._conf is not None:
            # replace in place, as someone may have referenced app.conf,
            # done some changes, accessed a key, and then try to make more
            # changes to the reference and not the finalized value.
            self._conf.swap_with(settings)
        else:
            self._conf = settings

        # load lazy config dict initializers.

guration setting will no longer determine
  whether broker connection retries are made during startup in Celery 6.0 and above.
  If you wish to refrain from retrying connections on startup,
  you should set broker_connection_retry_on_startup to False instead.
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
241 passed, 18 warnings, 37 subtests passed in 3.79s

                obj = self._smart_import(obj, imp=self.import_from_cwd)
            except (ImportError, AttributeError):
                if silent:
                    return False
                raise
        self._conf = force_mapping(obj)
        if self._conf.get('override_backends') is not None:
            self.override_backends = self._conf['override_backends']
        return True

    def _smart_import(self, path, imp=None):
        imp = self.import_module if imp is None else imp
        if ':' in path:
            # Path includes attribute so can just jump
  
…[2239 chars omitted]…
it
docs/userguide/configuration.rst:3166:``worker_eta_task_limit``
celery/app/amqp.py:280:                conf.task_create_missing_queue_exchange_type
celery/worker/consumer/tasks.py:51:            max_prefetch=c.app.conf.worker_eta_task_limit,
295:        queue_max_priority=Option(None, type='int'),
322:        concurrency=Option(None, type='int'),
323:        eta_task_limit=Option(None, type='int'),

@@ -47,6 +48,11 @@ class Queues(dict):
             added automatically, but if this flag is disabled the occurrence
             of unknown queues in `wanted` will raise :exc:`KeyError`.
         max_priority (int): Default x-max-priority for queues with none set.
+        create_missing_queue_type (str): Queue type for automatically created
+            queues. Defaults to ``classic``.
+        create_missing_queue_exchange_type (str): Exchange type for
+            automatically created queues. Defaults to the autoexchange type
+            (normally `
…[8629 chars omitted]…
te control mailbox queues durable. Durable mailbox queues cannot be
+configured as exclusive.
+
+.. setting:: control_queue_exclusive
+
+``control_queue_exclusive``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Default: Disabled.
+
+Make remote control mailbox queues exclusive. Exclusive mailbox queues are
+non-durable and are automatically deleted.
+
 .. setting:: control_queue_expires
 
 ``control_queue_expires``

underline mismatch 3475 '``event_exchange``' 18 22
underline mismatch 3627 '``control_exchange``' 20 22
underline mismatch 3783 '``security_digest``' 19 24