celery-celery-9815-9863

celery/__init__.py
celery/__main__.py
celery/_state.py
celery/app/__init__.py
celery/app/amqp.py
celery/app/annotations.py
celery/app/autoretry.py
celery/app/backends.py
celery/app/base.py
celery/app/builtins.py
celery/app/control.py
celery/app/defaults.py

celery/app/events.py
celery/app/log.py
celery/app/registry.py
celery/app/routes.py
celery/app/task.py
celery/app/trace.py
celery/app/utils.py
celery/apps/__init__.py
celery/apps/beat.py
celery/apps/multi.py
celery/apps/worker.py
celery/backends/__init__.py
celery/backends/arangodb.py
celery/backends/asynchronous.py
celery/backends/azureblockblob.py
celery/backends/base.py
celery/backends/cache.py
celery/backends/cassandra.py
celery/backends/consul.py
celery/backends/cosmosdbsql.py
celery/backends/couchbase.py
celery/backends/couchdb.py
celery/backends/dynamodb.py
celery/backends/elasticsearch.py
celery/backends/filesystem.py
celery/
…[1078 chars omitted]…
elery/events/snapshot.py
celery/events/state.py
celery/exceptions.py
celery/fixups/__init__.py
celery/fixups/django.py
celery/loaders/__init__.py
celery/loaders/app.py
celery/loaders/base.py
celery/loaders/default.py
celery/local.py
celery/platforms.py
celery/result.py
celery/schedules.py
celery/security/__init__.py
celery/security/certificate.py
celery/security/key.py
celery/security/serialization.py

"""Worker Task Consumer Bootstep."""

from __future__ import annotations

from kombu.common import QoS, ignore_errors

from celery import bootsteps
from celery.utils.log import get_logger
from celery.utils.quorum_queues import detect_quorum_queues

from .mingle import Mingle

__all__ = ('Tasks',)


logger = get_logger(__name__)
debug = logger.debug


class Tasks(bootsteps.StartStopStep):
    """Bootstep starting the task message consumer."""

    requires = (Mingle,)

    def __init__(self, c, **kwargs):
        c.task_consumer = c.qos = None
        super().__init__(c, **kwargs)

    def start(self, c):
        """Start task consumer."""
        c.update_strategies()

        qos_global = self.qos_global(c)

        # set initial prefetch count
        c.connection.default_channel.basic_qos(
            0, c.initial_prefetch_count, qos_global,
        )

        c.task_consumer = c.a
…[8629 chars omitted]…
elf._pending_operations.pop()()
                except Exception as exc:  # pylint: disable=broad-except
                    logger.exception('Pending callback raised: %r', exc)

    def bucket_for_task(self, type):
        limit = rate(getattr(type, 'rate_limit', None))
        return TokenBucket(limit, capacity=1) if limit else None

    def reset_rate_limits(self):
        self.task_buckets.update(

"""Internal worker state (global).

This includes the currently active and reserved tasks,
statistics, and revoked tasks.
"""
import os
import platform
import shelve
import sys
import weakref
import zlib
from collections import Counter

from kombu.serialization import pickle, pickle_protocol
from kombu.utils.objects import cached_property

from celery import __version__
from celery.exceptions import WorkerShutdown, WorkerTerminate
from celery.utils.collections import LimitedSet

__all__ = (
    'SOFTWARE_INFO', 'reserved_requests', 'active_requests',
    'total_count', 'revoked', 'task_reserved', 'maybe_shutdown',
    'task_accepted', 'task_ready', 'Persistent',
)

#: Worker software/platform information.
SOFTWARE_INFO = {
    'sw_ident': 'py-celery',
    'sw_ver': __version__,
    'sw_sys': platform.system(),
}

#: maximum number of revokes to keep in memory.
REVOKES_MAX = int(os.env
…[2146 chars omitted]…
t:136:``CELERY_CREATE_MISSING_QUEUES``           :setting:`task_create_missing_queues`
docs/userguide/configuration.rst:2611:.. setting:: task_create_missing_queues
docs/userguide/configuration.rst:2613:``task_create_missing_queues``
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).

celery/worker/autoscale.py:45:            w.pool, w.max_concurrency, w.min_concurrency,
celery/worker/autoscale.py:64:    def __init__(self, pool, max_concurrency,
celery/worker/autoscale.py:70:        self.max_concurrency = max_concurrency
celery/worker/autoscale.py:85:        cur = min(self.qty, self.max_concurrency)
celery/worker/autoscale.py:104:                self.max_concurrency = max
celery/worker/autoscale.py:109:            return self.max_concurrency, self.min_concurrency
celery/worker/autoscale.py:134:        diff = new_max - self.max_concurrency
celery/worker/autoscale.py:142:            'max': self.max_concurrency,
celery/worker/components.py:108:        * max_concurrency
celery/worker/components.py:116:        w.max_concurrency = None
celery/worker/components.py:124:            w.max_concurrency, w.min_concurrency = w.autoscale
celery/worker/components.py:221:        if

…[194 chars omitted]…
cesses
celery/worker/consumer/consumer.py:275:        num_processes = self.pool.num_processes
celery/worker/consumer/consumer.py:276:        if not self.initial_prefetch_count or not num_processes:
celery/worker/consumer/consumer.py:279:            self.pool.num_processes * self.prefetch_multiplier
celery/worker/consumer/consumer.py:724:        return self.pool.num_processes * self.prefetch_multiplier

celery/events/receiver.py:51:        self.queue = Queue(
"""Worker Pidbox (remote control)."""
import socket
import threading

from kombu.common import ignore_errors
from kombu.utils.encoding import safe_str

from celery.utils.collections import AttributeDict
from celery.utils.functional import pass1
from celery.utils.log import get_logger

from . import control

__all__ = ('Pidbox', 'gPidbox')

logger = get_logger(__name__)
debug, error, info = logger.debug, logger.error, logger.info


class Pidbox:
    """Worker mailbox."""

    consumer = None

    def __init__(self, c):
        self.c = c
        self.hostname = c.hostname
        self.node = c.app.control.mailbox.Node(
            safe_str(c.hostname),
            handlers=control.Panel.data,
            state=AttributeDict(
                app=c.app,
                hostname=c.hostname,
                consumer=c,
              
…[2385 chars omitted]…
                while not shutdown.is_set() and c.connection:
                    if resets[0] < self._resets:
                        resets[0] += 1
                        self._do_reset(c, connection)
                    try:
                        connection.drain_events(timeout=1.0)
                    except socket.timeout:
                        pass
        finally:
            stopped.set()

t/unit/app:
__init__.py
test_amqp.py
test_annotations.py
test_app.py
test_backends.py
test_beat.py
test_builtins.py
test_celery.py
test_control.py
test_defaults.py
test_exceptions.py

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)
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/events/test_events.py:217:    def test_event_queue_prefix__default(self):
t/unit/events/test_events.py:221:    def test_event_queue_prefix__setting(self):

t/unit/events/test_events.py:222:        self.app.conf.event_queue_prefix = 'eventq'
t/unit/events/test_events.py:226:    def test_event_queue_prefix__argument(self):

class Node:
    """Mailbox node."""

    #: hostname of the node.
    hostname = None

    #: the :class:`Mailbox` this is a node for.
    mailbox = None

    #: map of method name/handlers.
    handlers = None

    #: current context (passed on to handlers)
    state = None

    #: current channel.
    channel = None

    def __init__(self, hostname, state=None, channel=None,
                 handlers=None, mailbox=None):
        self.channel = channel
        self.mailbox = mailbox
        self.hostname = hostname
        self.state = state
        self.adjust_clock = self.mailbox.clock.adjust
        if handlers is None:
            handlers = {}
        self.handlers = handlers

    def Consumer(self, channel=None, no_ack=True, accept=None, **options):
        queue = self.mailbox.get_queue(self.hostname)

        def verify_exclusive(name, messages, consumers):
            if con
…[2547 chars omitted]…
   run_dispatch = True
        if run_dispatch:
            return self.dispatch(**body)
    dispatch_from_message = handle_message

    def reply(self, data, exchange, routing_key, ticket, **kwargs):
        self.mailbox._publish_reply(data, exchange, routing_key, ticket,
                                    channel=self.channel,
                                    serializer=self.mailbox.serializer)

86:        c._update_qos_eventually = Mock(name='update_qos')
94:        c._update_qos_eventually.assert_called_with(8)
117:        c.qos = Mock()
143:        c.qos = Mock()
153:        c.qos._mutex = MutexMock()
159:        c.qos = MagicMock()
160:        c.qos.value = 1
209:        bucket.can_consume.return_value = True
217:            bucket.can_consume.assert_called_with(3)
220:        bucket.can_consume.return_value = False
227:        bucket.can_consume.assert_called_with(4)
236:        bucket.can_consume.reset_mock()

238:        bucket.can_consume.assert_not_called()
254:        c.qos = Mock()
262:            c.qos.decrement_eventually.assert_called_with()
511:    def test_start_raises_connection_error(self,
595:    def test_start(self):
612:    def test_start_heartbeat_interval(self):
630:class test_Tasks:
635:        self.c.connection.qos_semantics_matches_spec = False
641:  
…[160 chars omitted]…
qos_global_worker_detect_quorum_queues_true_no_quorum_queues(self):
684:        assert tasks.qos_global(c) is True
686:    def test_qos_global_worker_detect_quorum_queues_true_with_quorum_queues(self):
691:        assert tasks.qos_global(c) is False
693:    def test_log_when_qos_is_false(self, caplog):
712:    def test_start(self):
723:    def test_start_no_replies(self):
731:    def test_start(self):

14:    def test_accept_content(self, app):
27:    def test_setup_nolimit(self, app):
45:    def test_setup(self, app):
68:class test_Queues:
70:    def test_queues_format(self):
74:    def test_with_defaults(self):
77:    def test_add(self):
84:    def test_setitem_adds_default_exchange(self):
92:    def test_select_add(self):
98:    def test_deselect(self):
104:    def test_add_default_exchange(self):
110:    def test_alias(self):

132:    def test_with_max_priority(self, queues_kwargs, qname, q, expected):
147:    def test_setting_default_queue(self, name, exchange, rkey, default_queue_type):
173:    def test_setting_default_exchange(self, name, exchange, rkey):
187:    def test_setting_default_exchange_type(self, name, extype, rkey):
198:    def test_kwargs_must_be_mapping(self):
202:    def test_args_must_be_list(self):
206:    def test_countdown_negative(self):
210:    def test_as
…[4827 chars omitted]…
ssert queue.queue_arguments is None


class test_default_exchange:

    @pytest.mark.parametrize('name,exchange,rkey', [
        ('default', 'foo', None),
celery/app/amqp.py:41:class Queues(dict):
celery/app/amqp.py:241:    def Queues(self, queues, create_missing=None,
celery/app/amqp.py:573:        return self.Queues(self.app.conf.task_queues)
celery/app/amqp.py:577:        return self.Queues(queues)

104:``CELERY_EVENT_QUEUE_EXPIRES``             :setting:`event_queue_expires`
105:``CELERY_EVENT_QUEUE_TTL``                 :setting:`event_queue_ttl`
106:``CELERY_EVENT_QUEUE_PREFIX``              :setting:`event_queue_prefix`
136:``CELERY_CREATE_MISSING_QUEUES``           :setting:`task_create_missing_queues`
172:``CELERYD_PREFETCH_MULTIPLIER``            :setting:`worker_prefetch_multiplier`
173:``CELERYD_ENABLE_PREFETCH_COUNT_REDUCTION``:setting:`worker_enable_prefetch_count_reduction`
2611:.. setting:: task_create_missing_queues
2613:``task_create_missing_queues``
3119:.. setting:: worker_prefetch_multiplier
3121:``worker_prefetch_multiplier``
3126:How many messages to prefetch at a time multiplied by the number of
3134:To disable prefetching, set :setting:`worker_prefetch_multiplier` to 1.

3138:For more on prefetching, read :ref:`optimizing-prefetch-limit`
3142:    Tasks with E
…[1124 chars omitted]…
storation behavior of the prefetch count following a connection loss, aiding in
3388:.. setting:: event_queue_ttl
3390:``event_queue_ttl``
3402:.. setting:: event_queue_expires
3404:``event_queue_expires``
3413:.. setting:: event_queue_prefix
3415:``event_queue_prefix``
3526:.. setting:: control_queue_ttl
3528:``control_queue_ttl``
3542:.. setting:: control_queue_expires
3544:``control_queue_expires``

150:    control=Namespace(
151-        queue_ttl=Option(300.0, type='float'),
152-        queue_expires=Option(10.0, type='float'),
153-        exchange=Option('celery', type='string'),
154-    ),
155-    couchbase=Namespace(
156-        __old__=old_ns('celery_couchbase'),
157-
158-        backend_settings=Option(None, type='dict'),
159-    ),
160-    arangodb=Namespace(
161-        __old__=old_ns('celery_arangodb'),

162-        backend_settings=Option(None, type='dict')
163-    ),
164-    mongodb=Namespace(
165-        __old__=old_ns('celery_mongodb'),
166-
167-        backend_settings=Option(type='dict'),
168-    ),
169-    cosmosdbsql=Namespace(
170-        database_name=Option('celerydb', type='string'),
171-        collection_name=Option('celerycol', type='string'),
172-        consistency_level=Option('Session', type='string'),
173-        max_retry_attempts=Option(9, type='int'
…[2859 chars omitted]…
ool', old={'celery_worker_direct'}),
319-        disable_rate_limits=Option(
320-            False, type='bool', old={'celery_disable_rate_limits'},
321-        ),
322-        deduplicate_successful_tasks=Option(
323-            False, type='bool'
324-        ),
150:    control=Namespace(
325:        enable_remote_control=Option(
326:            True, type='bool', old={'celery_enable_remote_control'},

celery/exceptions.py:8:        - :exc:`~celery.exceptions.ImproperlyConfigured`
celery/exceptions.py:69:    'ImproperlyConfigured', 'SecurityError',
celery/exceptions.py:206:class ImproperlyConfigured(CeleryError):

celery/worker/worker.py:369:                       prefetch_multiplier=None, disable_rate_limits=None,
celery/worker/worker.py:407:        self.disable_rate_limits = either(
celery/worker/worker.py:408:            'worker_disable_rate_limits', disable_rate_limits,
celery/worker/components.py:236:            worker_options=w.options,
celery/worker/components.py:237:            disable_rate_limits=w.disable_rate_limits,
"""WorkController can be used to instantiate in-process workers.

The command-line interface for the worker is in :mod:`celery.bin.worker`,
while the worker program is in :mod:`celery.apps.worker`.

The worker program is responsible for adding signal handlers,
setting up logging, etc.  This is a bare-bones worker without

global side-effects (i.e., except for the global state stored in
:mod:`celery.worker.state`).

The worker consists of several components, all managed by
…[3199 chars omitted]…
establish, only used for params
        self._conninfo = self.app.connection_for_read()
        self.use_eventloop = (
            self.should_use_eventloop() if use_eventloop is None
            else use_eventloop
        )
        self.options = kwargs

        signals.worker_init.send(sender=self)

        # Initialize bootsteps
        self.pool_cls = _concurrency.get_implementation(self.pool_cls)

"""Worker command-line program.

This module is the 'program-version' of :mod:`celery.worker`.

It does everything necessary to run that module
as an actual application, like installing signal handlers,
platform tweaks, and so on.
"""
import logging
import os
import platform as _platform
import sys

from datetime import datetime
from functools import partial

from billiard.common import REMAP_SIGTERM
from billiard.process import current_process
from kombu.utils.encoding import safe_str

from celery import VERSION_BANNER, platforms, signals
from celery.app import trace
from celery.loaders.app import AppLoader
from celery.platforms import EX_FAILURE, EX_OK, check_privileges, isatty
from celery.utils import static, term
from celery.utils.debug import cry
from celery.utils.imports import qualname
from celery.utils.log import get_logger, in_sighandler, set_in_sighandler
from celery.utils.te
…[5490 chars omitted]…
  @cached_property
1348-    def Task(self):
1349-        """Base task class for this app."""
1350-        return self.create_task_cls()
1351-
1352-    @cached_property
1353-    def annotations(self):
1354-        return prepare_annotations(self.conf.task_annotations)
1355-
1356-    @cached_property
1357-    def AsyncResult(self):
1358-        """Create new result instance.
1359-
1360-        See Also:

@click.option('--exclude-queues',
              '-X',
              type=COMMA_SEPARATED_LIST,
              cls=CeleryOption,
              help_group="Queue Options")
@click.option('--include',
              '-I',
              type=COMMA_SEPARATED_LIST,
              cls=CeleryOption,
              help_group="Queue Options")
@click.option('--without-gossip',
              is_flag=True,

              cls=CeleryOption,
              help_group="Features")
@click.option('--without-mingle',
              is_flag=True,
              cls=CeleryOption,
              help_group="Features")
@click.option('--without-heartbeat',
              is_flag=True,
              cls=CeleryOption,
              help_group="Features", )
@click.option('--heartbeat-interval',
              type=int,
              cls=CeleryOption,
              help_group="Features", )
@click.option('--autoscale',
      
…[2595 chars omitted]…
  logfile=logfile,  # node format handled by celery.app.log.setup
            pidfile=node_format(pidfile, hostname),
            statedb=node_format(statedb, hostname),
            no_color=ctx.obj.no_color,
            quiet=ctx.obj.quiet,
            **kwargs)
        worker.start()
        ctx.exit(worker.exitcode)
    except SecurityError as e:
        ctx.obj.error(e.args[0])
        ctx.exit(1)

import os
from unittest.mock import patch

import pytest
from click.testing import CliRunner

from celery.app.log import Logging
from celery.bin.celery import celery


@pytest.fixture(scope='session')
def use_celery_app_trap():

    return False


def test_cli(isolated_cli_runner: CliRunner):
    Logging._setup = True  # To avoid hitting the logging sanity checks
    res = isolated_cli_runner.invoke(
        celery,
        ["-A", "t.unit.bin.proj.app", "worker", "--pool", "solo"],
        catch_exceptions=False
    )
    assert res.exit_code == 1, (res, res.stdout)


def test_cli_skip_checks(isolated_cli_runner: CliRunner):
    Logging._setup = True  # To avoid hitting the logging sanity checks
    with patch.dict(os.environ, clear=True):
        res = isolated_cli_runner.invoke(
            celery,
            ["-A", "t.unit.bin.proj.app", "--skip-checks", "worker", "--pool", "solo"]
…[275 chars omitted]…
, capfd):
1235-        safe_say("hello")
1236-        captured = capfd.readouterr()
1237-        assert "\nhello\n" == captured.err
1238-        assert "" == captured.out
1239-
1240-    def test_safe_say_writes_to_std_out(self, capfd):
1241-        safe_say("out", sys.stdout)
1242-        captured = capfd.readouterr()
1243-        assert "\nout\n" == captured.out
1244-        assert "" == captured.err

from unittest.mock import Mock, patch

import pytest

import t.skip
from celery.exceptions import ImproperlyConfigured
from celery.worker.components import Beat, Hub, Pool, Timer

# some of these are tested in test_worker, so I've only written tests
# here to complete coverage.  Should move everything to this module at some
# point [-ask]

class test_Timer:

    def test_create__eventloop(self):
        w = Mock(name='w')
        w.use_eventloop = True
        Timer(w).create(w)
        assert not w.timer.queue


class test_Hub:

    def setup_method(self):
        self.w = Mock(name='w')
        self.hub = Hub(self.w)
        self.w.hub = Mock(name='w.hub')

    @patch('celery.worker.components.set_event_loop')
    @patch('celery.worker.components.get_event_loop')
    def test_create(self, get_event_loop, set_event_loop):
        self.hub._patch_thread_primitives = Mock(name='ptp')

…[1159 chars omitted]…
 True
        comp = Pool(w)
        comp.instantiate = Mock()
        w.max_memory_per_child = 32

        comp.create(w)

        assert comp.instantiate.call_args[1]['max_memory_per_child'] == 32


class test_Beat:

    def test_create__green(self):
        w = Mock(name='w')
        w.pool_cls.__module__ = 'foo_gevent'
        with pytest.raises(ImproperlyConfigured):
            Beat(w).create(w)

data.setdefault('id', uuid())
    m = Message(body=pickle.dumps(dict(**data)),
                channel=channel,
                content_type='application/x-python-serialize',
                content_encoding='binary',
                delivery_info={'consumer_tag': 'mock'})
    m.accept = ['application/x-python-serialize']
    return m


class ConsumerCase:

    def create_task_message(self, channel, *args, **kwargs):
        m = self.TaskMessage(*args, **kwargs)
        m.channel = channel
        m.delivery_info = {'consumer_tag': 'mock'}
        return m


class test_Consumer(ConsumerCase):

    def setup_method(self):
        self.buffer = FastQueue()
        self.timer = Timer()

        @self.app.task(shared=False)
        def foo_task(x, y, z):
            return x * y * z
        self.foo_task = foo_task

    def teardown_method(self):
        self.timer.stop()

    def Loo
…[7942 chars omitted]…
umer.restart_count = -1
        pool = components.Pool(w)
        pool.create(w)
        pool.register_with_event_loop(w, w.hub)
        if sys.platform != 'win32':
            assert isinstance(w.semaphore, LaxBoundedSemaphore)
            P = w.pool
            P.start()

    def test_wait_for_soft_shutdown(self):
        worker = self.worker
        worker.app.conf.worker_soft_shutdown_timeout = 10

* master
commit 5f8926c123e83ce5a1f96cab6cbc3f41f09f06e4
Author: benchgen <sandbox@benchgen>
Date:   Fri Nov 6 06:54:33 2026 +0000

    base

 .bumpversion.cfg                                   |   14 +
 .dockerignore                                      |   33 +
 .editorconfig                                      |   15 +
 .github/FUNDING.yml                                |    8 +
 .github/ISSUE_TEMPLATE.md                          |    4 +

 .github/ISSUE_TEMPLATE/Bug-Report.md               |  166 +
 .github/ISSUE_TEMPLATE/Documentation-Bug-Report.md |   56 +
 .github/ISSUE_TEMPLATE/Enhancement.md              |   94 +
 .github/ISSUE_TEMPLATE/Feature-Request.md          |   93 +
 .../Major-Version-Release-Checklist.md             |   48 +
 .../Minor-Version-Release-Checklist.md             |  136 +
 .../Patch-Version-Release-Checklist.md             |  136 +
 .github/ISSUE_TEMPLATE/config.yml                  |   13 +
 .github/PULL_REQUEST_TEMPLATE.md                   |   14 +

=== /root/.cache/pip/http-v2/f/4/8/d/5/f48d54f8259d6db30ecd31d2a1f0bc992d6ada4df11686317391422d.body
bash: line 113: file: command not found
-rw-r--r-- 1 root root 9.2K Nov  6 06:57 /root/.cache/pip/http-v2/f/4/8/d/5/f48d54f8259d6db30ecd31d2a1f0bc992d6ada4df11686317391422d.body
Metadata-Version: 2.4
Name: amqp
Version: 5.4.2
Summary: Low-level AMQP client for Python (fork of amqplib).
Home-page: http://github.com/celery/py-amqp
=== /root/.cache/pip/http-v2/f/9/c/b/8/f9cb8ee4247c64344e5c175b21d83ac8e0fa1db850e4026f1e983ddf
bash: line 113: file: command not found
-rw-r--r-- 1 root root 1.3K Nov  6 06:57 /root/.cache/pip/http-v2/f/9/c/b/8/f9cb8ee4247c64344e5c175b21d83ac8e0fa1db850e4026f1e983ddf
cc=4,

response
body
headers
Connection
=== /root/.cache/pip/http-v2/f/9/c/b/8/f9cb8ee4247c64344e5c175b21d83ac8e0fa1db850e4026f1e983ddf.body
bash: line 113: file: command not found
-rw-r--r-- 1 roo
…[1708 chars omitted]…
8f631d61332e75486a9788a3b286b50e9304f43993
cc=4,
response
body
headers
Connection
=== /root/.cache/pip/http-v2/2/e/6/2/b/2e62b6dea7ab248f631d61332e75486a9788a3b286b50e9304f43993.body
bash: line 113: file: command not found
-rw-r--r-- 1 root root 48K Nov  6 06:57 /root/.cache/pip/http-v2/2/e/6/2/b/2e62b6dea7ab248f631d61332e75486a9788a3b286b50e9304f43993.body
pytest_celery/__init__.py
3%*$
?d_jr
&BR^&	z

bash: line 116: wget: command not found

attr before <Mock name='mock.disable_prefetch' id='137457801034592'>
dict dict_keys(['_mock_return_value', '_mock_parent', '_mock_name', '_mock_new_name', '_mock_new_parent', '_mock_sealed', '_spec_class', '_spec_set', '_spec_signature', '_mock_methods', '_spec_asyncs', '_mock_children', '_mock_wraps', '_mock_delegate', '_mock_called', '_mock_call_args', '_mock_call_count', '_mock_call_args_list', '_mock_mock_calls', 'method_calls', '_mock_unsafe', '_mock_side_effect'])
attr after False
dict dict_keys(['_mock_return_value', '_mock_parent', '_mock_name', '_mock_new_name', '_mock_new_parent', '_mock_sealed', '_spec_class', '_spec_set', '_spec_signature', '_mock_methods', '_spec_asyncs', '_mock_children', '_mock_wraps', '_mock_delegate', '_mock_called', '_mock_call_args', '_mock_call_count', '_mock_call_args_list', '_mock_mock_calls', 'method_calls', '_mock_unsafe', '_mock_side_effect', 'disable_prefetch'])
_mock_children dict_keys(['disable_prefetch'])
isinstance bool True

Refused: `find /` walks the whole container FS. Scope under /repo.

__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
/usr/local/lib/python3.12/site-packages/celery-5.5.3.dist-info

class QoS:
    """Thread safe increment/decrement of a channels prefetch_count.

    Arguments:
    ---------
        callback (Callable): Function used to set new prefetch count,
            e.g. ``consumer.qos`` or ``channel.basic_qos``.  Will be called
            with a single ``prefetch_count`` keyword argument.
        initial_value (int): Initial prefetch count value..
        max_prefetch (int or None): Maximum allowed prefetch count. If specified
            as an integer, increment_eventually will not allow the value to exceed this limit.
            If None (the default), there is no upper limit on the prefetch count.

    Example:
    -------
        >>> from kombu import Consumer, Connection
        >>> connection = Connection('amqp://')
        >>> consumer = Consumer(connection)
        >>> qos = QoS(consumer.qos, initial_prefetch_count=2)
        >>> qos.update()  # se
…[2213 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)

--- QoS source ---
Traceback (most recent call last):
  File "<stdin>", line 7, in <module>
AttributeError: module 'amqp.channel' has no attribute 'QoS'

/usr/local/lib/python3.12/site-packages/amqp

/usr/local/lib/python3.12/site-packages/kombu/mixins.py:173:                if restart_limit.can_consume(_tokens):  # pragma: no cover
/usr/local/lib/python3.12/site-packages/kombu/utils/limits.py:53:    def can_consume(self, tokens=1):
/usr/local/lib/python3.12/site-packages/kombu/transport/redis.py:548:                if channel.qos.can_consume():
/usr/local/lib/python3.12/site-packages/kombu/transport/redis.py:578:        if chan.qos.can_consume():
/usr/local/lib/python3.12/site-packages/kombu/transport/redis.py:593:                    if channel.qos.can_consume():
/usr/local/lib/python3.12/site-packages/kombu/transport/confluentkafka.py:109:    def can_consume(self):
/usr/local/lib/python3.12/site-packages/kombu/transport/confluentkafka.py:118:    def can_consume_max_estimate(self):
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py:198:    def can_consume(sel
…[780 chars omitted]…
/transport/qpid.py:195:    def can_consume(self):
/usr/local/lib/python3.12/site-packages/kombu/transport/qpid.py:203:            can_consume will always return True.
/usr/local/lib/python3.12/site-packages/kombu/transport/qpid.py:212:    def can_consume_max_estimate(self):

/usr/local/lib/python3.12/site-packages/kombu/transport/gcpubsub.py:480:        max_allowed = self.qos.can_consume_max_estimate()

Arguments:
    ---------
        channel (ChannelT): Connection channel.
        prefetch_count (int): Initial prefetch count (defaults to 0).
    """

    #: current prefetch count value
    prefetch_count = 0

    #: :class:`~collections.OrderedDict` of active messages.
    #: *NOTE*: Can only be modified by the consuming thread.
    _delivered = None

    #: acks can be done by other threads than the consuming thread.
    #: Instead of a mutex, which doesn't perform well here, we mark
    #: the delivery tags as dirty, so subsequent calls to append() can remove
    #: them.
    _dirty = None

    #: If disabled, unacked messages won't be restored at shutdown.
    restore_at_shutdown = True

    def __init__(self, channel, prefetch_count=0):
        self.channel = channel
        self.prefetch_count = prefetch_count or 0

        # Standard Python dictionaries do not support se
…[4807 chars omitted]…
ery_tag, requeue=False):
248-        """Remove from transactional state and requeue message."""
249-        if requeue:
250-            self.channel._restore_at_beginning(self._delivered[delivery_tag])
251-        self._quick_ack(delivery_tag)
252-
253-    def restore_unacked(self):
254-        """Restore all unacknowledged messages."""
255-        self._flush()
256-        delivered = self._delivered

/usr/local/lib/python3.12/site-packages/amqp/connection.py:340: AMQPDeprecationWarning: The .frame_writer attribute on the connection was accessed before
the connection was established.  This is supported for now, but will
be deprecated in amqp 2.2.0.

Since amqp 2.0 you have to explicitly call Connection.connect()
before using the connection.

  warnings.warn(AMQPDeprecationWarning(
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/amqp/connection.py", line 518, in channel
    return self.channels[channel_id]
           ~~~~~~~~~~~~~^^^^^^^^^^^^

KeyError: None

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
  File "/usr/local/lib/python3.12/site-packages/amqp/connection.py", line 521, in channel
    channel.open()
  File "/usr/local/lib/python3.12/site-packa
…[431 chars omitted]…
te-packages/amqp/connection.py", line 327, in connect
    self.transport.connect()
  File "/usr/local/lib/python3.12/site-packages/amqp/transport.py", line 132, in connect
    self._connect(self.host, self.port, self.connect_timeout)
  File "/usr/local/lib/python3.12/site-packages/amqp/transport.py", line 187, in _connect
    self.sock.connect(sa)
ConnectionRefusedError: [Errno 111] Connection refused

grep: /usr/local/lib/python3.12/site-packages/kombu/transport/amqp.py: No such file or directory

SLMQ.py
SQS.py
__init__.py
__pycache__
azureservicebus.py
azurestoragequeues.py
base.py
confluentkafka.py
consul.py
etcd.py
filesystem.py
gcpubsub.py

librabbitmq.py
memory.py
mongodb.py
native_delayed_delivery.py
pyamqp.py
pyro.py
qpid.py
redis.py
sqlalchemy
virtual
zookeeper.py
/usr/local/lib/python3.12/site-packages/kombu/transport/redis.py
/usr/local/lib/python3.12/site-packages/kombu/transport/confluentkafka.py
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/__pycache__/base.cpython-312.pyc
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py
/usr/local/lib/python3.12/site-packages/kombu/transport/SQS.py
/usr/local/lib/python3.12/site-packages/kombu/transport/qpid.py
/usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/redis.cpython-312.pyc
/usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/SQS.cpython-312.pyc
/usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/gcpubsub.cpython-312.pyc
/usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/confluentkafka.cpython-312.pyc
/usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/qpid.cpython-312.pyc
/usr/local/lib/python3.12/site-packages/kombu/transport/gcpubsub.py

224:    def qos_semantics_matches_spec(self, connection):
False
['basic_consume', 'basic_qos']

"""pyamqp transport module for Kombu.

Pure-Python amqp transport using py-amqp library.

Features
========
* Type: Native
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: Yes
* Supports Priority: Yes
* Supports TTL: Yes

Connection String
=================
Connection string can have the following formats:

.. code-block::

    amqp://[USER:PASSWORD@]BROKER_ADDRESS[:PORT][/VIRTUALHOST]
    [USER:PASSWORD@]BROKER_ADDRESS[:PORT][/VIRTUALHOST]
    amqp://

For TLS encryption use:

.. code-block::

    amqps://[USER:PASSWORD@]BROKER_ADDRESS[:PORT][/VIRTUALHOST]

Transport Options
=================
Transport Options are passed to constructor of underlying py-amqp
:class:`~kombu.connection.Connection` class.

Using TLS
=========
Transport over TLS can be enabled by ``ssl`` parameter of
:class:`~kombu.Connection` class. By setting ``ssl=True``, TLS transport is
used::

    con
…[1930 chars omitted]…
e, _Message=amqp.Message):
        """Prepare message so that it can be sent using this transport."""
        return _Message(
            body,
            priority=priority,
            content_type=content_type,
            content_encoding=content_encoding,
            application_headers=headers,
            **properties or {}
        )

    def prepare_queue_arguments(self, arguments, **kwargs):

grep: /usr/local/lib/python3.12/site-packages/amqp/__pycache__/exceptions.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/amqp/__pycache__/channel.cpython-312.pyc: binary file matches
/usr/local/lib/python3.12/site-packages/amqp/exceptions.py:257:    (60, 10): 'Basic.qos',
/usr/local/lib/python3.12/site-packages/amqp/exceptions.py:258:    (60, 11): 'Basic.qos_ok',
/usr/local/lib/python3.12/site-packages/amqp/channel.py:1833:    def basic_qos(self, prefetch_size, prefetch_count, a_global,
/usr/local/lib/python3.12/site-packages/amqp/channel.py:1840:        a qos method always depend on the content class semantics.
/usr/local/lib/python3.12/site-packages/amqp/channel.py:1841:        Though the qos method could in principle apply to both peers,

base_path=Option('', type='string'),
        connection_timeout=Option(20, type='int'),
        read_timeout=Option(120, type='int'),
    ),
    gcs=Namespace(
        bucket=Option(type='string'),
        project=Option(type='string'),
        base_path=Option('', type='string'),
        ttl=Option(0, type='float'),
    ),
    control=Namespace(
        queue_ttl=Option(300.0, type='float'),

        queue_expires=Option(10.0, type='float'),
        exchange=Option('celery', type='string'),
    ),
    couchbase=Namespace(
        __old__=old_ns('celery_couchbase'),

        backend_settings=Option(None, type='dict'),
    ),
    arangodb=Namespace(
        __old__=old_ns('celery_arangodb'),
        backend_settings=Option(None, type='dict')
    ),
    mongodb=Namespace(
        __old__=old_ns('celery_mongodb'),

        backend_settings=Option(type='dict'),
    ),
    cosmosdbs
…[3798 chars omitted]…
,
        send_task_events=Option(
            False, type='bool', old={'celery_send_events'},
        ),
        state_db=Option(),
        task_log_format=Option(DEFAULT_TASK_LOG_FMT),
        timer=Option(type='string'),
        timer_precision=Option(1.0, type='float'),
        detect_quorum_queues=Option(True, type='bool'),
    ),
)


def _flatten_keys(ns, key, opt):
    return [(ns + key, opt)]

t/unit/worker/test_consumer.py:87:        c.initial_prefetch_count = None
t/unit/worker/test_consumer.py:91:        c.initial_prefetch_count = 10
t/unit/worker/test_consumer.py:95:        assert c.initial_prefetch_count == 10 * 10
t/unit/worker/test_consumer.py:131:            assert c.initial_prefetch_count != 0
t/unit/worker/test_consumer.py:134:            assert c.initial_prefetch_count == expected_initial
t/unit/worker/test_consumer.py:178:            assert c.initial_prefetch_count != 0
t/unit/worker/test_consumer.py:181:            assert c.initial_prefetch_count == 2
t/unit/worker/test_control.py:45:        self.initial_prefetch_count = 1
t/unit/worker/test_control.py:361:        assert consumer.initial_prefetch_count == 16
t/unit/worker/test_control.py:365:        assert consumer.initial_prefetch_count == 8

4:from celery.app.defaults import (_OLD_DEFAULTS, _OLD_SETTING_KEYS, _TO_NEW_KEY, _TO_OLD_KEY, DEFAULTS, NAMESPACES,
8:class test_defaults:
11:        self._prev = sys.modules.pop('celery.app.defaults', None)
15:            sys.modules['celery.app.defaults'] = self._prev
18:        assert repr(NAMESPACES['broker']['url'])
22:        assert self.defaults.Option.typemap['any'](val) is val
40:        find = self.defaults.find
42:        assert find('default_queue')[2].default == 'celery'
43:        assert find('task_default_exchange')[2] is None
46:    def defaults(self):
47:        return import_module('celery.app.defaults')
import sys

from importlib import import_module

from celery.app.defaults import (_OLD_DEFAULTS, _OLD_SETTING_KEYS, _TO_NEW_KEY, _TO_OLD_KEY, DEFAULTS, NAMESPACES,
                                 SETTING_KEYS)


class test_defaults:

    def setup_method(self):
    
…[937 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')

celery/app/backends.py:51:        reraise(ImproperlyConfigured, ImproperlyConfigured(
celery/app/backends.py:54:        raise ImproperlyConfigured(UNKNOWN_BACKEND.strip().format(
celery/app/base.py:117:        raise ImproperlyConfigured('You need to install pydantic to use pydantic model serialization.') from ex
celery/app/base.py:715:            raise ImproperlyConfigured(
celery/app/task.py:942:            raise ImproperlyConfigured(
celery/app/task.py:946:            raise ImproperlyConfigured("Cannot replace with an empty chain")
celery/app/utils.py:274:        raise ImproperlyConfigured(info.mix_error.format(renames='\n'.join(
celery/backends/arangodb.py:59:            raise ImproperlyConfigured(
celery/backends/arangodb.py:81:                raise ImproperlyConfigured(
celery/backends/azureblockblob.py:42:            raise ImproperlyConfigured(
celery/backends/azureblockblob.py:
…[2855 chars omitted]…

ry/backends/s3.py:47:            raise ImproperlyConfigured('Missing bucket name')
celery/backends/s3.py:86:            raise ImproperlyConfigured('Missing aws s3 creds')
celery/exceptions.py:206:class ImproperlyConfigured(CeleryError):
celery/fixups/django.py:47:        raise ImproperlyConfigured('Celery 5.x requires Django 1.11 or later.')
celery/result.py:767:            raise ImproperlyConfigured(

done part 1

done part 2

celery/app/amqp.py                 | 36 +++++++++++++++++++++++++++++++++---
 celery/app/control.py              | 11 ++++++++++-
 celery/app/defaults.py             |  7 +++++++
 celery/bin/worker.py               |  6 ++++++
 celery/events/receiver.py          | 11 ++++++++++-
 celery/worker/components.py        |  1 +
 celery/worker/consumer/consumer.py |  2 ++
 celery/worker/consumer/tasks.py    | 38 ++++++++++++++++++++++++++++++++++++++
 celery/worker/worker.py            |  4 ++++
 9 files changed, 111 insertions(+), 5 deletions(-)
diff --git a/celery/app/amqp.py b/celery/app/amqp.py
index 8dcec36..f62885a 100644

--- a/celery/app/amqp.py
+++ b/celery/app/amqp.py
@@ -46,6 +46,10 @@ class Queues(dict):
         create_missing (bool): By default any unknown queues will be
             added automatically, but if this flag is disabled the occurrence
             of unknown queues 
…[8629 chars omitted]…
mit:
+                return False
+            return original_can_consume()
+
+        c.task_consumer.channel.qos.can_consume = can_consume
+        return can_consume
+
+    # Kept as an alias for callers using the private spelling.
+    def _disable_prefetch(self, c):
+        return self.disable_prefetch(c)
+
     def stop(self, c):
         """Stop task consumer."""
         if c.task_consumer:

direct Queues new_missing OK
default new_missing OK
app.amqp.queues config OK
app.amqp.Queues explicit OK
control both raises: control_queue_durable and control_queue_exclusive cannot both be True
control OK
event both raises: event_queue_durable and event_queue_exclusive cannot both be True
event OK
tasks prefetch enabled OK
tasks prefetch off untouched OK
tasks config only OK
worker disable_prefetch OK

=============================== warnings summary ===============================
t/unit/worker/test_consumer.py: 16 warnings
t/unit/worker/test_worker.py: 1 warning
  /repo/celery/worker/consumer/consumer.py:394: CPendingDeprecationWarning: 
  In Celery 5.1 we introduced an optional breaking change which
  on connection loss cancels all currently executed tasks with late acknowledgement enabled.
  These tasks cannot be acknowledged as the connection is gone, and the tasks are automatically redelivered
  back to the queue. You can enable this behavior using the worker_cancel_long_running_tasks_on_connection_loss
  setting. In Celery 5.1 it is set to False by default. The setting will be set to True by default in Celery 6.0.
  
    warnings.warn(CANCEL_TASKS_BY_DEFAULT, CPendingDeprecationWarning)

t/unit/worker/test_consumer.py::test_Consumer::test_ensure_connected[True-None-False]
t/u
…[1126 chars omitted]…
ng: You're running the worker with superuser privileges: this is
  absolutely not recommended!
  
  Please specify a different user using the --uid option.
  
  User information: uid=0 euid=0 gid=0 egid=0
  
    warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
324 passed, 1 skipped, 25 warnings, 37 subtests passed in 13.54s

1424:    @cached_property
1466:    @cached_property
1467:    def control(self):
1471:    @cached_property
1476:    @cached_property
1467:    def control(self):
1468-        """Remote control: :class:`~@control`."""
1469-        return instantiate(self.control_cls, app=self)
1470-
1471-    @cached_property
1472-    def events(self):
1473-        """Consuming and sending events: :class:`~@events`."""

1474-        return instantiate(self.events_cls, app=self)
1475-

updated
event args OK

File "/usr/local/lib/python3.12/unittest/mock.py", line 1198, in _execute_mock_call
      raise effect
  Exception
  
    warn(RuntimeWarning(

t/unit/bin/test_daemonization.py::test_daemon_options_from_config[events]
  /usr/local/lib/python3.12/site-packages/click/core.py:1369: UserWarning: The parameter -d is used more than once. Remove its duplicate as parameters should be unique.
    parser = self.make_parser(ctx)

t/unit/bin/test_daemonization.py::test_daemon_options_from_config[events]
  /usr/local/lib/python3.12/site-packages/click/core.py:1362: UserWarning: The parameter -d is used more than once. Remove its duplicate as parameters should be unique.

    self.parse_args(ctx, args)

t/unit/bin/test_worker.py::test_cli
t/unit/bin/test_worker.py::test_cli_skip_checks
  /repo/celery/platforms.py:841: SecurityWarning: You're running the worker with superuser privileges: this is

…[198 chars omitted]…
at(

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

F
=================================== FAILURES ===================================
________________ test_preload_options[subcommand_with_params0] _________________

subcommand_with_params = ('purge', '-f')
isolated_cli_runner = <click.testing.CliRunner object at 0x738f5c0090d0>

    @pytest.mark.parametrize(
        "subcommand_with_params",
        [
            ("purge", "-f"),
            ("shell",),

        ]
    )
    def test_preload_options(subcommand_with_params: Tuple[str, ...], isolated_cli_runner: CliRunner):
        # Verify commands like shell and purge can accept preload options.
        # Projects like Pyramid-Celery's ini option should be valid preload
        # options.
        res_without_preload = isolated_cli_runner.invoke(
            celery,
            ["-A", "t.unit.bin.proj.app", *subcommand_with_params, "--ini", "some_ini.ini"],
            catch_exceptions=F
…[850 chars omitted]…
ths instead.
    with cli_runner.isolated_filesystem():

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
FAILED t/unit/app/test_preload_cli.py::test_preload_options[subcommand_with_params0]
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 1 warning in 1.14s

exit 0
start called True
output /repo/celery/platforms.py:841: SecurityWarning: You're running the worker with superuser privileges: this is
absolutely not recommended!

Please specify a different user using the --uid option.

User information: uid=0 euid=0 gid=0 egid=0

  warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(

exit 0 captured 1
disable_prefetch True <class 'bool'>
consumer arg True

2 []

0 False

grep: celery/app/__pycache__/amqp.cpython-312.pyc: binary file matches
celery/app/amqp.py:61:                 create_missing=True, autoexchange=None,
celery/app/amqp.py:73:        self.autoexchange = Exchange if autoexchange is None else autoexchange
celery/app/amqp.py:201:            self.autoexchange(name, type=exchange_type),
celery/app/amqp.py:233:    # For example, you can use ``autoexchange = lambda n: None`` to use the
celery/app/amqp.py:236:    autoexchange = None
celery/app/amqp.py:262:               autoexchange=None, max_priority=None,
celery/app/amqp.py:286:        autoexchange = (self.autoexchange if autoexchange is None
celery/app/amqp.py:287:                        else autoexchange)
celery/app/amqp.py:290:            autoexchange, max_priority, default_routing_key,

Please specify a different user using the --uid option.
  
  User information: uid=0 euid=0 gid=0 egid=0
  
    warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
324 passed, 1 skipped, 25 warnings, 37 subtests passed in 14.73s

diff --git a/celery/app/amqp.py b/celery/app/amqp.py
index 8dcec36..f62885a 100644
--- a/celery/app/amqp.py
+++ b/celery/app/amqp.py
@@ -46,6 +46,10 @@ class Queues(dict):
         create_missing (bool): By default any unknown queues will be
             added automatically, but if this flag is disabled the occurrence
             of unknown queues in `wanted` will raise :exc:`KeyError`.
+        create_missing_queue_type (str): Default queue type for
+            automatically created queues.
+        create_missing_queue_exchange_type (str): Default exchange type for
+            automatically created queues.

         max_priority (int): Default x-max-priority for queues with none set.
     """
 
@@ -55,12 +59,17 @@ class Queues(dict):
 
     def __init__(self, queues=None, default_exchange=None,
                  create_missing=True, autoexchange=None,
-                 max_priorit
…[8629 chars omitted]…
elf.app.either
@@ -407,6 +408,9 @@ class WorkController:
         self.disable_rate_limits = either(
             'worker_disable_rate_limits', disable_rate_limits,
         )
+        self.disable_prefetch = either(
+            'worker_disable_prefetch', disable_prefetch,
+        )
         self.worker_lost_wait = either('worker_lost_wait', worker_lost_wait)
 
     def wait_for_soft_shutdown(self):

True

t/unit/conftest.py:124:def app(celery_app):
t/unit/conftest.py-125-    yield celery_app
t/unit/conftest.py-126-
t/unit/conftest.py-127-
t/unit/conftest.py-128-@pytest.fixture(autouse=True, scope='session')
t/unit/conftest.py-129-def AAA_disable_multiprocessing():
t/unit/conftest.py-130-    # pytest-cov breaks if a multiprocessing.Process is started,
t/unit/conftest.py-131-    # so disable them completely to make sure it doesn't happen.
t/unit/conftest.py-132-    stuff = [
t/unit/conftest.py-133-        'multiprocessing.Process',
t/unit/conftest.py-134-        'billiard.Process',
t/unit/conftest.py-135-        'billiard.context.Process',

t/unit/conftest.py-136-        'billiard.process.Process',
t/unit/conftest.py-137-        'billiard.process.BaseProcess',
t/unit/conftest.py-138-        'multiprocessing.Process',
t/unit/conftest.py-139-    ]
t/unit/conftest.py-140-    ctxs = [patch(s) for s in stuff]
t/unit/conftest.py-141-    [ctx.__enter__() for ctx in ctxs]
t/unit/conftest.py-142-
t/unit/conftest.py-143-    yield
t/unit/conftest.py-144-
__init__.py
__pycache__
benchmarks
integration
skip.py
smoke
unit

==================================== ERRORS ====================================
___________ ERROR collecting t/unit/backends/test_azureblockblob.py ____________
ImportError while importing test module '/repo/t/unit/backends/test_azureblockblob.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/local/lib/python3.12/importlib/__init__.py:90: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
t/unit/backends/test_azureblockblob.py:6: in <module>
    from celery.backends import azureblockblob
celery/backends/azureblockblob.py:2: in <module>
    from kombu.transport.azurestoragequeues import Transport as AzureStorageQueuesTransport
/usr/local/lib/python3.12/site-packages/kombu/transport/azurestoragequeues.py:59: in <module>

    from azure.core.exceptions import ResourceExistsError
E   ModuleNotFoundError: No module named
…[540 chars omitted]…
ps://cloud.google.com/security/resources/post-quantum-cryptography
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
ERROR t/unit/backends/test_azureblockblob.py
!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
8 skipped, 1 warning, 1 error in 8.33s

True True False

grep: t/unit/bin/__pycache__/test_beat.cpython-312-pytest-8.3.5.pyc: binary file matches
grep: t/unit/bin/__pycache__/test_control.cpython-312-pytest-8.3.5.pyc: binary file matches
grep: t/unit/bin/__pycache__/test_daemonization.cpython-312-pytest-8.3.5.pyc: binary file matches
grep: t/unit/bin/__pycache__/test_worker.cpython-312-pytest-8.3.5.pyc: binary file matches
grep: t/unit/app/__pycache__/test_preload_cli.cpython-312-pytest-8.3.5.pyc: binary file matches
t/unit/bin/test_beat.py:2:from click.testing import CliRunner
t/unit/bin/test_beat.py:13:def test_cli(isolated_cli_runner: CliRunner):
t/unit/bin/test_beat.py:25:def test_cli_quiet(isolated_cli_runner: CliRunner):
t/unit/bin/test_control.py:6:from click.testing import CliRunner
t/unit/bin/test_control.py:30:def test_custom_remote_command(celery_cmd, custom_cmd, isolated_cli_runner: CliRunner):
t/unit/bin/test_control.py:51:def 
…[1562 chars omitted]…

gging sanity checks
    with patch.dict(os.environ, clear=True):
        res = isolated_cli_runner.invoke(
            celery,
            ["-A", "t.unit.bin.proj.app", "--skip-checks", "worker", "--pool", "solo"],
            catch_exceptions=False,
        )
        assert res.exit_code == 1, (res, res.stdout)
        assert os.environ["CELERY_SKIP_CHECKS"] == "true", "should set CELERY_SKIP_CHECKS"

celery/app/control.py:675:    def disable_events(self, destination=None, **kwargs):
celery/security/__init__.py:73:def disable_untrusted_serializers(whitelist=None):
celery/worker/consumer/tasks.py:62:    def disable_prefetch(self, c):
celery/worker/consumer/tasks.py:85:    def _disable_prefetch(self, c):
celery/worker/control.py:349:def disable_events(state):
celery/_state.py:188:def disable_trace():
celery/worker/autoscale.py:133:    def _update_consumer_prefetch_count(self, new_max):
celery/worker/consumer/consumer.py:266:    def _update_prefetch_count(self, index=0):
celery/worker/consumer/consumer.py:703:    def _restore_prefetch_count_after_connection_restart(self, p, *args):
celery/worker/consumer/consumer.py:725:    def max_prefetch_count(self):
celery/worker/consumer/consumer.py:729:    def _new_prefetch_count(self):
celery/worker/consumer/tasks.py:45:        def set_prefetch_count(prefetch_count):

celery/worker/consumer/tasks.py:55:    def _prefetch_disabled(self, c):
celery/worker/consumer/tasks.py:62:    def disable_prefetch(self, c):
celery/worker/consumer/tasks.py:85:    def _disable_prefetch(self, c):

zip 257601 /root/.cache/pip/http-v2/1/c/3/9/a/1c39af2369d8f13af63e9d28afaa2cd539297d562f9121b3b37b278d.body
zip 99506 /root/.cache/pip/http-v2/1/c/c/1/7/1cc172b86d33cc913942115a6e1ccb72e3434e2c6e99926b871d1849.body
zip 16223768 /root/.cache/pip/http-v2/1/e/e/8/a/1ee8a22928711d0023b44a9326ae539cb7a55212484d6132cba9552c.body
? 327180 /root/.cache/pip/http-v2/1/6/9/3/2/1693297fb9daf7bfe370bf51d371acfeb8ff40759bf8650dfd404ba4.body
zip 263687 /root/.cache/pip/http-v2/1/7/9/0/3/179035e06c5c0599483ee50bf263fe51bed508bb5d9644011f3eb745.body
zip 129956 /root/.cache/pip/http-v2/b/1/1/9/0/b11909f451f61fb29636217082b1c638b177c59eac335e0200a136e6.body
zip 67548 /root/.cache/pip/http-v2/b/1/4/e/c/b14ec1286f8ec83b268f9c52cdccd1001d22496d6e01ba5b4fd8aeb4.body
zip 62600 /root/.cache/pip/http-v2/b/4/d/c/4/b4dc47ae9d349a950bbc69678b6f001f8a20486e60aee5a2e1823fed.body
zip 155560 /root/.cache/pip/http-v2/
…[7933 chars omitted]…
pip/http-v2/7/3/f/3/a/73f3abcfc6f7bae7a4f7224bca9394ab0b67176b94f9acd9c89ea7a6.body
zip 81515 /root/.cache/pip/http-v2/7/c/2/7/7/7c277f31c85503989f692653801137b2b9fe655cd86578309ac1e34f.body

zip 229892 /root/.cache/pip/http-v2/7/d/0/6/d/7d06ddc2f8070ce09a5a2290e2020c460e8ab32b6a2c6b69fc06e979.body
? 173795 /root/.cache/pip/http-v2/7/7/7/6/6/7776683a2864f491fb2b97c042ff46c9822fd902be50e2c129f5626a.body

t/unit/app/test_control.py:251:        assert control.Control(self.app).mailbox.serializer == 'test'
t/unit/app/test_control.py:252:        assert control.Control(self.app).mailbox.accept == ['test']
t/unit/app/test_control.py:560:        c = control.Control(self.app)
t/unit/app/test_control.py:565:        c = control.Control(self.app)
t/unit/events/test_events.py:205:        r = self.app.events.Receiver(
t/unit/events/test_events.py:214:        r = self.app.events.Receiver(Mock(), accept={'app/foo'})
t/unit/events/test_events.py:218:        r = self.app.events.Receiver(Mock())
t/unit/events/test_events.py:223:        r = self.app.events.Receiver(Mock())
t/unit/events/test_events.py:227:        r = self.app.events.Receiver(Mock(), queue_prefix='fooq')
t/unit/events/test_events.py:231:        r = self.app.events.Receiver(Mock())
t/unit/events/test_events.py:236:        r = self.app.eve
…[1696 chars omitted]…

y:281:        r = self.app.events.Receiver(Mock(), node_id='celery.tests')
t/unit/events/test_events.py:292:        r = self.app.events.Receiver(Mock(name='connection'))
t/unit/events/test_events.py:314:            r = self.app.events.Receiver(
t/unit/events/test_snapshot.py:101:        def Receiver(self, *args, **kwargs):
t/unit/events/test_snapshot.py:102:            return test_evcam.MockReceiver()

grep: t/unit/worker/__pycache__/test_consumer.cpython-312-pytest-8.3.5.pyc: binary file matches
grep: t/unit/worker/__pycache__/test_control.cpython-312-pytest-8.3.5.pyc: binary file matches
t/unit/worker/test_consumer.py:87:        c.initial_prefetch_count = None
t/unit/worker/test_consumer.py:91:        c.initial_prefetch_count = 10
t/unit/worker/test_consumer.py:95:        assert c.initial_prefetch_count == 10 * 10
t/unit/worker/test_consumer.py:131:            assert c.initial_prefetch_count != 0
t/unit/worker/test_consumer.py:134:            assert c.initial_prefetch_count == expected_initial
t/unit/worker/test_consumer.py:178:            assert c.initial_prefetch_count != 0
t/unit/worker/test_consumer.py:181:            assert c.initial_prefetch_count == 2
t/unit/worker/test_control.py:45:        self.initial_prefetch_count = 1
t/unit/worker/test_control.py:361:        assert consumer.initial_prefetch_count == 16
t/unit/worker/test_control.py:365:        assert consumer.initial_prefetch_count == 8

setup.cfg:6:[flake8]
setup.cfg:9:max-line-length = 117
tox.ini:9:    flake8

/usr/local/bin/python: No module named flake8

True True

done
Traceback (most recent call last):
  File "<stdin>", line 9, in <module>
AssertionError

spontaneous {'x-queue-type': 'quorum'} Exchange spontaneous(topic) topic
x {'x-queue-type': 'quorum'} Exchange x(topic) topic
x None Exchange ''(direct)

...........................................................              [100%]
59 passed in 0.65s

t/unit/worker/test_worker.py:282:        c = self.NoopConsumer(task_events=False, pool=BasePool())
t/unit/worker/test_worker.py:295:        c = self.NoopConsumer(task_events=False, pool=BasePool())
t/unit/worker/test_worker.py:684:        c = self.NoopConsumer(task_events=False, init_callback=init_callback)

M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/components.py
 M celery/worker/consumer/consumer.py
 M celery/worker/consumer/tasks.py
 M celery/worker/worker.py
?? statefilename
 celery/app/amqp.py                 | 44 +++++++++++++++++++++++++++++++++++---
 celery/app/control.py              | 11 +++++++++-

 celery/app/defaults.py             |  7 ++++++
 celery/bin/worker.py               |  6 ++++++
 celery/events/receiver.py          | 16 ++++++++++++--
 celery/worker/components.py        |  1 +
 celery/worker/consumer/consumer.py |  2 ++
 celery/worker/consumer/tasks.py    | 37 ++++++++++++++++++++++++++++++++
 celery/worker/worker.py            |  4 ++++
 9 files changed, 122 insertions(+), 6 deletions(-)

M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/components.py
 M celery/worker/consumer/consumer.py
 M celery/worker/consumer/tasks.py
 M celery/worker/worker.py

t/unit/app/test_app.py:349:            with pytest.raises(ImproperlyConfigured):
t/unit/app/test_app.py:377:            with pytest.raises(ImproperlyConfigured):
t/unit/app/test_app.py:412:            with pytest.raises(ImproperlyConfigured):
t/unit/app/test_app.py:806:                with pytest.raises(ImproperlyConfigured, match=msg):
t/unit/app/test_app.py:982:        with pytest.raises(ImproperlyConfigured) as exc:
t/unit/app/test_app.py:996:        with pytest.raises(ImproperlyConfigured) as exc:
t/unit/app/test_app.py:1077:        with pytest.raises(ImproperlyConfigured):
t/unit/app/test_backends.py:109:            with pytest.raises(ImproperlyConfigured):
t/unit/app/test_backends.py:113:        with pytest.raises(ImproperlyConfigured):
t/unit/backends/test_arangodb.py:28:            with pytest.raises(ImproperlyConfigured):
t/unit/backends/test_arangodb.py:35:        with pytes
…[2274 chars omitted]…

ch='Missing project'):
t/unit/backends/test_gcs.py:64:        with pytest.raises(ImproperlyConfigured, match='Invalid ttl'):
t/unit/backends/test_gcs.py:74:        with pytest.raises(ImproperlyConfigured, match='Missing TTL policy'):
t/unit/backends/test_mongodb.py:95:        with pytest.raises(ImproperlyConfigured):
t/unit/backends/test_mongodb.py:100:        with pytest.raises(ImproperlyConfigured):

144-              help="Set custom hostname (e.g., 'w1@%%h').  "
145-                   "Expands: %%h (hostname), %%n (name) and %%d, (domain).")
146-@click.option('-D',
147-              '--detach',
148-              cls=CeleryOption,
149:              is_flag=True,
150-              default=False,
151-              help_group="Worker Options",
152-              help="Start worker as a background process.")
153-@click.option('-S',
154-              '--statedb',
155-              cls=CeleryOption,

--
181-              cls=CeleryOption,
182-              help_group="Worker Options",
183-              help="Set custom prefetch multiplier value "
184-                   "for this worker instance.")
185-@click.option('--disable-prefetch',
186:              is_flag=True,
187-              default=None,
188-              cls=CeleryOption,
189-              help_group="Worker Options",
190-  
…[1940 chars omitted]…
-B',
287-              '--beat',
288-              type=CELERY_BEAT,
289-              cls=CeleryOption,
290:              is_flag=True,
291-              help_group="Embedded Beat Options")
292-@click.option('-s',
293-              '--schedule-filename',
294-              '--schedule',
295-              callback=lambda ctx, _,
296-              value: value or ctx.obj.app.conf.beat_schedule_filename,

M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/components.py
 M celery/worker/consumer/consumer.py
 M celery/worker/consumer/tasks.py
 M celery/worker/worker.py
            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)

    def new_missing(self, name):
        queue_arguments = None
        if self.create_missing_queue_type not in (None, 'classic'):
            queue_arguments = {
                'x-queue-type': self.create_missing_queue_type,
            }
        exchange_type = self.create_missing_queue_exchange_type or 'direct'
        try:
            exchange = self.autoexchange(n
…[2315 chars omitted]…
op task consumer."""
        if c.task_consumer:
            debug('Canceling task consumer...')
            ignore_errors(c, c.task_consumer.cancel)

    def shutdown(self, c):
        """Shutdown task consumer."""
        if c.task_consumer:
            self.stop(c)
            debug('Closing consumer channel...')
            ignore_errors(c, c.task_consumer.close)
            c.task_consumer = None

all reproducer checks passed

Traceback (most recent call last):
  File "<stdin>", line 3, in <module>
AttributeError: 'bool' object has no attribute 'default'

control_queue_durable False
control_queue_exclusive False
event_queue_durable False
event_queue_exclusive False
task_create_missing_queue_type 'classic'
task_create_missing_queue_exchange_type 'direct'
worker_disable_prefetch False
old worker []

[]
['CELERYD_DISABLE_PREFETCH']

raise MessageNacked()

        if not self._confirm_selected:
            self._confirm_selected = True
            self.confirm_select()
        ret = self._basic_publish(*args, **kwargs)
        # Waiting for confirmation of message.
        timeout = confirm_timeout or kwargs.get('timeout', None)
        self.wait([spec.Basic.Ack, spec.Basic.Nack],
                  callback=confirm_handler,
                  timeout=timeout)
        return ret

    def basic_qos(self, prefetch_size, prefetch_count, a_global,
                  argsig='lBb'):
        """Specify quality of service.

        This method requests a specific quality of service.  The QoS
        can be specified for the current channel or for all channels
        on the connection.  The particular properties and semantics of
        a qos method always depend on the content class semantics.
        Though
…[3645 chars omitted]…
      """Request a channel close.

        This method indicates that the sender wants to close the
        channel. This may be due to internal conditions (e.g. a forced
        shut-down) or due to an error handling a specific method, i.e.
        an exception.  When a close is due to an exception, the sender
        provides the class and method id of the method which caused
        the exception.

/usr/local/lib/python3.12/site-packages/kombu/transport/pyamqp.py:224:    def qos_semantics_matches_spec(self, connection):
/usr/local/lib/python3.12/site-packages/kombu/messaging.py:387:    #: Can also be changed using :meth:`qos`.
/usr/local/lib/python3.12/site-packages/kombu/messaging.py:439:            self.qos(prefetch_count=self.prefetch_count)
/usr/local/lib/python3.12/site-packages/kombu/messaging.py:572:    def qos(self, prefetch_size=0, prefetch_count=0, apply_global=False):
/usr/local/lib/python3.12/site-packages/kombu/messaging.py:597:        return self.channel.basic_qos(prefetch_size,
/usr/local/lib/python3.12/site-packages/kombu/common.py:360:            e.g. ``consumer.qos`` or ``channel.basic_qos``.  Will be called
/usr/local/lib/python3.12/site-packages/kombu/common.py:372:        >>> qos = QoS(consumer.qos, initial_prefetch_count=2)
/usr/local/lib/python3.12/site-pa
…[592 chars omitted]…
b/python3.12/site-packages/kombu/common.py:392:        >>> QoS(channel.basic_qos, 10)
/usr/local/lib/python3.12/site-packages/kombu/common.py:395:        >>> def set_qos(prefetch_count):
/usr/local/lib/python3.12/site-packages/kombu/common.py:397:        >>> QoS(set_qos, 10)

/usr/local/lib/python3.12/site-packages/kombu/common.py:448:            logger.debug('basic.qos: prefetch_count->%s', new_value)

5.4.2
1833:    def basic_qos(self, prefetch_size, prefetch_count, a_global,
1840:        a qos method always depend on the content class semantics.
1841:        Though the qos method could in principle apply to both peers,

241:    def qos_semantics_matches_spec(self, connection):

Traceback (most recent call last):
  File "<stdin>", line 23, in <module>
  File "/repo/celery/worker/consumer/tasks.py", line 32, in start
    c.update_strategies()
  File "/repo/celery/worker/consumer/consumer.py", line 634, in update_strategies
    self.strategies[name] = task.start_strategy(self.app, self)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/app/task.py", line 431, in start_strategy
    return instantiate(self.Strategy, self, app, consumer, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/utils/imports.py", line 44, in instantiate
    return symbol_by_name(name)(*args, **kwargs)

           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/worker/strategy.py", line 120, in default
    call_at = consumer.timer.call_at
              ^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'call_at'

qos type <class 'kombu.transport.virtual.base.QoS'>
below True
at False

195:    def prepare_args(self, **kwargs):
196-        return kwargs
197-
198-    def _send_worker_shutdown(self):
199-        signals.worker_shutdown.send(sender=self)
200-
201-    def start(self):
202-        try:
203-            self.blueprint.start(self)
204-        except WorkerTerminate:
205-            self.terminate()
206-        except Exception as exc:

207-            logger.critical('Unrecoverable error: %r', exc, exc_info=True)
208-            self.stop(exitcode=EX_FAILURE)
209-        except SystemExit as exc:
210-            self.stop(exitcode=exc.code)
211-        except KeyboardInterrupt:
212-            self.stop(exitcode=EX_FAILURE)
213-
214-    def register_with_event_loop(self, hub):
215-        self.blueprint.send_all(
216-            self, 'register_with_event_loop', args=(hub,),
217-            description='hub.register',
218-        )
219-
220-    def _process_t
…[1933 chars omitted]…
   self.blueprint.stop(self, terminate=not warm)
267-                self.blueprint.join()
268-
269-    def reload(self, modules=None, reload=False, reloader=None):
270-        list(self._reload_modules(
271-            modules, force_reload=reload, reloader=reloader))
272-
273-        if self.consumer:
274-            self.consumer.update_strategies()
275-            self.consumer.reset_rate_limits()

186:    def apply(self, parent, **kwargs):
187-        """Apply the steps in this blueprint to an object.
188-
189-        This will apply the ``__init__`` and ``include`` methods
190-        of each step, with the object as argument::
191-
192-            step = Step(obj)
193-            ...
194-            step.include(obj)
195-
196-        For :class:`StartStopStep` the services created
197-        will also be added to the objects ``steps`` attribute.

198-        """
199-        self._debug('Preparing bootsteps.')
200-        order = self.order = []
201-        steps = self.steps = self.claim_steps()
202-
203-        self._debug('Building graph...')
204-        for S in self._finalize_steps(steps):
205-            step = S(parent, **kwargs)
206-            steps[step.name] = step
207-            order.append(step)
208-        self._debug('New boot order: {%s}',
209-               
…[1049 chars omitted]…
:
236-        last = self._find_last()
237-        self._firstpass(steps)
238-        it = ((C, C.requires) for C in steps.values())
239-        G = self.graph = DependencyGraph(
240-            it, formatter=self.GraphFormatter(root=last),
241-        )
242-        if last:
243-            for obj in G:
244-                if obj != last:
245-                    G.add_edge(last, obj)
246-        try:

def by_name(backend=None, loader=None,
            extension_namespace='celery.result_backends'):
    """Get backend class by name/alias."""
    backend = backend or 'disabled'
    loader = loader or current_app.loader
    aliases = dict(BACKEND_ALIASES, **loader.override_backends)
    aliases.update(load_extension_class_names(extension_namespace))
    try:
        cls = symbol_by_name(backend, aliases)
    except ValueError as exc:
        reraise(ImproperlyConfigured, ImproperlyConfigured(
            UNKNOWN_BACKEND.strip().format(backend, exc)), sys.exc_info()[2])

    if isinstance(cls, types.ModuleType):
        raise ImproperlyConfigured(UNKNOWN_BACKEND.strip().format(
            backend, 'is a Python module, not a backend class.'))
    return cls


def by_url(backend=None, loader=None):
    """Get backend class by URL."""
        yield from self.get_many(
            task_ids
…[1335 chars omitted]…
        return self.config_from_object(module_name, silent=silent, force=force)

    def config_from_cmdline(self, argv, namespace='celery'):
        self._conf.update(
            self.loader.cmdline_config_parser(argv, namespace)
        )

    def setup_security(self, allowed_serializers=None, key=None, key_password=None, cert=None,
                       store=None, digest=DEFAULT_SECURITY_DIGEST,

t/unit/app/test_app.py:1010:        assert module_name in exc.value.args[0]
t/unit/app/test_app.py:1011:        assert attr_name in exc.value.args[0]
t/unit/app/test_loaders.py:294:            assert exc.value.name == 'foo'
t/unit/app/test_loaders.py:307:            assert exc.value.msg == expected_error
t/unit/app/test_schedules.py:664:                if 'AssertionError' in str(exc):
t/unit/backends/test_base.py:1322:            assert 'foo' in str(exc)
t/unit/backends/test_base.py:1336:            assert 'Dependency culprit raised' in str(exc)
t/unit/tasks/test_trace.py:277:            traceback_clear(str(exc))
t/unit/worker/test_request.py:1017:                raise Retry(str(exc), exc=exc)

M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/components.py
 M celery/worker/consumer/consumer.py
 M celery/worker/consumer/tasks.py
 M celery/worker/worker.py

moved arg
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
151 passed, 1 skipped, 21 warnings, 37 subtests passed in 6.11s

moved worker arg
    warnings.warn(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
81 passed, 1 skipped, 3 warnings in 5.21s

User information: uid=0 euid=0 gid=0 egid=0
  
    warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
324 passed, 1 skipped, 25 warnings, 37 subtests passed in 13.61s
 M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/components.py

 M celery/worker/consumer/consumer.py
 M celery/worker/consumer/tasks.py
 M celery/worker/worker.py

all checks passed

diff --git a/celery/app/control.py b/celery/app/control.py
index 603d930..22ed192 100644
--- a/celery/app/control.py
+++ b/celery/app/control.py
@@ -20,7 +20,7 @@ from kombu.utils.compat import register_after_fork
 from kombu.utils.functional import lazy
 from kombu.utils.objects import cached_property
 
-from celery.exceptions import DuplicateNodenameWarning
+from celery.exceptions import DuplicateNodenameWarning, ImproperlyConfigured
 from celery.utils.log import get_logger
 from celery.utils.text import pluralize

 
@@ -428,6 +428,13 @@ class Control:
 
     def __init__(self, app=None):
         self.app = app
+        queue_durable = app.conf.control_queue_durable
+        queue_exclusive = app.conf.control_queue_exclusive
+        if queue_durable and queue_exclusive:
+            raise ImproperlyConfigured(
+                'control_queue_durable and control_queue_exclusive '
+ 
…[7144 chars omitted]…
file = logfile
@@ -407,6 +408,9 @@ class WorkController:
         self.disable_rate_limits = either(
             'worker_disable_rate_limits', disable_rate_limits,
         )
+        self.disable_prefetch = either(
+            'worker_disable_prefetch', disable_prefetch,
+        )
         self.worker_lost_wait = either('worker_lost_wait', worker_lost_wait)
 
     def wait_for_soft_shutdown(self):

/repo
CONTRIBUTING.rst
CONTRIBUTORS.txt
Changelog.rst
LICENSE
MANIFEST.in
Makefile
README.rst
SECURITY.md
TODO
bandit.json
celery

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

celery/worker/autoscale.py:45:            w.pool, w.max_concurrency, w.min_concurrency,
celery/worker/autoscale.py:64:    def __init__(self, pool, max_concurrency,
celery/worker/autoscale.py:70:        self.max_concurrency = max_concurrency
celery/worker/autoscale.py:85:        cur = min(self.qty, self.max_concurrency)
celery/worker/autoscale.py:104:                self.max_concurrency = max
celery/worker/autoscale.py:109:            return self.max_concurrency, self.min_concurrency
celery/worker/autoscale.py:134:        diff = new_max - self.max_concurrency
celery/worker/autoscale.py:142:            'max': self.max_concurrency,
celery/worker/components.py:108:        * max_concurrency
celery/worker/components.py:116:        w.max_concurrency = None
celery/worker/components.py:124:            w.max_concurrency, w.min_concurrency = w.autoscale
celery/worker/components.py:221:        if

…[621 chars omitted]…
worker/test_consumer.py:217:            bucket.can_consume.assert_called_with(3)
t/unit/worker/test_consumer.py:220:        bucket.can_consume.return_value = False
t/unit/worker/test_consumer.py:227:        bucket.can_consume.assert_called_with(4)
t/unit/worker/test_consumer.py:236:        bucket.can_consume.reset_mock()
t/unit/worker/test_consumer.py:238:        bucket.can_consume.assert_not_called()

celery/bin/worker.py:176:@click.option('--prefetch-multiplier',
celery/bin/worker.py:180:              value: value or ctx.obj.app.conf.worker_prefetch_multiplier,
celery/bin/worker.py:276:@click.option('--autoscale',
celery/bin/worker.py:312:    $ celery worker --autoscale=10,0
celery/bin/multi.py:31:    celery worker -n celery1@myhost -c 3
celery/bin/multi.py:32:    celery worker -n celery2@myhost -c 3
celery/bin/multi.py:33:    celery worker -n celery3@myhost -c 3
celery/bin/multi.py:37:    celery worker -n worker1@myhost -c 3
celery/bin/multi.py:38:    celery worker -n worker2@myhost -c 3
celery/bin/multi.py:39:    celery worker -n worker3@myhost -c 3
celery/bin/multi.py:43:    celery worker -n image@myhost -c 3
celery/bin/multi.py:44:    celery worker -n video@myhost -c 3

celery/bin/multi.py:45:    celery worker -n data@myhost -c 3
celery/bin/multi.py:49:    celery worker -n cele
…[1778 chars omitted]…
ltiplier = 4
t/unit/worker/test_consumer.py:89:        c.prefetch_multiplier = 10
t/unit/worker/test_control.py:44:        self.prefetch_multiplier = 1
t/unit/worker/test_control.py:353:        consumer.prefetch_multiplier = 8
t/unit/worker/test_worker.py:809:        # Given: a test celery worker instance with auto scaling
t/unit/worker/test_worker.py:858:        # Given: a test celery worker instance

"""WorkController can be used to instantiate in-process workers.

The command-line interface for the worker is in :mod:`celery.bin.worker`,
while the worker program is in :mod:`celery.apps.worker`.

The worker program is responsible for adding signal handlers,
setting up logging, etc.  This is a bare-bones worker without
global side-effects (i.e., except for the global state stored in
:mod:`celery.worker.state`).

The worker consists of several components, all managed by bootsteps
(mod:`celery.bootsteps`).

"""

import os
import sys
from datetime import datetime, timezone
from time import sleep

from billiard import cpu_count
from kombu.utils.compat import detect_environment

from celery import bootsteps
from celery import concurrency as _concurrency
from celery import signals
from celery.bootsteps import RUN, TERMINATE
from celery.exceptions import ImproperlyConfigured, TaskRevokedErr
…[8629 chars omitted]…
ck = None

    #: The current worker pool instance.
    pool = None

    #: A timer used for high-priority internal tasks, such
    #: as sending heartbeats.
    timer = None

    restart_count = -1  # first start is the same as a restart

    #: This flag will be turned off after the first failed
    #: connection attempt.
    first_connection_attempt = True

    class Blueprint(bootsteps.Blueprint):

try:
            if executable is not None:
                path = executable
            os.execv(path, [path] + argv)
            return EX_OK
        except Exception:  # pylint: disable=broad-except
            if app is None:
                from celery import current_app
                app = current_app
            app.log.setup_logging_subsystem(
                'ERROR', logfile, hostname=hostname)
            logger.critical("Can't exec %r", ' '.join([path] + argv),

                            exc_info=True)
            return EX_FAILURE


@click.command(cls=CeleryDaemonCommand,
               context_settings={'allow_extra_args': True})
@click.option('-n',
              '--hostname',
              default=host_format(default_nodename(None)),
              cls=CeleryOption,
              type=HOSTNAME,
              help_group="Worker Options",
              help="Set
…[3704 chars omitted]…
 replaced "
351:    def setup_defaults(self, concurrency=None, loglevel='WARN', logfile=None,
369:                       prefetch_multiplier=None, disable_rate_limits=None,
404:        self.prefetch_multiplier = int(either(
405:            'worker_prefetch_multiplier', prefetch_multiplier,
407:        self.disable_rate_limits = either(
408:            'worker_disable_rate_limits', disable_rate_limits,

t/unit/app/test_amqp.py:152:        self.app.conf.task_default_queue_type = default_queue_type
t/unit/app/test_amqp.py:162:            assert queue.queue_arguments == {'x-queue-type': 'quorum'}
t/unit/app/test_amqp.py:152:        self.app.conf.task_default_queue_type = default_queue_type
t/unit/app/test_amqp.py:162:            assert queue.queue_arguments == {'x-queue-type': 'quorum'}
t/unit/worker/test_consumer.py:654:        c.app.amqp.queues = {"celery": Mock(queue_arguments={"x-queue-type": "quorum"})}
t/unit/worker/test_consumer.py:689:        c.app.amqp.queues = {"celery": Mock(queue_arguments={"x-queue-type": "quorum"})}
t/unit/worker/test_consumer.py:697:        c.app.amqp.queues = {"celery": Mock(queue_arguments={"x-queue-type": "quorum"})}

from datetime import datetime, timedelta, timezone
from unittest.mock import Mock, patch

import pytest
from kombu import Exchange, Queue

from celery import uuid
from celery.app.amqp import Queues, utf8dict
from celery.utils.time import to_utc


class test_TaskConsumer:

    def test_accept_content(self, app):
        with app.pool.acquire(block=True) as con:
            app.conf.accept_content = ['application/json']
            assert app.amqp.TaskConsumer(con).accept == {
                'application/json',
            }
            assert app.amqp.TaskConsumer(con, accept=['json']).accept == {
                'application/json',
            }


class test_ProducerPool:

    def test_setup_nolimit(self, app):
        app.conf.broker_pool_limit = None
        try:
            delattr(app, '_pool')
        except AttributeError:
            pass
        app.amqp._producer_pool = None
…[8629 chars omitted]…
ern = pattern
        self.matcher = matcher

    def _prepare(self, reply):
        if reply:
            by_node = flatten_reply(reply)
            if (self.destination and
                    not isinstance(self.destination, (list, tuple))):
                return by_node.get(self.destination)
            if self.pattern:
                pattern = self.pattern
                matcher = self.matcher

bucket=Option(type='string'),
        base_path=Option(type='string'),
        endpoint_url=Option(type='string'),
        region=Option(type='string'),
    ),
    azureblockblob=Namespace(
        container_name=Option('celery', type='string'),
        retry_initial_backoff_sec=Option(2, type='int'),
        retry_increment_base=Option(2, type='int'),
        retry_max_attempts=Option(3, type='int'),
        base_path=Option('', type='string'),
        connection_timeout=Option(20, type='int'),

        read_timeout=Option(120, type='int'),
    ),
    gcs=Namespace(
        bucket=Option(type='string'),
        project=Option(type='string'),
        base_path=Option('', type='string'),
        ttl=Option(0, type='float'),
    ),
    control=Namespace(
        queue_ttl=Option(300.0, type='float'),
        queue_expires=Option(10.0, type='float'),
        exchange=Option('celer
…[3776 chars omitted]…
  acks_on_failure_or_timeout=Option(True, type='bool'),
        always_eager=Option(False, type='bool'),
        annotations=Option(type='any'),
233:    @cached_property
237:    @cached_property
566:    @cached_property
570:    @cached_property
571:    def queues(self):
576:    def queues(self, queues):
585:    @cached_property
602:    @cached_property
607:    @cached_property
611:    @cached_property

assert h.heartbeat_interval == 20
            assert c.heart is None

            h.start(c)
            assert c.heart
            hcls.assert_called_with(c.timer, c.event_dispatcher,
                                    h.heartbeat_interval)
            c.heart.start.assert_called_with()


class test_Tasks:

    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
…[3837 chars omitted]…
unit/app/test_control.py:565:        c = control.Control(self.app)
t/unit/app/test_control.py:251:        assert control.Control(self.app).mailbox.serializer == 'test'
t/unit/app/test_control.py:252:        assert control.Control(self.app).mailbox.accept == ['test']
t/unit/app/test_control.py:560:        c = control.Control(self.app)
t/unit/app/test_control.py:565:        c = control.Control(self.app)

celery/app/amqp.py:57:                 create_missing=True, autoexchange=None,
celery/app/amqp.py:64:        self.autoexchange = Exchange if autoexchange is None else autoexchange
celery/app/amqp.py:87:            return self.add(self.new_missing(name))
celery/app/amqp.py:183:    def new_missing(self, name):
celery/app/amqp.py:184:        return Queue(name, self.autoexchange(name), name)
celery/app/amqp.py:213:    # For example, you can use ``autoexchange = lambda n: None`` to use the
celery/app/amqp.py:216:    autoexchange = None
celery/app/amqp.py:242:               autoexchange=None, max_priority=None):
celery/app/amqp.py:259:        autoexchange = (self.autoexchange if autoexchange is None
celery/app/amqp.py:260:                        else autoexchange)
celery/app/amqp.py:263:            autoexchange, max_priority, default_routing_key,
class QoS:

    """Thread safe increment/decr
…[3977 chars omitted]…

        self.tag_prefix = tag_prefix
        self._active_tags = {}
        if auto_declare is not None:
            self.auto_declare = auto_declare
        if on_decode_error is not None:
            self.on_decode_error = on_decode_error
        self.accept = prepare_accept_content(accept)
        self.prefetch_count = prefetch_count

        if self.channel:
            self.revive(self.channel)

Exchange a(direct) direct <class 'kombu.entity.Exchange'>
(name='', type='', channel=None, **kwargs)
    def __init__(self, name='', type='', channel=None, **kwargs):
        super().__init__(**kwargs)
        self.name = name or self.name
        self.type = type or self.type
        self.maybe_bind(channel)

136:``CELERY_CREATE_MISSING_QUEUES``           :setting:`task_create_missing_queues`
139:``CELERY_DEFAULT_EXCHANGE_TYPE``           :setting:`task_default_exchange_type`
141:``CELERY_DEFAULT_QUEUE_TYPE``              :setting:`task_default_queue_type`
2611:.. setting:: task_create_missing_queues
2613:``task_create_missing_queues``
2641:.. setting:: task_default_queue_type
2643:``task_default_queue_type``
2679:.. setting:: task_default_exchange_type
2681:``task_default_exchange_type``

qos <function Consumer.qos at 0x74e2a11013a0>
    def qos(self, prefetch_size=0, prefetch_count=0, apply_global=False):
        """Specify quality of service.

        The client can request that messages should be sent in
        advance so that when the client finishes processing a message,
        the following message is already held locally, rather than needing
        to be sent down the channel. Prefetching gives a performance
        improvement.

        The prefetch window is Ignored if the :attr:`no_ack` option is set.

        Arguments:
        ---------
            prefetch_size (int): Specify the prefetch window in octets.
                The server will send a message in advance if it is equal to
                or smaller in size than the available prefetch size (and
                also falls within other prefetch limits). May be set to zero,
                meaning 
…[38 chars omitted]…
efetch limits
                may still apply.

            prefetch_count (int): Specify the prefetch window in terms of
                whole messages.

            apply_global (bool): Apply new settings globally on all channels.
        """
        return self.channel.basic_qos(prefetch_size,
                                      prefetch_count,
                                      apply_global)

99:    def __init__(self, steps=None, name=None,
186:    def apply(self, parent, **kwargs):
319:    def __init__(self, parent, **kwargs):

def apply(self, parent, **kwargs):
        """Apply the steps in this blueprint to an object.

        This will apply the ``__init__`` and ``include`` methods
        of each step, with the object as argument::

            step = Step(obj)
            ...
            step.include(obj)

        For :class:`StartStopStep` the services created
        will also be added to the objects ``steps`` attribute.

        """
        self._debug('Preparing bootsteps.')
        order = self.order = []
        steps = self.steps = self.claim_steps()

        self._debug('Building graph...')
        for S in self._finalize_steps(steps):
            step = S(parent, **kwargs)
            steps[step.name] = step
            order.append(step)
        self._debug('New boot order: {%s}',
                    ', '.join(s.alias for s in self.order))
        for step in order:
            step.include
…[3270 chars omitted]…
tep):
    """Bootstep that must be started and stopped in order."""

    #: Optional obj created by the :meth:`create` method.
    #: This is used by :class:`StartStopStep` to keep the
    #: original service object.
    obj = None

    def start(self, parent):
        if self.obj:
            return self.obj.start()

    def stop(self, parent):
        if self.obj:
            return self.obj.stop()

get_queue
    def get_queue(self, hostname):
        return Queue(
            f'{hostname}.{self.namespace}.pidbox',
            exchange=self.exchange,
            durable=self.queue_durable,
            exclusive=self.queue_exclusive,
            auto_delete=not self.queue_durable,
            expires=self.queue_expires,
            message_ttl=self.queue_ttl,
        )

 get_reply_queue
    def get_reply_queue(self):
        oid = self.oid
        return Queue(
            f'{oid}.{self.reply_exchange.name}',
            exchange=self.reply_exchange,
            routing_key=oid,
            durable=self.queue_durable,
            exclusive=self.queue_exclusive,
            auto_delete=not self.queue_durable,
            expires=self.reply_queue_expires,
            message_ttl=self.reply_queue_ttl,
        )

bash: line 124: warning: here-document at line 121 delimited by end-of-file (wanted `PY')
/usr/local/lib/python3.12/site-packages/kombu/common.py
5.6.2
bash: line 121: rg: command not found

grep: /usr/local/lib/python3.12/site-packages/kombu/__pycache__/mixins.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/kombu/utils/__pycache__/limits.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/kombu/transport/virtual/__pycache__/base.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/redis.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/SQS.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/gcpubsub.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/confluentkafka.cpython-312.pyc: binary file matches
grep: /usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__
…[2388 chars omitted]…
/transport/qpid.py:195:    def can_consume(self):
/usr/local/lib/python3.12/site-packages/kombu/transport/qpid.py:203:            can_consume will always return True.
/usr/local/lib/python3.12/site-packages/kombu/transport/qpid.py:212:    def can_consume_max_estimate(self):

/usr/local/lib/python3.12/site-packages/kombu/transport/gcpubsub.py:480:        max_allowed = self.qos.can_consume_max_estimate()

bindings = None

    #: The queue index is used to access directly (constant time)
    #: all the bindings of a certain queue.  It has the following structure::
    #:
    #:     {
    #:         queue: {
    #:             (queue, exchange, routing_key),
    #:             # ...,
    #:         },
    #:         # ...,
    #:     }

    queue_index = None

    def __init__(self, exchanges=None):
        self.exchanges = {} if exchanges is None else exchanges
        self.bindings = {}
        self.queue_index = defaultdict(set)

    def clear(self):
        self.exchanges.clear()
        self.bindings.clear()
        self.queue_index.clear()

    def has_binding(self, queue, exchange, routing_key):
        return (queue, exchange, routing_key) in self.bindings

    def binding_declare(self, queue, exchange, routing_key, arguments):
        key = binding_key_t(queue, exchange, rout
…[7154 chars omitted]…
ings[key]
        except KeyError:
            pass
        else:
            self.queue_index[queue].remove(key)

    def queue_bindings_delete(self, queue):
        try:
            bindings = self.queue_index.pop(queue)
        except KeyError:
            pass
        else:
            [self.bindings.pop(binding, None) for binding in bindings]

    def queue_bindings(self, queue):
        return (

def __init__(self, name='', exchange=None, routing_key='',
                 channel=None, bindings=None, on_declared=None,
                 **kwargs):
        super().__init__(**kwargs)
        self.name = name or self.name
        if isinstance(exchange, str):
            self.exchange = Exchange(exchange)
        elif isinstance(exchange, Exchange):
            self.exchange = exchange
        self.routing_key = routing_key or self.routing_key
        self.bindings = set(bindings or [])
        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)

206:class ImproperlyConfigured(CeleryError):
"""Celery error types.

Error Hierarchy
===============

- :exc:`Exception`
    - :exc:`celery.exceptions.CeleryError`
        - :exc:`~celery.exceptions.ImproperlyConfigured`
        - :exc:`~celery.exceptions.SecurityError`
        - :exc:`~celery.exceptions.TaskPredicate`
            - :exc:`~celery.exceptions.Ignore`

            - :exc:`~celery.exceptions.Reject`
            - :exc:`~celery.exceptions.Retry`
        - :exc:`~celery.exceptions.TaskError`
            - :exc:`~celery.exceptions.QueueNotFound`
            - :exc:`~celery.exceptions.IncompleteStream`
            - :exc:`~celery.exceptions.NotRegistered`
            - :exc:`~celery.exceptions.AlreadyRegistered`
            - :exc:`~celery.exceptions.TimeoutError`
            - :exc:`~celery.exceptions.MaxRetriesExceededError`
            - :exc:`~celery.exceptions.TaskRevoked
…[5710 chars omitted]…
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

``control_exchange``
~~~~~~~~~~~~~~~~~~~~~~

Default: ``"celery"``.

Name of the control command exchange.

.. warning::

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

.. _conf-logging:

Logging
-------

"""A task can raise this if it wants to reject/re-queue the message."""

    def __init__(self, reason=None, requeue=False):
        self.reason = 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."""


class QueueNotFound(KeyError, TaskError):
    """Task routed to a queue not in ``conf.queues``."""

"""Monitoring Event Receiver+Dispatcher.

Events is a stream of messages sent for certain actions occurring
in the worker (and clients if :setting:`task_send_sent_event`
is enabled), used for monitoring purposes.
"""

from .dispatcher import EventDispatcher
from .event import Event, event_exchange, get_exchange, group_from
from .receiver import EventReceiver

__all__ = (

    'Event', 'EventDispatcher', 'EventReceiver',
    'event_exchange', 'get_exchange', 'group_from',
)

39:class StepFormatter(GraphFormatter):
40-    """Graph formatter for :class:`Blueprint`."""
41-
42-    blueprint_prefix = '⧉'
43-    conditional_prefix = '∘'
44-    blueprint_scheme = {
45-        'shape': 'parallelogram',
46-        'color': 'slategray4',
47-        'fillcolor': 'slategray3',
48-    }
49-
50-    def label(self, step):

51-        return step and '{}{}'.format(
52-            self._get_prefix(step),
53-            bytes_to_str(
54-                (step.label or _label(step)).encode('utf-8', 'ignore')),
55-        )
56-
57-    def _get_prefix(self, step):
58-        if step.last:
59-            return self.blueprint_prefix
60-        if step.conditional:
61-            return self.conditional_prefix
62-        return ''
63-
64-    def node(self, obj, **attrs):
65-        scheme = self.blueprint_scheme if obj.last else self.node_scheme
66-        return self.draw_node(o
…[7454 chars omitted]…
 """Initialize a Celery command with common daemon options."""
196-        super().__init__(*args, **kwargs)
197-        self.params.extend((
198-            DaemonOption("--logfile", "-f", help="Log destination; defaults to stderr"),
199-            DaemonOption("--pidfile", help="PID file path; defaults to no PID file"),
200-            DaemonOption("--uid", help="Drops privileges to this user ID"),

celery/backends/cassandra.py:33:E_CASSANDRA_MISCONFIGURED = 'Cassandra backend improperly configured.'
celery/exceptions.py:207:    """Celery is somehow improperly configured."""

t/unit/app/test_amqp.py:75:        assert Queues(None) == {}
t/unit/app/test_amqp.py:78:        q = Queues()
t/unit/app/test_amqp.py:85:        q = Queues(default_exchange=Exchange('bar'))
t/unit/app/test_amqp.py:93:        q = Queues()
t/unit/app/test_amqp.py:99:        q = Queues()
t/unit/app/test_amqp.py:106:        q = Queues(default_exchange=ex)
t/unit/app/test_amqp.py:111:        q = Queues()
t/unit/app/test_amqp.py:133:        queues = Queues(**queues_kwargs)
t/unit/app/test_amqp.py:240:        x = self.app.amqp.Queues({}, max_priority=23)
t/unit/app/test_routes.py:25:    app.amqp.queues = app.amqp.Queues(queues)
t/unit/app/test_routes.py:105:        expand = E(self.app, self.app.amqp.Queues(
t/unit/utils/test_text.py:43:        app.amqp.queues = app.amqp.Queues(QUEUES)

./t/unit/app/test_amqp.py:240:        x = self.app.amqp.Queues({}, max_priority=23)
./t/unit/app/test_routes.py:25:    app.amqp.queues = app.amqp.Queues(queues)
./t/unit/app/test_routes.py:105:        expand = E(self.app, self.app.amqp.Queues(
./t/unit/utils/test_text.py:43:        app.amqp.queues = app.amqp.Queues(QUEUES)

def _patch_thread_primitives(self, w):
        # make clock use dummy lock
        w.app.clock.mutex = DummyLock()
        # multiprocessing's ApplyResult uses this lock.
        try:
            from billiard import pool
        except ImportError:
            pass
        else:
            pool.Lock = DummyLock

class Pool(bootsteps.StartStopStep):
    """Bootstep managing the worker pool.

    Describes how to initialize the worker pool, and starts and stops
    the pool during worker start-up/shutdown.

    Adds attributes:

        * autoscale
        * pool
        * max_concurrency
        * min_concurrency
    """

    requires = (Hub,)

    def __init__(self, w, autoscale=None, **kwargs):
        w.pool = None
        w.max_concurrency = None
        w.min_concurrency = w.concurrency
        self.optimization = w.optimization
        if isinstance(autoscale, str):
     
…[5639 chars omitted]…
        controller=w,
            hub=w.hub,
            worker_options=w.options,
            disable_rate_limits=w.disable_rate_limits,
            prefetch_multiplier=w.prefetch_multiplier,
        )
        return c
celery/concurrency/base.py:179:    def num_processes(self):
celery/concurrency/gevent.py:144:    def num_processes(self):
celery/concurrency/prefork.py:171:    def num_processes(self):

confirmed missing queue options: Queues.__init__() got an unexpected keyword argument 'create_missing_queue_type'
confirmed app queue options are missing: AMQP.Queues() got an unexpected keyword argument 'create_missing_queue_type'

56:    def __init__(self, queues=None, default_exchange=None,
57:                 create_missing=True, autoexchange=None,
61:        self.default_exchange = default_exchange
64:        self.autoexchange = Exchange if autoexchange is None else autoexchange
79:        if self.default_exchange and not queue.exchange:
80:            queue.exchange = self.default_exchange
120:            queue.exchange = self.default_exchange
184:        return Queue(name, self.autoexchange(name), name)
213:    # For example, you can use ``autoexchange = lambda n: None`` to use the
216:    autoexchange = None
225:    def __init__(self, app):
242:               autoexchange=None, max_priority=None):

256:                            exchange=self.default_exchange,
259:        autoexchange = (self.autoexchange if autoexchange is None
260:                        else autoexchange)
262:            queues, self.d
…[2988 chars omitted]…
                                      create_missing), app=self.app)
celery/app/amqp.py:64:        self.autoexchange = Exchange if autoexchange is None else autoexchange
celery/app/amqp.py:213:    # For example, you can use ``autoexchange = lambda n: None`` to use the
celery/app/amqp.py:216:    autoexchange = None
celery/app/amqp.py:259:        autoexchange = (self.autoexchange if autoexchange is None

18:    def Receiver(self):
19-        return self.app.subclass_with_self(
20-            self.receiver_cls, reverse='events.Receiver')
21-
22-    @cached_property
23-    def Dispatcher(self):
24-        return self.app.subclass_with_self(
25-            self.dispatcher_cls, reverse='events.Dispatcher')
26-
27-    @cached_property
28-    def State(self):
29-        return self.app.subclass_with_self(

30-            self.state_cls, reverse='events.State')
docs/userguide/configuration.rst:172:``CELERYD_PREFETCH_MULTIPLIER``            :setting:`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.

celery/app/amqp.py              | 26 +++++++++++++++++++++++---
 celery/app/control.py           | 10 +++++++++-
 celery/app/defaults.py          |  7 +++++++
 celery/bin/worker.py            |  5 +++++
 celery/events/receiver.py       | 10 +++++++++-
 celery/worker/consumer/tasks.py | 20 +++++++++++++++++++-
 6 files changed, 72 insertions(+), 6 deletions(-)
diff --git a/celery/app/amqp.py b/celery/app/amqp.py
index 8dcec36..a677fe2 100644
--- a/celery/app/amqp.py
+++ b/celery/app/amqp.py
@@ -55,7 +55,9 @@ class Queues(dict):

 
     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'
…[8628 chars omitted]…
             )
+                limit = max_concurrency or c.pool.num_processes
+                if len(state.reserved_requests) >= limit:
+                    return False
+                return original_can_consume()
+
+            qos.can_consume = can_consume
+
         def set_prefetch_count(prefetch_count):
             return c.task_consumer.qos(
                 prefetch_count=prefetch_count,

programmatic queue: spontaneous {'x-queue-type': 'quorum'} topic
configured queue: auto None direct
control durability: True False
control invalid combination: validated
event exclusive: False True True
event durable: True False False
event invalid combination: validated

auto {'x-queue-type': 'quorum'} topic
auto {'x-queue-type': 'quorum'} topic
........................................................................ [ 30%]
........................................................................ [ 60%]
....................,,,.,,,.,,,.,,,.,,,.,,,..,,,......................., [ 81%]
,..,..,,..,.,,.,,.,,.,,.,,..................................             [100%]
239 passed, 22 warnings, 37 subtests passed in 8.07s

['__copy__']
class Exchange(MaybeChannelBound):
    """An Exchange declaration.

    Arguments:
    ---------
        name (str): See :attr:`name`.
        type (str): See :attr:`type`.
        channel (kombu.Connection, ChannelT): See :attr:`channel`.
        durable (bool): See :attr:`durable`.
        auto_delete (bool): See :attr:`auto_delete`.
        delivery_mode (enum): See :attr:`delivery_mode`.

        arguments (Dict): See :attr:`arguments`.
        no_declare (bool): See :attr:`no_declare`

    Attributes
    ----------
        name (str): Name of the exchange.
            Default is no name (the default exchange).

        type (str):
            *This description of AMQP exchange types was shamelessly stolen
            from the blog post `AMQP in 10 minutes: Part 4`_ by
            Rajith Attapattu. Reading this article is recommended if you're
            new to amqp.*
…[8628 chars omitted]…
= other.delivery_mode)
        return NotImplemented

    def __ne__(self, other):
        return not self.__eq__(other)

    def __repr__(self):
        return self._repr_entity(self)

    def __str__(self):
        return 'Exchange {}({})'.format(
            _reprstr(self.name) or repr(''), self.type,
        )

    @property
    def can_cache_declaration(self):
        return not self.auto_delete

def __copy__(self: _ObjectType) -> _ObjectType:
        return self.__class__(**self.as_dict())

    def as_dict(self, recurse: bool = False) -> dict[str, Any]:
        def f(obj: Any, type: Callable[[Any], Any] | None = None) -> Any:
            if recurse 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
        }

{'arguments': None, 'name': 'a', 'type': 'direct'}

diff --git a/celery/app/amqp.py b/celery/app/amqp.py
index 8dcec36..a677fe2 100644
--- a/celery/app/amqp.py
+++ b/celery/app/amqp.py
@@ -55,7 +55,9 @@ class Queues(dict):
 
     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 = WeakValueDictionary()
         self.default_exchange = default_exchange
@@ -63,6 +65,8 @@ class Queues(dict):
         self.create_missing = create_missing
         self.autoexchange = Exchange if autoexchange is None else autoexchange
         self.max_priority = max_priority
+        self.create_missing_q
…[8628 chars omitted]…

+                    return False
+                return original_can_consume()
+
+            qos.can_consume = can_consume
+
         def set_prefetch_count(prefetch_count):
             return c.task_consumer.qos(
                 prefetch_count=prefetch_count,
Traceback (most recent call last):
  File "<stdin>", line 17, in <module>
AttributeError: 'method' object has no attribute 'return_value'

['__copy__']
<class 'AttributeError'> 'Exchange' object has no attribute 'copy'
<class 'kombu.exceptions.NotBoundError'> Can't call method on Exchange not bound to a channel

def apply_async(self, target, args=None, kwargs=None, **options):
        """Equivalent of the :func:`apply` built-in function.

        Callbacks should optimally return as soon as possible since
        otherwise the thread which handles the result will get blocked.
        """
        kwargs = {} if not kwargs else kwargs
        args = [] if not args else args
        if self._does_debug:
            logger.debug('TaskPool: Apply %s (args:%s kwargs:%s)',
                         target, truncate(safe_repr(args), 1024),
                         truncate(safe_repr(kwargs), 1024))

        return self.on_apply(target, args, kwargs,
                             waitforslot=self.putlocks,
                             callbacks_propagate=self.callbacks_propagate,
                             **options)

    def _get_info(self) -> Dict[str, Any]:
        """
        Return configurat
…[1772 chars omitted]…
_make_killable_target(target):
        def killable_target(*args, **kwargs):
            from greenlet import GreenletExit
            try:
                return target(*args, **kwargs)
            except GreenletExit:
                return (False, None, None)

        return killable_target

    def _add_to_pool_map(self, pid, greenlet):
        self._pool_map[pid] = greenlet
        greenlet.link(

setup.cfg:6:[flake8]
setup.cfg:9:max-line-length = 117
tox.ini:9:    flake8

unbound <class 'kombu.exceptions.NotBoundError'> Can't call method on Exchange not bound to a channel
bound <class 'kombu.exceptions.NotBoundError'> Can't call method on Exchange not bound to a channel

disable-prefetch reserved limits: passed
enabled-off callable untouched: passed

@classmethod
    def from_dict(cls, queue, **options):
        binding_key = options.get('binding_key') or options.get('routing_key')

        e_durable = options.get('exchange_durable')
        if e_durable is None:
            e_durable = options.get('durable')

        e_auto_delete = options.get('exchange_auto_delete')
        if e_auto_delete is None:
            e_auto_delete = options.get('auto_delete')

        q_durable = options.get('queue_durable')
        if q_durable is None:
            q_durable = options.get('durable')

        q_auto_delete = options.get('queue_auto_delete')
        if q_auto_delete is None:
            q_auto_delete = options.get('auto_delete')

        e_arguments = options.get('exchange_arguments')
        q_arguments = options.get('queue_arguments')
        b_arguments = options.get('binding_arguments')
        c_arguments = options.get('consu
…[747 chars omitted]…
  no_ack=options.get('no_ack'),
                     queue_arguments=q_arguments,
                     binding_arguments=b_arguments,
                     consumer_arguments=c_arguments,
                     bindings=bindings)

Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
AttributeError: type object 'Queue' has no attribute '_get_exchange'. Did you mean: '_create_exchange'?

dict {'arguments': None, 'name': 'a', 'type': 'direct'}
asdict {'name': 'a', 'type': 'direct', 'arguments': None, 'durable': True, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}
('name', None) a
('type', None) direct
('arguments', None) None
('durable', <class 'bool'>) True
('passive', <class 'bool'>) False
('auto_delete', <class 'bool'>) False
('delivery_mode', <function Exchange.<lambda> at 0x7a91700d7ce0>) None
('no_declare', <class 'bool'>) False

-O, --optimization [default|fair]
                                  Apply optimization profile.
  --disable-prefetch              Disable prefetching tasks from the broker.
  --prefetch-multiplier <prefetch multiplier>
                                  Set custom prefetch multiplier value for

as_dict {'name': 'a', 'type': 'headers', 'arguments': {'x': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}
exchange attrs a headers False False False None False {'x': 1}
as_dict {'name': 'a', 'type': 'headers', 'arguments': {'x': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}
exchange attrs a headers False False False None False {'x': 1}

M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/consumer/tasks.py

error <class 'kombu.exceptions.NotBoundError'> Can't call method on Exchange not bound to a channel
error <class 'kombu.exceptions.NotBoundError'> Can't call method on Exchange not bound to a channel

{'name': 'x', 'type': 'direct', 'arguments': None, 'durable': True, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}
Exchange x(direct)

False
['basic_qos']
class Channel(amqp.Channel, base.StdChannel):
    """AMQP Channel."""

    Message = Message

    def prepare_message(self, body, priority=None,
                        content_type=None, content_encoding=None,
                        headers=None, properties=None, _Message=amqp.Message):
        """Prepare message so that it can be sent using this transport."""
        return _Message(

            body,
            priority=priority,
            content_type=content_type,
            content_encoding=content_encoding,
            application_headers=headers,
            **properties or {}
        )

    def prepare_queue_arguments(self, arguments, **kwargs):
        return to_rabbitmq_queue_arguments(arguments, **kwargs)

    def message_to_python(self, raw_message):
        """Convert encoded message body back to a Python value."""
        return self.Message(raw_message, channel=self)

grep: /usr/local/lib/python3.12/site-packages/kombu/transport/__pycache__/qpid.cpython-312.pyc: binary file matches
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py:638:                self.qos.append(message, message.delivery_tag)
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py:663:                self.qos.append(message, message.delivery_tag)
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py:670:        self.qos.ack(delivery_tag)
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py:675:            return self.qos.restore_unacked()
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py:680:        self.qos.reject(delivery_tag, requeue=requeue)
/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py:690:        self.qos.prefetch_count = prefetch_count
/usr/local/lib
…[1551 chars omitted]…
/usr/local/lib/python3.12/site-packages/kombu/transport/azureservicebus.py:138:        self.qos.restore_at_shutdown = False
/usr/local/lib/python3.12/site-packages/kombu/transport/azureservicebus.py:293:            delivery_info = self.qos.get(delivery_tag).delivery_info
/usr/local/lib/python3.12/site-packages/kombu/transport/SLMQ.py:123:        delivery_info = self.qos.get(delivery_tag).delivery_info

<function worker at 0x703319b3c2c0>
(ctx, hostname=None, pool_cls=None, app=None, uid=None, gid=None, loglevel=None, logfile=None, pidfile=None, statedb=None, **kwargs)
@click.command(cls=CeleryDaemonCommand,
               context_settings={'allow_extra_args': True})
@click.option('-n',
              '--hostname',
              default=host_format(default_nodename(None)),
              cls=CeleryOption,
              type=HOSTNAME,
              help_group="Worker Options",
              help="Set custom hostname (e.g., 'w1@%%h').  "
                   "Expands: %%h (hostname), %%n (name) and %%d, (domain).")

@click.option('-D',
              '--detach',
              cls=CeleryOption,
              is_flag=True,
              default=False,
              help_group="Worker Options",
              help="Start worker as a background process.")
@click.option('-S',
              '--stat
…[7597 chars omitted]…
 logfile=logfile,  # node format handled by celery.app.log.setup
            pidfile=node_format(pidfile, hostname),
            statedb=node_format(statedb, hostname),
            no_color=ctx.obj.no_color,
            quiet=ctx.obj.quiet,
            **kwargs)
        worker.start()
        ctx.exit(worker.exitcode)
    except SecurityError as e:
        ctx.obj.error(e.args[0])
        ctx.exit(1)

0 None
callback keyword: True

"""Pool Autoscaling.

This module implements the internal thread responsible
for growing and shrinking the pool according to the
current autoscale settings.

The autoscale thread is only enabled if
the :option:`celery worker --autoscale` option is used.
"""
import os
import threading
from time import monotonic, sleep

from kombu.asynchronous.semaphore import DummyLock

from celery import bootsteps
from celery.utils.log import get_logger
from celery.utils.threads import bgThread

from . import state
from .components import Pool

__all__ = ('Autoscaler', 'WorkerComponent')

logger = get_logger(__name__)
debug, info, error = logger.debug, logger.info, logger.error

AUTOSCALE_KEEPALIVE = float(os.environ.get('AUTOSCALE_KEEPALIVE', 30))


class WorkerComponent(bootsteps.StartStopStep):
    """Bootstep that starts the autoscaler thread/timer in the worker."""

    label = 'Autoscaler'
    c
…[6214 chars omitted]…
at process.

    Enabled when the ``beat`` argument is set.
    """

    label = 'Beat'
    conditional = True
celery/worker/autoscale.py:70:        self.max_concurrency = max_concurrency
celery/worker/autoscale.py:104:                self.max_concurrency = max
celery/worker/components.py:116:        w.max_concurrency = None
celery/worker/consumer/tasks.py:52:                max_concurrency = getattr(

d {'name': 'x', 'type': 'direct', 'arguments': None, 'durable': True, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}
ok Exchange x(direct)

dict {'arguments': {'a': 1}, 'durable': False, 'name': 'x', 'type': 'headers'} asdict {'name': 'x', 'type': 'headers', 'arguments': {'a': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}
copy Exchange x(headers) {'arguments': {'a': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'no_declare': False, 'name': 'x', 'type': 'headers'}
copy-method Exchange x(headers) {'arguments': {'a': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'no_declare': False, 'name': 'x', 'type': 'headers'}
dict {'arguments': {'a': 1}, 'durable': False, 'name': 'x', 'type': 'headers', '_channel': <kombu.transport.memory.Channel object at 0x719dd78a91f0>, '_is_bound': True} asdict {'name': 'x', 'type': 'headers', 'arguments': {'a': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'delivery_mode': None, 'no_declare': False}
copy Exchange x(headers) {'arguments': {'a': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'no_declare': False, 'name': 'x', 'type': 'headers'}
copy-method Exchange x(headers) {'arguments': {'a': 1}, 'durable': False, 'passive': False, 'auto_delete': False, 'no_declare': False, 'name': 'x', 'type': 'headers'}

automatic queue exchange is independently configured: passed

import sys
from importlib import import_module

from celery.app.defaults import (_OLD_DEFAULTS, _OLD_SETTING_KEYS, _TO_NEW_KEY, _TO_OLD_KEY, DEFAULTS, NAMESPACES,
                                 SETTING_KEYS)


class test_defaults:

    def setup_method(self):
        self._prev = sys.modules.pop('celery.app.defaults', None)

    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
…[2608 chars omitted]…
ore_result=Option(False, type='bool'),
        store_eager_result=Option(False, type='bool'),
        protocol=Option(2, type='int', old={'celery_task_protocol'}),
        publish_retry=Option(
            True, type='bool', old={'celery_task_publish_retry'},
        ),
        publish_retry_policy=Option(
            {'max_retries': 3,
             'interval_start': 0,
             'interval_max': 1,

96:def task_reserved(request,
97-                  add_request=requests.__setitem__,
98-                  add_reserved_request=reserved_requests.add):
99-    """Update global state when a task has been reserved."""
100-    add_request(request.id, request)
101-    add_reserved_request(request)
102-
103-
104:def task_accepted(request,
105-                  _all_total_count=None,
106-                  add_request=requests.__setitem__,
107-                  add_active_request=active_requests.add,

108-                  add_to_total_count=total_count.update):
109-    """Update global state when a task has been accepted."""
110-    if not _all_total_count:
111-        _all_total_count = all_total_count
112-    add_request(request.id, request)
113-    add_active_request(request)
114-    add_to_total_count({request.name: 1})
115-    all_total_count[0] += 1
116-
117-
118:def task_ready(request,
…[3473 chars omitted]…

220-
221-    def sync(self):
222-        self._sync_with(self.db)
223-        self.db.sync()
224-
celery/worker/consumer/consumer.py:289:        task_reserved(request)
celery/worker/consumer/consumer.py:574:        task_reserved(task)
celery/worker/state.py:96:def task_reserved(request,
celery/worker/state.py:162:    def task_reserved(request):
celery/worker/strategy.py:204:        task_reserved(req)

Default: :const:`False`.

If enabled, child tasks will inherit priority of the parent task.

.. code-block:: python

    # The last task in chain will also have priority set to 5.
    chain = celery.chain(add.s(2) | add.s(2).set(priority=5) | add.s(3))

Priority inheritance also works when calling child tasks from a parent task
with `delay` or `apply_async`.

See :ref:`routing-options-rabbitmq-priorities`.


.. setting:: worker_direct

``worker_direct``
~~~~~~~~~~~~~~~~~

Default: Disabled.

This option enables so that every worker has a dedicated queue,
so that tasks can be routed to specific workers.

The queue name for each worker is automatically generated based on
the worker hostname and a ``.dq`` suffix, using the ``C.dq2`` exchange.

For example the queue name for the worker with node name ``w1@example.com``
becomes::

    w1@example.com.dq

Then you can route the task to the 
…[8628 chars omitted]…
lso applies to remote control reply queues.

.. setting:: control_exchange

``control_exchange``
~~~~~~~~~~~~~~~~~~~~~~

Default: ``"celery"``.

Name of the control command 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``
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

........................................................................ [ 29%]
........................................................................ [ 59%]
........................,,,.,,,.,,,.,,,.,,,.,,,..,,,.................... [ 80%]
...,,..,..,,..,.,,.,,.,,.,,.,,..................................         [100%]
243 passed, 22 warnings, 37 subtests passed in 7.90s

celery/app/amqp.py              | 35 ++++++++++++++++++++++++++++++++---
 celery/app/control.py           | 10 +++++++++-
 celery/app/defaults.py          |  7 +++++++
 celery/bin/worker.py            |  5 +++++
 celery/events/receiver.py       | 10 +++++++++-
 celery/worker/consumer/tasks.py | 20 +++++++++++++++++++-
 6 files changed, 81 insertions(+), 6 deletions(-)
diff --git a/celery/app/amqp.py b/celery/app/amqp.py
index 8dcec36..7f33cfd 100644
--- a/celery/app/amqp.py
+++ b/celery/app/amqp.py
@@ -1,5 +1,6 @@

 """Sending/Receiving Messages (Kombu integration)."""
 import numbers
+from copy import copy
 from collections import namedtuple
 from collections.abc import Mapping
 from datetime import timedelta
@@ -55,7 +56,9 @@ class Queues(dict):
 
     def __init__(self, queues=None, default_exchange=None,
                  create_missing=True, autoexchange=None,
-                 m
…[8629 chars omitted]…
eturn False
+                return original_can_consume()
+
+            qos.can_consume = can_consume
+
         def set_prefetch_count(prefetch_count):
             return c.task_consumer.qos(
                 prefetch_count=prefetch_count,
 M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/consumer/tasks.py

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

from kombu import Connection, Consumer, Exchange, Producer, Queue, pools
from kombu.common import Broadcast
from kombu.utils.functional import maybe_list
from kombu.utils.objects import cached_property

from celery import signals
from celery.utils.nodenames import anon_nodename
from celery.utils.saferepr import saferepr

"""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

from kombu import Connection, Consumer, Exchange, Producer, Queue, pools
from kombu.common import Broadcast
from kombu.utils.functional import maybe_list
from kombu.utils.objects import cached_property

    """

    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()
       
…[916 chars omitted]…
eue_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
        if accept is None:
            accept = {self.app.conf.event_serializer, 'json'}
        self.accept = accept

    def process(self, type, event):

amqp channel class <class 'amqp.channel.Channel'>
qos class attr None
qos names ['basic_qos']
        props = message['properties']
        props.update(
            body_encoding=body_encoding,
            delivery_tag=self._next_delivery_tag(),
        )
        props['delivery_info'].update(
            exchange=exchange,
            routing_key=routing_key,
        )

    def basic_consume(self, queue, no_ack, callback, consumer_tag, **kwargs):
        """Consume from `queue`."""
        self._tag_to_queue[consumer_tag] = queue
        self._active_queues.append(queue)

        def _callback(raw_message):
            message = self.Message(raw_message, channel=self)
            if not no_ack:
                self.qos.append(message, message.delivery_tag)
            return callback(message)

        self.connection._callbacks[queue] = _callback
        self._consumers.add(consumer
…[3498 chars omitted]…
.compat import register_after_fork
from kombu.utils.encoding import bytes_to_str
from kombu.utils.eventio import ERR, READ, poll
from kombu.utils.functional import accepts_argument
from kombu.utils.json import dumps, loads
from kombu.utils.objects import cached_property
from kombu.utils.scheduling import cycle_by_name
from kombu.utils.url import _parse_url

from . import virtual

try:
    import redis

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

from kombu import Connection, Consumer, Exchange, Producer, Queue, pools
from kombu.common import Broadcast
from kombu.utils.functional import maybe_list
from kombu.utils.objects import cached_property

                # 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)

    def new_missing(self, name):
        exchange = self.autoexchange(name)
        if exchange is not None and self.create_missing_queue_exchange_type:
            if isinstance(exchange, str):
                exchange = Exchange(
     
…[303 chars omitted]…
= None
        if self.create_missing_queue_type not in (None, 'classic'):
            queue_arguments = {
                'x-queue-type': self.create_missing_queue_type,
            }
        return Queue(name, exchange, name, queue_arguments=queue_arguments)

    @property
    def consume_from(self):
        if self._consume_from is not None:
            return self._consume_from
        return self

memory:// <class 'kombu.transport.memory.Channel'> True <kombu.transport.virtual.base.QoS object at 0x71d629e37f20>
amqp:// ConnectionRefusedError [Errno 111] Connection refused

==================================== ERRORS ====================================
___________ ERROR collecting t/unit/backends/test_azureblockblob.py ____________
ImportError while importing test module '/repo/t/unit/backends/test_azureblockblob.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/local/lib/python3.12/importlib/__init__.py:90: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
t/unit/backends/test_azureblockblob.py:6: in <module>
    from celery.backends import azureblockblob
celery/backends/azureblockblob.py:2: in <module>
    from kombu.transport.azurestoragequeues import Transport as AzureStorageQueuesTransport
/usr/local/lib/python3.12/site-packages/kombu/transport/azurestoragequeues.py:59: in <module>

    from azure.core.exceptions import ResourceExistsError
E   ModuleNotFoundError: No module named 'azure'
=========================== short test summary info ============================
ERROR t/unit/backends/test_azureblockblob.py
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
2 skipped, 1 error in 1.98s

Mock <class 'celery.exceptions.ImproperlyConfigured'> Event queues cannot be both durable and exclusive.
Connection <class 'kombu.exceptions.OperationalError'> [Errno 111] Connection refused

"""Creating events, and event exchange definition."""
import time
from copy import copy

from kombu import Exchange

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

EVENT_EXCHANGE_NAME = 'celeryev'
#: Exchange used to send events on.

#: 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']
…[660 chars omitted]…
xchange)
    if conn.transport.driver_type in {'redis', 'gcpubsub'}:
        # quick hack for Issue #436
        ex.type = 'fanout'
    if name != ex.name:
        ex.name = name
    return ex
"""Actual App instance implementation."""
import functools
import importlib
import inspect
import os
import sys
import threading
import typing
import warnings
from collections import UserDict, defaultdict, deque

........................................................................ [ 29%]
........................................................................ [ 59%]
........................,,,.,,,.,,,.,,,.,,,.,,,..,,,.................... [ 80%]
...,,..,..,,..,.,,.,,.,,.,,.,,..................................         [100%]
243 passed, 22 warnings, 37 subtests passed in 7.89s

225:    @cached_property
1320:    @cached_property
1329:    @cached_property
1338:    @cached_property
1347:    @cached_property
1352:    @cached_property
1356:    @cached_property
1365:    @cached_property
1369:    @cached_property
1406:    @cached_property
1424:    @cached_property
1466:    @cached_property

1467:    def control(self):
1471:    @cached_property
1476:    @cached_property
1481:    @cached_property
1486:    @cached_property
1491:    @cached_property
1509:    @cached_property
1467:    def control(self):
1468-        """Remote control: :class:`~@control`."""
1469-        return instantiate(self.control_cls, app=self)
1470-
1471-    @cached_property
1472-    def events(self):
1473-        """Consuming and sending events: :class:`~@events`."""
1474-        return instantiate(self.events_cls, app=self)
1475-
1476-    @cached_property
1477-    def loader(self):
1472:    def events(self):
1473-        """Consuming and sending events: :class:`~@events`."""
1474-        return instantiate(self.events_cls, app=self)
1475-
1476-    @cached_property
1477-    def loader(self):
1478-        """Current loader instance."""
1479-        return get_loader_cls(self.loader_cls)(app=self)
1480-
1481-    @cached_property
1482-    def log(self):

(<class 'kombu.transport.pyamqp.Channel'>, <class 'amqp.channel.Channel'>, <class 'amqp.abstract_channel.AbstractChannel'>, <class 'kombu.transport.base.StdChannel'>, <class 'object'>)
    def __init__(self, connection,
                 channel_id=None, auto_decode=True, on_open=None):
        if channel_id:
            connection._claim_channel_id(channel_id)
        else:
            channel_id = connection._get_free_channel_id()

        AMQP_LOGGER.debug('using channel_id: %s', channel_id)

        super().__init__(connection, channel_id)

        self.is_open = False
        self.active = True  # Flow control
        self.returned_messages = Queue()
        self.callbacks = {}
        self.cancel_callbacks = {}
        self.auto_decode = auto_decode
        self.events = defaultdict(set)
        self.no_ack_consumers = set()

        self.on_open = ensure_promise(on_open)

      
…[105 chars omitted]…
channel.
        self._confirm_selected = False
        if self.connection.confirm_publish:
            self.basic_publish = self.basic_publish_confirm

        __slots__ = (
        "is_open",
        "active",
        "returned_messages",
        "callbacks",
        "cancel_callbacks",
        "events",
        "no_ack_consumers",
        "on_open",
        "_confirm_selected",
        )

qos False

........................................................................ [  5%]
........................................................................ [ 11%]
........................................................................ [ 17%]
........................................................................ [ 23%]
........................................................................ [ 29%]
........................................................................ [ 34%]
F
=================================== FAILURES ===================================
________________ test_preload_options[subcommand_with_params0] _________________

subcommand_with_params = ('purge', '-f')
isolated_cli_runner = <click.testing.CliRunner object at 0x73d44b0a8080>

    @pytest.mark.parametrize(
        "subcommand_with_params",
        [
            ("purge", "-f"),
            ("shell",),
        ]

…[813 chars omitted]…
No such option '--ini'.\n" = <Result SystemExit(2)>.output

/repo/t/unit/app/test_preload_cli.py:41: AssertionError
=========================== short test summary info ============================
FAILED t/unit/app/test_preload_cli.py::test_preload_options[subcommand_with_params0]
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 432 passed, 2 warnings in 6.62s

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()
        self.queue_prefix = queue_prefix or self.app.conf.event_queue_prefix
        queue_durable = self.app.conf.event_queue_durable
        queue_exclusive = self.app.conf.event_queue_exclusive
        if queue_durable and queue_exclusive:
            raise ImproperlyConfigured(

                'Event queues cannot be both durable and exclusive.'
            )
        self.exchange = get_exchange(
            self.connection or self.app.connection_for_write(),
            name=self.app.conf.event_exchange)
        if queue_
…[525 chars omitted]…
lock = self.app.clock
        self.adjust_clock = self.clock.adjust
        self.forward_clock = self.clock.forward
        if accept is None:
            accept = {self.app.conf.event_serializer, 'json'}
        self.accept = accept

    def process(self, type, event):
        """Process event by dispatching to configured handler."""
        handler = self.handlers.get(type) or self.handlers.get('*')

........................................................................ [  5%]
........................................................................ [ 11%]
........................................................................ [ 17%]
........................................................................ [ 23%]
........................................................................ [ 29%]
........................................................................ [ 34%]
....................sssssssss........................................... [ 40%]
........................................................................ [ 46%]
.....s.................................................................. [ 52%]
..........................s............................................. [ 58%]
....................................,,,.,,,.,,,.,,,.,,,.,,,..,,,........ [ 62%]
...............,,

…[205 chars omitted]…
.......... [ 78%]
........................................................................ [ 84%]
........................................................................ [ 90%]
....................................................................s... [ 95%]
..................................................                       [100%]
1225 passed, 12 skipped, 41 warnings, 37 subtests passed in 32.84s

198:                exchange = copy(exchange)

........................................................................ [ 29%]
........................................................................ [ 59%]
........................,,,.,,,.,,,.,,,.,,,.,,,..,,,.................... [ 80%]
...,,..,..,,..,.,,.,,.,,.,,.,,..................................         [100%]
243 passed, 22 warnings, 37 subtests passed in 7.89s
 M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/consumer/tasks.py
?? statefilename

-rw-r--r-- 1 root root 16384 Nov  3 11:14 statefilename
 M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/consumer/tasks.py
diff --git a/celery/app/amqp.py b/celery/app/amqp.py
index 8dcec36..9c66345 100644
--- a/celery/app/amqp.py
+++ b/celery/app/amqp.py
@@ -2,6 +2,7 @@

 import numbers
 from collections import namedtuple
 from collections.abc import Mapping
+from copy import copy
 from datetime import timedelta
 from weakref import WeakValueDictionary
 
@@ -55,7 +56,9 @@ class Queues(dict):
 
     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',
+   
…[8629 chars omitted]…
             )
+                limit = max_concurrency or c.pool.num_processes
+                if len(state.reserved_requests) >= limit:
+                    return False
+                return original_can_consume()
+
+            qos.can_consume = can_consume
+
         def set_prefetch_count(prefetch_count):
             return c.task_consumer.qos(
                 prefetch_count=prefetch_count,

........................................................................ [ 29%]
........................................................................ [ 59%]
........................,,,.,,,.,,,.,,,.,,,.,,,..,,,.................... [ 80%]
...,,..,..,,..,.,,.,,.,,.,,.,,..................................         [100%]
243 passed, 22 warnings, 37 subtests passed in 7.89s
feature reproductions passed

grep: celery/app/__pycache__/amqp.cpython-312.pyc: binary file matches
celery/app/amqp.py:42:class Queues(dict):
celery/app/amqp.py:226:    queues_cls = Queues
celery/app/amqp.py:292:        return self.queues_cls(
t/unit/app/test_amqp.py:68:class test_Queues:
264:    def Queues(self, queues, create_missing=None,
265-               autoexchange=None, max_priority=None,
266-               create_missing_queue_type=None,
267-               create_missing_queue_exchange_type=None):
268-        # Create new :class:`Queues` instance, using queue defaults
269-        # from the current configuration.
270-        conf = self.app.conf

271-        default_routing_key = conf.task_default_routing_key
272-        if create_missing is None:
273-            create_missing = conf.task_create_missing_queues
274-        if max_priority is None:
275-            max_priority = conf.task_queue_max_priori
…[1332 chars omitted]…
outes, queues or self.queues,
 celery/app/amqp.py              | 38 +++++++++++++++++++++++++++++++++++---
 celery/app/control.py           | 10 +++++++++-
 celery/app/defaults.py          |  7 +++++++
 celery/bin/worker.py            |  5 +++++
 celery/events/receiver.py       | 10 +++++++++-
 celery/worker/consumer/tasks.py | 20 +++++++++++++++++++-
 6 files changed, 84 insertions(+), 6 deletions(-)

celery/worker/consumer/tasks.py:46:        disable_prefetch = c.app.conf.worker_disable_prefetch
grep: celery/worker/consumer/__pycache__/tasks.cpython-312.pyc: binary file matches
celery/app/amqp.py:277:            create_missing_queue_type = conf.task_create_missing_queue_type
celery/app/control.py:431:        queue_durable = app.conf.control_queue_durable
celery/events/receiver.py:45:        queue_durable = self.app.conf.event_queue_durable

M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/consumer/tasks.py
 celery/app/amqp.py              | 38 +++++++++++++++++++++++++++++++++++---
 celery/app/control.py           | 10 +++++++++-
 celery/app/defaults.py          |  7 +++++++
 celery/bin/worker.py            |  5 +++++
 celery/events/receiver.py       | 10 +++++++++-
 celery/worker/consumer/tasks.py | 20 +++++++++++++++++++-

 6 files changed, 84 insertions(+), 6 deletions(-)
diff --git a/celery/app/control.py b/celery/app/control.py
index 603d930..9dcd2b5 100644
--- a/celery/app/control.py
+++ b/celery/app/control.py
@@ -20,7 +20,7 @@ from kombu.utils.compat import register_after_fork
 from kombu.utils.functional import lazy
 from kombu.utils.objects import cached_property
 
-from celery.exceptions import DuplicateNodenameWarning
…[6324 chars omitted]…
             )
+                limit = max_concurrency or c.pool.num_processes
+                if len(state.reserved_requests) >= limit:
+                    return False
+                return original_can_consume()
+
+            qos.can_consume = can_consume
+
         def set_prefetch_count(prefetch_count):
             return c.task_consumer.qos(
                 prefetch_count=prefetch_count,

None direct auto
Exchange ''(direct) {'x-queue-type': 'quorum'}

M celery/app/amqp.py
 M celery/app/control.py
 M celery/app/defaults.py
 M celery/bin/worker.py
 M celery/events/receiver.py
 M celery/worker/consumer/tasks.py
 celery/app/amqp.py              | 38 +++++++++++++++++++++++++++++++++++---
 celery/app/control.py           | 10 +++++++++-
 celery/app/defaults.py          |  7 +++++++
 celery/bin/worker.py            |  5 +++++
 celery/events/receiver.py       | 10 +++++++++-
 celery/worker/consumer/tasks.py | 20 +++++++++++++++++++-

 6 files changed, 84 insertions(+), 6 deletions(-)
diff --git a/celery/app/amqp.py b/celery/app/amqp.py
index 8dcec36..44ead20 100644
--- a/celery/app/amqp.py
+++ b/celery/app/amqp.py
@@ -2,6 +2,7 @@
 import numbers
 from collections import namedtuple
 from collections.abc import Mapping
+from copy import copy
 from datetime import timedelta
 from weakref import WeakValueDictionary
 
@@ -55,7 +56,9 @@ class Q
…[8629 chars omitted]…
             )
+                limit = max_concurrency or c.pool.num_processes
+                if len(state.reserved_requests) >= limit:
+                    return False
+                return original_can_consume()
+
+            qos.can_consume = can_consume
+
         def set_prefetch_count(prefetch_count):
             return c.task_consumer.qos(
                 prefetch_count=prefetch_count,

submitted.