celery-celery-10408-10404

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
774dbb5 base
On branch master
nothing to commit, working tree clean

celery/canvas.py:356:    def apply(self, args=None, kwargs=None, **options):
celery/canvas.py:370:    def apply_async(self, args=None, kwargs=None, route_name=None, **options):
celery/canvas.py:935:class _chain(Signature):
celery/canvas.py:1036:    def apply_async(self, args=None, kwargs=None, **options):
celery/canvas.py:1294:    def apply(self, args=None, kwargs=None, **options):
celery/canvas.py:1322:class chain(_chain):
celery/canvas.py:1398:    def apply_async(self, args=None, kwargs=None, **opts):
celery/canvas.py:1455:    def apply_async(self, args=None, kwargs=None, **opts):
celery/canvas.py:1471:    def apply_chunks(cls, task, it, n, app=None):
celery/canvas.py:1492:class group(Signature):
celery/canvas.py:1616:    def apply_async(self, args=None, kwargs=None, add_to_parent=True,
celery/canvas.py:1652:    def apply(self, args=None, kwargs=None, **options):

celery/canvas.py:21
…[265 chars omitted]…
t.py:482:                result_from_tuple(child, self.app) for child in children
celery/result.py:930:class GroupResult(ResultSet):
celery/result.py:1005:    def as_tuple(self):
celery/result.py:1113:def result_from_tuple(r, app=None):
celery/result.py:1123:            parent = result_from_tuple(parent, app)
celery/result.py:1127:                id, [result_from_tuple(child, app) for child in nodes],

# of tasks in the chain instead of setting a new one from the chain_
    # variable above.
    # This causes Celery to execute a chain, even though there might not be
    # one to begin with. Alternatively, it causes Celery to execute more tasks
    # that were previously present in the previous task in the group.
    # The solution is to be careful and never mutate the options dictionary
    # to begin with.
    # Here is an example of a canvas which triggers this issue:
    # add.s(5, 6) | group((add.s(1) | add.s(2), add.s(3))).
    # The expected result is [14, 14]. However, when we extend the `chain`
    # key the `add.s(3)` task erroneously has `add.s(2)` in its chain since
    # it was previously applied to `add.s(1)`.

    # Without being careful not to mutate the options dictionary, the result
    # in this case is [16, 14].
    # To avoid deep-copying the entire options di
…[8629 chars omitted]…
get()
            4

        Using a generator expression:

        .. code-block:: pycon

            >>> lazy_chain = chain(add.s(i) for i in range(10))
            >>> res = lazy_chain(3)

    Arguments:
        *tasks (Signature): List of task signatures to chain.
            If only one argument is passed and that argument is
            an iterable, then that'll be used as the list of signatures

The tasks list would be cloned from the chain's tasks.
        All of the chain callbacks would be added to the last task in the (cloned) chain.
        All of the tasks would be linked to the same error callback
        as the chain itself, to ensure that the correct error callback is called
        if any of the (cloned) tasks of the chain fail.
        """
        # Clone chain's tasks assigning signatures from link_error
        # to each task and adding the chain's links to the last task.
        tasks = [t.clone() for t in self.tasks]
        for sig in maybe_list(self.options.get('link')) or []:
            tasks[-1].link(sig)
        for sig in maybe_list(self.options.get('link_error')) or []:

            for task in tasks:
                task.link_error(sig)
        return tasks

    def apply_async(self, args=None, kwargs=None, **options):
        # python is best a
…[8629 chars omitted]…
okupError:
                pass
        return app or current_app

    def __repr__(self):
        if not self.tasks:
            return f'<{type(self).__name__}@{id(self):#x}: empty>'
        return remove_repeating_from_task(
            self.tasks[0]['task'],
            ' | '.join(repr(t) for t in self.tasks))


class chain(_chain):
    """Chain tasks together.

    Each tasks follows one another,

536:    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
722:                def apply_function_async(fun, *args, **kwargs):
869:    def apply(self, args=None, kwargs=None,
            options['priority'] = priority
        if queue:
            options['queue'] = queue
        else:
            exchange = delivery_info.get('exchange')
            routing_key = delivery_info.get('routing_key')
            if exchange == '' and routing_key:
                # sent to anon-exchange
                options['queue'] = routing_key

            else:
                options.update(delivery_info)
        return self.signature(
            args, kwargs, options, type=self, **extra_options
        )
    subtask_from_request = signature_from_request  # XXX compat

    def retry(self, args=None, kwargs=None, exc=None, throw=True,
              eta=None, countdown=None, ma
…[3949 chars omitted]…
     # On Py3: will augment any current exception with
                # the exc' argument provided (raise exc from orig)
                raise_with_context(exc)
            raise self.MaxRetriesExceededError(
                "Can't retry {}[{}] args:{} kwargs:{}".format(
                    self.name, request.id, S.args, S.kwargs
                ), task_args=S.args, task_kwargs=S.kwargs
            )

try:
            S.apply_async()
        except Exception as exc:
            raise Reject(exc, requeue=False)
        if throw:
            raise ret
        return ret

    def apply(self, args=None, kwargs=None,
              link=None, link_error=None,
              task_id=None, retries=None, throw=None,
              logfile=None, loglevel=None, headers=None, **options):

        """Execute this task locally, by blocking until the task returns.

        Arguments:
            args (Tuple): positional arguments passed on to the task.
            kwargs (Dict): keyword arguments passed on to the task.
            throw (bool): Re-raise task exceptions.
                Defaults to the :setting:`task_eager_propagates` setting.

        Returns:
            celery.result.EagerResult: pre-evaluated result.
        """
        # trace imports Task, so need to import inline.
   
…[3100 chars omitted]…
g arguments and execution options
                for a single task invocation.
        """
        starkwargs.setdefault('app', self.app)
        return signature(self, args, *starargs, **starkwargs)
    subtask = signature

    def s(self, *args, **kwargs):
        """Create signature.

        Shortcut for ``.s(*a, **k) -> .signature(a, k)``.
        """
        return self.signature(args, kwargs)

23:from celery.exceptions import BackendGetMetaError, Ignore, InvalidTaskError, Reject, Retry
73:LOG_IGNORED = """\
78:LOG_REJECTED = """\
92:log_policy_reject = log_policy_t(LOG_REJECTED, 'rejected', logging.WARN, 1, 1)
93:log_policy_ignore = log_policy_t(LOG_IGNORED, 'ignored', logging.INFO, 0, 0)
109:IGNORED = states.IGNORED
110:REJECTED = states.REJECTED
114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
138:    if isinstance(exc, Reject):
140:    elif isinstance(exc, Ignore):
339:            # Ignore the exception raised if the frame is still executing.
436:        chain step fails (or vice-versa), a Reject + redeliver may

521:                        info(LOG_IGNORED, {
560:                            raise Reject(exc, requeue=True)
587:                except Reject as exc:
588:                    I, R = Info(REJECTED, exc), ExceptionInfo(internal=True)
593:               
…[2778 chars omitted]…
   return False

    @ignored.setter
    def ignored(self, value):
        """Enable/disable task result retrieval."""
        self._ignored = value

    def then(self, callback, on_error=None, weak=False):
        self.backend.add_pending_result(self, weak=weak)
        return self.on_ready.then(callback, on_error)

    def _on_fulfilled(self, result):
        self.backend.remove_pending_result(self)

"""`hash(self) -> hash(self.id)`."""
        return hash(self.id)

    def __repr__(self):
        return f'<{type(self).__name__}: {self.id}>'

    def __eq__(self, other):
        if isinstance(other, AsyncResult):
            return other.id == self.id
        elif isinstance(other, str):
            return other == self.id
        return NotImplemented

    def __copy__(self):
        return self.__class__(
            self.id, self.backend, None, self.app, self.parent,
        )

    def __reduce__(self):
        return self.__class__, self.__reduce_args__()

    def __reduce_args__(self):
        return self.id, self.backend, None, None, self.parent

    def __del__(self):
        """Cancel pending operations when the instance is destroyed."""
        if self.backend is not None:
            self.backend.remove_pending_result(self)

    @cached_property
    def graph(sel
…[8628 chars omitted]…
    res, nodes = r
        id, parent = res if isinstance(res, (list, tuple)) else (res, None)
        if parent:
            parent = result_from_tuple(parent, app)

        if nodes is not None:
            return app.GroupResult(
                id, [result_from_tuple(child, app) for child in nodes],
                parent=parent,
            )

        return Result(id, parent=parent)
    return r

@ignored.setter
    def ignored(self, value):
        """Enable/disable task result retrieval."""
        self._ignored = value

    def then(self, callback, on_error=None, weak=False):
        self.backend.add_pending_result(self, weak=weak)
        return self.on_ready.then(callback, on_error)

    def _on_fulfilled(self, result):
        self.backend.remove_pending_result(self)
        return result

    def as_tuple(self):
        parent = self.parent
        return (self.id, parent and parent.as_tuple()), None

    def as_list(self):
        """Return as a list of task IDs."""
        results = []
        parent = self.parent
        results.append(self.id)
        if parent is not None:
            results.extend(parent.as_list())
        return results

    def forget(self):
        """Forget the result of this task and its parents."""
        self._cache = None
        if 
…[5296 chars omitted]…
     if self._app is None:
            self._app = (self.results[0].app if self.results else
                         current_app._get_current_object())
        return self._app

    @app.setter
    def app(self, app):
        self._app = app

    @property
    def backend(self):
        return self.app.backend if self.app else self.results[0].backend


@Thenable.register
class GroupResult(ResultSet):

t/unit/tasks/test_result.py:14:from celery.result import AsyncResult, EagerResult, GroupResult, ResultSet, assert_will_not_block, result_from_tuple
t/unit/tasks/test_result.py:163:        child = [self.app.AsyncResult(uuid()).as_tuple()
t/unit/tasks/test_result.py:235:        r1._set_cache({'children': [r2.as_tuple()]})
t/unit/tasks/test_result.py:1036:        assert x, result_from_tuple(x.as_tuple() == self.app)
t/unit/tasks/test_result.py:1037:        assert x, result_from_tuple(x == self.app)
t/unit/tasks/test_result.py:1042:        y = result_from_tuple(x.as_tuple(), self.app)
t/unit/tasks/test_result.py:1049:        x = result_from_tuple([uid, []], app=self.app)
t/unit/tasks/test_result.py:1062:        assert x, result_from_tuple(x.as_tuple() == self.app)
t/unit/tasks/test_result.py:1063:        assert x, result_from_tuple(x == self.app)
t/unit/tasks/test_result.py:1071:        s
…[21 chars omitted]…
_from_tuple(result.as_tuple(), self.app)

t/unit/tasks/test_result.py:1075:    def test_GroupResult_as_tuple(self):
t/unit/tasks/test_result.py:1083:        (result_id, parent_tuple), group_results = result.as_tuple()
t/unit/tasks/test_result.py:1085:        assert parent_tuple == parent.as_tuple()
t/unit/tasks/test_chord.py:92:                         GroupResult=GroupResult, result_from_tuple=Mock())

def test_revoke(self):
        res = self.raising.apply(args=[3, 3])
        assert not res.revoke()

    @patch('celery.result.task_join_will_block')
    def test_get_sync_subtask_option(self, task_join_will_block):
        task_join_will_block.return_value = True
        tid = uuid()
        res_subtask_async = EagerResult(tid, 'x', 'x', states.SUCCESS)
        with pytest.raises(RuntimeError):
            res_subtask_async.get()
        res_subtask_async.get(disable_sync_subtasks=False)

    def test_populate_name(self):
        res = EagerResult('x', 'x', states.SUCCESS, None, 'test_task')
        assert res.name == 'test_task'

        res = EagerResult('x', 'x', states.SUCCESS, name='test_task_named_argument')
        assert res.name == 'test_task_named_argument'


class test_tuples:

    def test_AsyncResult(self):
        x = self.app.AsyncResult(uuid())
        assert x,
…[1607 chars omitted]…
   (result_id, parent_tuple), group_results = result.as_tuple()
        assert result_id == result.id
        assert parent_tuple == parent.as_tuple()
        assert parent_tuple[0][0] == parent.id
        assert isinstance(group_results, list)
        expected_grp_res = [((f'async-result-{i}', None), None)
                            for i in range(2)]
        assert group_results == expected_grp_res

t/unit/tasks/test_context.py:10:            continue   # Ignore pseudo-private attributes
t/unit/tasks/test_context.py:13:            continue   # Ignore methods and other non-trivial types
t/unit/tasks/test_stamping.py:9:from celery.exceptions import Ignore
t/unit/tasks/test_stamping.py:1042:        with pytest.raises(Ignore):
t/unit/tasks/test_tasks.py:14:from celery.exceptions import Ignore, ImproperlyConfigured, Retry
t/unit/tasks/test_tasks.py:1175:        with pytest.raises(Ignore):
t/unit/tasks/test_tasks.py:1183:        with pytest.raises(Ignore):
t/unit/tasks/test_tasks.py:1216:            with pytest.raises(Ignore):
t/unit/tasks/test_tasks.py:1230:        with pytest.raises(Ignore):
t/unit/tasks/test_tasks.py:1239:        with pytest.raises(Ignore):
t/unit/tasks/test_tasks.py:1243:        with pytest.raises(Ignore):
t/unit/tasks/test_tasks.py:1247:        with pytest.raises(

…[753 chars omitted]…
es.REJECTED
t/unit/tasks/test_trace.py:711:                with pytest.raises(Reject):
t/unit/tasks/test_trace.py:910:                with pytest.raises(Reject):
t/unit/tasks/test_trace.py:975:                with pytest.raises(Reject):
t/unit/tasks/test_trace.py:1049:        is caught and re-raised as Reject(requeue=True)."""
t/unit/tasks/test_trace.py:1089:                with pytest.raises(Reject):

27:__version__ = '5.6.2'
43:VERSION_BANNER = f'{__version__} ({SERIES})'
52:    r'(\d+)\.(\d+)\.(\d+)(.+)?', __version__).groups()
53:VERSION = version_info = version_info_t(
171:    __path__=__path__, __doc__=__doc__, __version__=__version__,
174:    VERSION=VERSION, SERIES=SERIES, VERSION_BANNER=VERSION_BANNER,
commit 774dbb5eb597d04ff96c969028aff81f7d8408ad
Author: benchgen <sandbox@benchgen>
Date:   Thu Nov 5 13:56:18 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_TEMPLAT
…[618 chars omitted]…
alysis.yml              |   71 +
 .github/workflows/docker.yml                       |   76 +
 .github/workflows/integration-tests.yml            |   76 +
 .github/workflows/linter.yml                       |   17 +
 .github/workflows/python-package.yml               |  139 +
 .github/workflows/semgrep.yml                      |   29 +
 .github/workflows/smoke-tests.yml                  |   63 +
5.6.2

"""Composing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
"""

import itertools
import operator
import types
import warnings
from abc import ABCMeta, abstractmethod

from collections import deque
from collections.abc import MutableSequence
from copy import deepcopy
from functools import partial as _partial
from functools import reduce
from operator import itemgetter
from types import GeneratorType

from kombu.utils.functional import fxrange, reprcall
from kombu.utils.objects import cached_property
from kombu.utils.uuid import uuid
from vine import barrier

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
from c
…[1392 chars omitted]…
irst one.

    Example:
    >>> d1 = {'dict': {'a': 1}, 'list': [1, 2], 'tuple': (1, 2)}
    >>> d2 = {'dict': {'b': 2}, 'list': [3, 4], 'set': {'a', 'b'}}
    >>> _merge_dictionaries(d1, d2)

    d1 will be modified to: {
        'dict': {'a': 1, 'b': 2},
        'list': [1, 2, 3, 4],
        'tuple': (1, 2),
        'set': {'a', 'b'}
    }

    Arguments:
        d1 (dict): Dictionary to merge into.

visitor_headers = visitor.on_chain_start(self, **headers) or {}
        headers = self._stamp_headers(visitor_headers, append_stamps, **headers)
        self.stamp_links(visitor, **headers)

        for task in self.tasks:
            task.stamp(visitor, append_stamps, **headers)

        if visitor is not None:
            visitor.on_chain_end(self, **headers)

    def prepare_steps(self, args, kwargs, tasks,
                      root_id=None, parent_id=None, link_error=None, app=None,

                      last_task_id=None, group_id=None, chord_body=None,
                      clone=True, from_dict=Signature.from_dict,
                      group_index=None):
        """Prepare the chain for execution.

        To execute a chain, we first need to unpack it correctly.
        During the unpacking, we might encounter other chains, groups, or chords
        which we need
…[6972 chars omitted]…
           prev_res = node
        # Use the last task's actual ID, not the input parameter.
        self.id = results[0].id if results else last_task_id
        return tasks, results

    def apply(self, args=None, kwargs=None, **options):
        args = args if args else ()
        kwargs = kwargs if kwargs else {}
        last, (fargs, fkwargs) = None, (args, kwargs)
        for task in self.tasks:

docs/faq.rst:556:    * The `header` and `fanout` exchange types
docs/faq.rst:557:        (`fanout` is supported by Redis).
docs/history/changelog-2.2.rst:482:    fanout exchanges so it's able to perform worker remote control
docs/history/changelog-3.1.rst:99:    result list are not properly tupled with the as_tuple() method when it is
docs/history/changelog-3.1.rst:100:    a parent result. This is due to the as_tuple() method performing a logical
docs/history/changelog-3.1.rst:929:        BROKER_TRANSPORT_OPTIONS = {'fanout_patterns': True}
docs/history/changelog-4.2.rst:285:- **Group Result**: Modify GroupResult.as_tuple() to include parent (fixes #4106) (#4205)
docs/history/changelog-4.2.rst:397:- Matched the behavior of `GroupResult.as_tuple()` to that of `AsyncResult.as_tuple()`.
docs/history/changelog-4.4.rst:20:- Remove redis fanout caveats.
docs/history/whatsnew-3.1.rst:585:   
…[1938 chars omitted]…
        (result_id, parent_tuple), group_results = result.as_tuple()

t/unit/tasks/test_result.py:1085:        assert parent_tuple == parent.as_tuple()
t/unit/worker/test_native_delayed_delivery.py:76:    def test_start_native_delayed_delivery_fanout_exchange(self, caplog):
t/unit/worker/test_native_delayed_delivery.py:81:            'celery': Queue('celery', exchange=Exchange('celery', type='fanout'))

import json
import math
from collections.abc import Iterable
from unittest.mock import ANY, MagicMock, Mock, call, patch, sentinel

import pytest

from celery._state import _task_stack
from celery.canvas import (Signature, _chain, _maybe_group, _merge_dictionaries, chain, chord, chunks, group,
                           maybe_signature, maybe_unroll_group, signature, xmap, xstarmap)
from celery.result import AsyncResult, EagerResult, GroupResult

SIG = Signature({
    'task': 'TASK',
    'args': ('A1',),
    'kwargs': {'K1': 'V1'},
    'options': {'task_id': 'TASK_ID'},
    'subtask_type': ''},
)


def return_True(*args, **kwargs):
    # Task run functions can't be closures/lambdas, as they're pickled.
    return True


class test_maybe_unroll_group:

    def test_when_no_len_and_no_length_hint(self):
        g = MagicMock(name='group')
        g.tasks.__len__.side_effect = TypeError(
…[3051 chars omitted]…
sult
479:        assert isinstance(result, GroupResult)
490:        assert isinstance(result, GroupResult)
803:        assert not isinstance(res, GroupResult)
805:        assert not isinstance(res.parent, GroupResult)
806:        assert isinstance(res.parent.parent, GroupResult)
808:        assert not isinstance(res.parent.parent.parent, GroupResult)
1509:        assert isinstance(result, GroupResult)

def chain_add():
                return (self.add.s(4, 4) | self.add.s(8)).apply_async()

            r = chain_add.apply_async(throw=True).get()
            assert r.get() == 16
        finally:
            _state.task_join_will_block = fixture_task_join_will_block
            result.task_join_will_block = fixture_task_join_will_block

    def test_apply(self):
        x = chain(self.add.s(4, 4), self.add.s(8), self.add.s(10))
        res = x.apply()

        assert isinstance(res, EagerResult)
        assert res.get() == 26

        assert res.parent.get() == 16
        assert res.parent.parent.get() == 8
        assert res.parent.parent.parent is None

    def test_kwargs_apply(self):
        x = chain(self.add.s(), self.add.s(8), self.add.s(10))
        res = x.apply(kwargs={'x': 1, 'y': 1}).get()
        assert res == 20

    def test_single_expresion(self):
        x 
…[4108 chars omitted]…
x.app is current_app

    def test_chord_size_simple(self):
        sig = chord(self.add.s())
        assert sig.__length_hint__() == 1

    def test_chord_size_with_body(self):
        sig = chord(self.add.s(), self.add.s())
        assert sig.__length_hint__() == 1

    def test_chord_size_explicit_group_single(self):
        sig = chord(group(self.add.s()))
        assert sig.__length_hint__() == 1

t/unit/conftest.py:124:def app(celery_app):
__init__.py
test_canvas.py
test_chord.py
test_context.py
test_result.py
test_stamping.py
test_states.py
test_tasks.py
test_trace.py
unit_tasks.py

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


@pytest.fixture(autouse=True)
def reset_cache_backend_state(celery_app):
    """Fixture that resets the internal state of the cache result backend."""
    yield
    backend = celery_app.__dict__.get('backend')
    if backend is not None:
        if isinstance(backend, CacheBackend):

            if isinstance(backend.client, DummyClient):
                backend.client.cache.clear()
            backend._cache.clear()


@contextmanager
def assert_signal_called(signal, **expected):
    """Context that verifies signal is called before exiting."""
    handler = Mock()

    def on_call(**kwargs):
        return handler(**kwargs)

    signal.connect(on_call)
    try:
        yield handler
    finally:
        signal.disconnect(on_call)
    handler.assert_called_with(signal=signal, **expected)


@pytest.fixture
def 
…[267 chars omitted]…
       'multiprocessing.Process',
        'billiard.Process',
        'billiard.context.Process',
        'billiard.process.Process',
        'billiard.process.BaseProcess',
        'multiprocessing.Process',
    ]
    ctxs = [patch(s) for s in stuff]
    [ctx.__enter__() for ctx in ctxs]

    yield

    [ctx.__exit__(*sys.exc_info()) for ctx in ctxs]


def alive_threads():
    return [
        thread

grep: t/conftest.py: No such file or directory
t/unit/conftest.py:23:from celery.contrib.pytest import celery_app, celery_enable_logging, celery_parameters, depends_on_current_app
t/unit/conftest.py:29:    'celery_app', 'celery_enable_logging', 'depends_on_current_app',
t/unit/conftest.py:91:def use_celery_app_trap():
t/unit/conftest.py:96:def reset_cache_backend_state(celery_app):
t/unit/conftest.py:99:    backend = celery_app.__dict__.get('backend')
t/unit/conftest.py:124:def app(celery_app):
t/unit/conftest.py:125:    yield celery_app
import builtins
import inspect
import io
import logging

import os
import platform
import sys
import threading
import types
import warnings
from contextlib import contextmanager
from functools import wraps
from importlib import import_module, reload
from unittest.mock import MagicMock, Mock, patch

import pytest
from kombu import Queue

from celery.bac
…[2061 chars omitted]…
O_PORT') or 27017,
            'database': os.environ.get('MONGO_DB') or 'celery_unittests',
            'taskmeta_collection': (
                os.environ.get('MONGO_TASKMETA_COLLECTION') or
                'taskmeta_collection'
            ),
            'user': os.environ.get('MONGO_USER'),
            'password': os.environ.get('MONGO_PASSWORD'),
        }
    }


@pytest.fixture(scope='session')

<class 'urllib.error.URLError'> <urlopen error [Errno -3] Temporary failure in name resolution>

protocol 1
 AsyncResult 556e65d1-0890-4895-91c4-1f98c47d615a state? PENDING
 parent ->
   GroupResult 28a618d0-f87d-4cb6-9173-7916e9f87d75 state? None
   children: 2
     AsyncResult 9b43c880-5820-4c4b-8511-0961cea1eafa state? PENDING
     AsyncResult 944836e7-dad2-48cf-b38f-09eeb1f25f69 state? PENDING
   parent ->
     GroupResult 80980a24-4b0c-4e1c-8382-62dfd5d01f01 state? None
     children: 2
       AsyncResult 68a2abd3-7250-4527-ab6c-3fa809f2de02 state? PENDING
       AsyncResult 63ec2c19-1dfa-45f8-a04c-1342f403c504 state? PENDING

     parent ->
       AsyncResult 2320f05c-7ec1-49f1-ba3d-ef4cb052e5f5 state? PENDING
tuple: (('556e65d1-0890-4895-91c4-1f98c47d615a', (('28a618d0-f87d-4cb6-9173-7916e9f87d75', (('80980a24-4b0c-4e1c-8382-62dfd5d01f01', (('2320f05c-7ec1-49f1-ba3d-ef4cb052e5f5', None), None)), [(('68a2abd3-7250-4527-ab6c-3fa809f2de02', None), None), (('63ec2c19-1dfa-45f8
…[2066 chars omitted]…
te? PENDING
     AsyncResult 7f1c995c-bad7-4e2e-b09c-0f5539e88a77 state? PENDING
   parent ->
     GroupResult 3a56788e-9b2a-4979-a3f1-72615442ba24 state? None
     children: 2
       AsyncResult 3b33df03-ae84-4205-8b87-6a2bca5a2f14 state? PENDING
       AsyncResult 2f4f3d9b-99dd-4739-abac-1ae8f0272c4a state? PENDING
     parent ->
       AsyncResult eb5e0a61-ed09-40d5-a3ca-c4279a1ff613 state? PENDING

AsyncResult 082b39b4-45e6-4be3-aca1-59f05354b7ea
   AsyncResult 099fa0a4-80e9-4311-9ca4-33ca02ee00b5
 parent ->
   GroupResult 1036a422-be8e-4a5c-959b-4a41ccb30def
   children: 2
     AsyncResult 79a0831d-6efa-413e-a55c-6a68dabd908c
     AsyncResult 30b44928-0c3f-41c9-95b6-4415e35f1c53
   parent ->
     AsyncResult b27beeee-77b7-428a-8906-6ae7565064b1
restored:
 GroupResult 5c1ae64e-b334-49cc-a387-48c1c0f8174d
 children: 2

   AsyncResult 082b39b4-45e6-4be3-aca1-59f05354b7ea
   AsyncResult 099fa0a4-80e9-4311-9ca4-33ca02ee00b5
 parent ->
   GroupResult 1036a422-be8e-4a5c-959b-4a41ccb30def
   children: 2
     AsyncResult 79a0831d-6efa-413e-a55c-6a68dabd908c
     AsyncResult 30b44928-0c3f-41c9-95b6-4415e35f1c53
   parent ->
     AsyncResult b27beeee-77b7-428a-8906-6ae7565064b1
orig counts (2, 4, 1) restored counts (2, 4, 1)

=== chain no before protocol 1 type _chord
 AsyncResult 41abf
…[6039 chars omitted]…
241b21f-900c-42f7-b17e-3cda55d71444
     AsyncResult 455ef136-32d3-4cb6-8a18-7801416d325e
   parent ->
     GroupResult d6f7ec5d-733a-47b7-983f-ecdd91c0c124
     children: 2
       AsyncResult 526c8597-4f7c-4b57-adc1-5e11f4662b87
       AsyncResult 118e6f98-2095-499a-ba8f-0fdcb13ffc3f
     parent ->
       AsyncResult 6dd60a58-164e-443f-9355-a38cdcfc3ca9
orig counts (2, 4, 2) restored counts (2, 4, 2)

* master

.. _changelog:

================
 Change history
================

This document contains change notes for bugfix & new features
in the main branch & 5.6.x series, please see :ref:`whatsnew-5.6` for
an overview of what's new in Celery 5.6.

.. _version-5.6.2:

5.6.2
=====

:release-date: 2026-01-04
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix recursive WorkController instantiation in DjangoWorkerFixup + AttributeError when pool_cls is a string (#10045)
- Bugfix: Revoked tasks now immediately update backend status to REVOKED (#9869)
- Prepare for release: v5.6.2 (#10049)

.. _version-5.6.1:

5.6.1
=====

:release-date: 2025-12-29
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix Redis Sentinel ACL authentication support (#10013)
- Fix: Broker heartbeats not sent during graceful shutdown (#9986)
- docs #5410 -- Document confirm_publish broker transport
…[2858 chars omitted]…
s99dev>`_ in
`#9806 <https://github.com/celery/celery/pull/9806>`_.

ETA Task Memory Limit
---------------------
8:in the main branch & 5.6.x series, please see :ref:`whatsnew-5.6` for
9:an overview of what's new in Celery 5.6.
11:.. _version-5.6.2:
13:5.6.2
24:- Prepare for release: v5.6.2 (#10049)
26:.. _version-5.6.1:
28:5.6.1
47:- Prepare for release: v5.6.1 (#10037)
49:.. _version-5.6.0:
51:5.6.0

protocol 1
Traceback (most recent call last):
  File "/tmp/repro_eager_groups.py", line 16, in <module>
    res = c.apply_async()
          ^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1044, in apply_async
    return self.apply(args, kwargs, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1300, in apply
    last and (last.get(),), **dict(self.options, **options))
              ^^^^^^^^^^
  File "/repo/celery/result.py", line 735, in get

    return (self.join_native if self.supports_native_join else self.join)(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join
    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: add() takes 2 positional arguments but 3 were given

fail 0

@Signature.register_type()
class group(Signature):
    """Creates a group of tasks to be executed in parallel.

    A group is lazy so you must call it to take action and evaluate
    the group.

    Note:
        If only one argument is passed, and that argument is an iterable
        then that'll be used as the list of tasks instead: this
        allows us to use ``group`` with generator expressions.

    Example:
        >>> lazy_group = group([add.s(2, 2), add.s(4, 4)])
        >>> promise = lazy_group()  # <-- evaluate: returns lazy result.
        >>> promise.get()  # <-- will wait for the task to return
        [4, 8]

    Arguments:
        *tasks (List[Signature]): A list of signatures that this group will
            call. If there's only one argument, and that argument is an
            iterable, then that'll define the list of signatures instead.
        **options (Any): 
…[8629 chars omitted]…

        """Return the number of tasks in this chord's header (recursively)."""
        tasks = getattr(self.tasks, "tasks", self.tasks)
        return sum(self._descend(task) for task in tasks)

    def run(self, header, body, partial_args, app=None, interval=None,
            countdown=1, max_retries=None, eager=False,
            task_id=None, kwargs=None, **options):
        """Execute the chord.

# TODO: Not sure if this is still used anywhere (besides its own tests). Consider removing.
        it = fxrange(start, stop, step, repeatlast=True)
        for task in self.tasks:
            task.set(countdown=next(it))
        return self

    def apply_async(self, args=None, kwargs=None, add_to_parent=True,
                    producer=None, link=None, link_error=None, **options):
        args = args if args else ()
        if link is not None:
            raise TypeError('Cannot add link to group: use a chord')
        if link_error is not None:

            raise TypeError(
                'Cannot add link to group: do that on individual tasks')
        app = self.app
        if app.conf.task_always_eager:
            return self.apply(args, kwargs, **options)
        if not self.tasks:
            return self.freeze()

        options, group_id, root_id = self._freeze_gi
…[5884 chars omitted]…
)

        if self.app.conf.task_allow_error_cb_on_chord_header:
            for task in maybe_list(self.tasks) or []:
                task.link_error(errback.clone(immutable=True))
        else:
            # Once this warning is removed, the whole method needs to be refactored to:
            # 1. link the error callback to each task in the header
            # 2. link the error callback to the body

- **Canvas** Keyword arguments are passed to tasks in chain as expected.

  Contributed by :github_user:`tothegump`

- **Django** Fix a regression causing Celery to crash when using Django.

  Contributed by **Jonas Haag**

- **Canvas** Chain with one task now runs as expected.

  Contributed by :github_user:`tothegump`

- **Kombu** Celery 4.2 now requires Kombu 4.2 or better.

  Contributed by **Omer Katz & Asif Saifuddin Auvi**

- `GreenletExit` is not in `__all__` in greenlet.py which can not be imported by Python 3.6.

  The import was adjusted to work on Python 3.6 as well.

  Contributed by **Hsiaoming Yang**

- Fixed a regression that occurred during the development of Celery 4.2 which caused `celery report` to crash when Django is installed.

  Contributed by **Josue Balandrano Coronel**

- Matched the behavior of `GroupResult.as_tuple()` to that of `AsyncResult.as_tuple()`.

…[850 chars omitted]…
or GroupResult with an empty
    result list are not properly tupled with the as_tuple() method when it is
    a parent result. This is due to the as_tuple() method performing a logical
    and operation on the ResultSet.

    Fix contributed by Colin McIntosh.

- **Worker**: Fixed wrong values in autoscale related logging message.

    Fix contributed by ``@raducc``.

- Documentation improvements by

/repo/celery/backends/base.py:752: RuntimeWarning: Results are not stored in backend and should not be retrieved when task_always_eager is enabled, unless task_store_eager_result is enabled.
  self._ensure_not_eager()

protocol 1
 EagerResult f3692d1a-e84a-4480-a4ab-6ab16734f844 SUCCESS
 parent ->
   GroupResult 216cfd37-a45a-4e20-86a9-aeb4c0a01fd8 None
   children: 2
     EagerResult 60c9c876-652a-4a5a-a6ca-5432e34545bb SUCCESS
     EagerResult b17c91bf-c32d-443d-9ffb-df7ef46e8daf SUCCESS
   parent ->
     EagerResult 8ad10a25-3f25-4312-9e94-26ea2332c1a4 SUCCESS

get [[1, 1], [1, 1]]
tuple (('f3692d1a-e84a-4480-a4ab-6ab16734f844', (('216cfd37-a45a-4e20-86a9-aeb4c0a01fd8', (('8ad10a25-3f25-4312-9e94-26ea2332c1a4', None), None)), [(('60c9c876-652a-4a5a-a6ca-5432e34545bb', None), None), (('b17c91bf-c32d-443d-9ffb-df7ef46e8daf', None), None)])), None)
restored
 AsyncResult f3692d1a-e84a-4
…[898 chars omitted]…
a7-30cf-40f6-b414-f0a677646232', None), None)])), None)
restored
 AsyncResult 47a480d1-7821-4900-a421-2e516c4ce27a PENDING
 parent ->
   GroupResult 6c324ee3-98fd-4e74-8bfc-8e54a142e7ef None
   children: 2
     AsyncResult a7fa0110-ca2a-4085-b01a-831ee658a418 PENDING
     AsyncResult 7f8f01a7-30cf-40f6-b414-f0a677646232 PENDING
   parent ->
     AsyncResult fc6efe91-08b4-4f18-b01d-bd26a2fdcfc0 PENDING

@property
    def date_done(self):
        """UTC date and time."""
        date_done = self._get_task_meta().get('date_done')
        if date_done and not isinstance(date_done, datetime.datetime):
            return isoparse(date_done)
        return date_done

    @property
    def retries(self):
        return self._get_task_meta().get('retries')

    @property
    def queue(self):
        return self._get_task_meta().get('queue')


@Thenable.register
class ResultSet(ResultBase):
    """A collection of results.

    Arguments:
        results (Sequence[AsyncResult]): List of result instances.
    """

    _app = None

    #: List of results in in the set.
    results = None

    def __init__(self, results, app=None, ready_barrier=None, **kwargs):
        self._app = app
        self.results = results
        self.on_ready = promise(args=(proxy(self),))
        self._on_full = 
…[8628 chars omitted]…
lf.id} [{", ".join(r.id for r in self.results)}]>'

    def __str__(self):
        """`str(self) -> self.id`."""
        return str(self.id)

    def __hash__(self):
        """`hash(self) -> hash(self.id)`."""
        return hash(self.id)

    def as_tuple(self):
        return (
            (self.id, self.parent and self.parent.as_tuple()),
            [r.as_tuple() for r in self.results]
        )

chain tasks: [repro.ident(1), group((repro.ident(2), ident(3))), %repro.ident((ident(4), ident(5)), 6)]
0 Signature repro.ident(1)
1 group group((repro.ident(2), ident(3)))
2 _chord %repro.ident((ident(4), ident(5)), 6)

step 0 Signature
before last None None last.parent None
arg None
res EagerResult c9154815-61c8-443a-be61-5d916befc52f
after set last EagerResult c9154815-61c8-443a-be61-5d916befc52f parent None None

step 1 group

before last EagerResult c9154815-61c8-443a-be61-5d916befc52f last.parent None
arg (1,)
res GroupResult 09cd9d66-a51e-4c57-8c4f-e279eb691db2
res children ['449b3397-22a6-4e28-8f18-6d02a08a2259', 'de7f9ed0-6e99-4424-b535-fee9a92e9034']
after set last GroupResult 09cd9d66-a51e-4c57-8c4f-e279eb691db2 parent EagerResult c9154815-61c8-443a-be61-5d916befc52f

step 2 _chord
before last GroupResult 09cd9d66-a51e-4c57-8c4f-e279eb691db2 last.parent EagerResult
arg ([1, 1],)
res EagerResult cce5c8b4-6819-42e6-b079-2834dfdd1008
after set last EagerResult cce5c8b4-6819-42e6-b079-2834dfdd1008 parent GroupResult 09cd9d66-a51e-4c57-8c4f-e279eb691db2

class chain(_chain):
    """Chain tasks together.

    Each tasks follows one another,
    by being applied as a callback of the previous task.

    Note:
        If called with only one argument, then that argument must
        be an iterable of tasks to chain: this allows us
        to use generator expressions.

    Example:

        This is effectively :math:`((2 + 2) + 4)`:

        .. code-block:: pycon

            >>> res = chain(add.s(2, 2), add.s(4))()
            >>> res.get()
            8

        Calling a chain will return the result of the last task in the chain.
        You can get to the other tasks by following the ``result.parent``'s:

        .. code-block:: pycon

            >>> res.parent.get()
            4

        Using a generator expression:

        .. code-block:: pycon

            >>> lazy_chain = chain(add.s(i) for i in range(10))
            >>> res =
…[3465 chars omitted]…
s)

    def apply_async(self, args=None, kwargs=None, **opts):
323:    def __init__(self, task=None, args=None, kwargs=None, options=None,
947:    def __init__(self, *tasks, **options):
1393:    def __init__(self, task, it, **options):
1446:    def __init__(self, task, it, n, **options):
1580:    def __init__(self, *tasks, **options):
2064:    def __init__(self, header, body=None, task='celery.chord',

34:        assert maybe_unroll_group(g) is g
36:        assert maybe_unroll_group(g) is g
73:        def replaced_group(self, x, y):
74:            return self.replace(group(add.si(x, y), mul.si(x, y)))
79:        def replace_with_group(self, x, y):
80:            return self.replace(group(add.si(x, y), mul.si(x, y)))
86:            return self.replace(group(add.si(x, y) | mul.s(y), add.si(x, y)))
384:        assert x.group()
385:        assert len(x.group().tasks) == 10
465:        g1 = group([self.add.s(2, 2), self.add.s(4, 4)])
466:        g2 = group([self.add.s(3, 3), self.add.s(5, 5)])
472:            group([self.add.s(2, 2), self.add.s(4, 4)], app=self.app),

473:            group(app=self.app),
474:            group(app=self.app),
483:        c = self.add.s(1, 1) | group(
486:        ) | group(app=self.app)
497:        c = group([self.add.s(2, 2)], app=self.app) | group(tasks(),
…[6263 chars omitted]…
signature('t'))
1977:        c = chord([signature('header')], group(signature('body')))
1985:        group(signature('s1'), signature('s2'))
2006:        header = group([self.add.s(1, 1), self.add.s(2, 2)])
2027:        header = group([self.add.s(1, 1), self.add.s(2, 2)])
2057:        header = group([self.add.s(1, 1), self.add.s(2, 2)])
2079:        header = group([self.add.s(1, 1), self.add.s(2, 2)])

new_chain = c | g  # g should be chained with the body of c[0]
        assert isinstance(new_chain, _chain)
        assert isinstance(new_chain.tasks[0].body, _chain)

    def test_chain_of_chord_upgrade_on_chaining(self):
        c = chord([signature('header')], group(signature('body')))
        c = chain(c)
        t = signature('t')
        new_chain = c | t  # t should be chained with the body of c[0] and create a new chord
        assert isinstance(new_chain, _chain)
        assert isinstance(new_chain.tasks[0].body, chord)

    @pytest.mark.parametrize(
        "group_last_task",
        [False, True],
    )
    def test_chain_of_chord_upgrade_on_chaining__protocol_2(
            self, group_last_task):
        c = chain(
            group([self.add.s(i, i) for i in range(5)], app=self.app),
            group([self.add.s(i, i) for i in range(10, 15)], app=self.app),
    
…[8029 chars omitted]…
.s(3)
        result = c.freeze()
        assert c.id is not None
        assert c.id == result.id

    def test_chain_id_with_explicit_id_preserved(self):
        """Chain.freeze(explicit_id) must set chain.id to that value."""
        c = self.add.s(1, 1) | self.add.s(2)
        result = c.freeze('my-explicit-id')
        assert c.id == 'my-explicit-id'
        assert result.id == 'my-explicit-id'

celery/worker/strategy.py:67:    """Convert Task message protocol 1 arguments to protocol 2.

1 _chain tasks ['Signature', 'group', 'group', 'Signature']
 AsyncResult 4b663f7c-6b9e-4cd2-af19-eac09f7fd077
 parent ->
   GroupResult 8e5f7a92-0d64-4852-9d65-58a1bd2439c9
   children: 2
     AsyncResult 98d41111-6b1e-40fb-97cf-8bc24f440c69
     AsyncResult 719f82b1-7519-416f-9d56-948c8916c6ee
   parent ->
     GroupResult 79bcf6da-651f-4674-8f9e-34f94ee582af
     children: 2
       AsyncResult 68960c7a-d0e3-4314-98fc-a88cd4fb0cab
       AsyncResult 340a4110-652e-453e-a290-ba1d7f1b91eb

     parent ->
       AsyncResult cde9a84f-4787-465b-8640-7e2b601e2523
restored:
 AsyncResult 4b663f7c-6b9e-4cd2-af19-eac09f7fd077
 parent ->
   GroupResult 8e5f7a92-0d64-4852-9d65-58a1bd2439c9
   children: 2
     AsyncResult 98d41111-6b1e-40fb-97cf-8bc24f440c69
     AsyncResult 719f82b1-7519-416f-9d56-948c8916c6ee
   parent ->
     GroupResult 79bcf6da-651f-4674-8f9e-34f94ee582af
     children: 2
  
…[3016 chars omitted]…
8-fe2415bce00e
   children: 2
     AsyncResult e3989152-93ba-4190-9231-d794df6f0551
     AsyncResult 41eb46f0-e872-4bec-8b4c-265e99151595
   parent ->
     GroupResult c9c3d97b-15cc-4c6b-b7d1-dd4d1e35501d
     children: 2
       AsyncResult e5e6eb17-2088-455b-ab58-0b2b85bdab0a
       AsyncResult afe3e026-09aa-4edf-af1d-2998505da640
     parent ->
       AsyncResult 458f8705-6102-415b-8071-47588b8ef68b

1 _chain tasks ['Signature', 'group', 'group', 'Signature']
Traceback (most recent call last):
  File "/tmp/repro_direct_chain_eager.py", line 31, in <module>
    res=c.apply_async()
        ^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1044, in apply_async
    return self.apply(args, kwargs, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1300, in apply
    last and (last.get(),), **dict(self.options, **options))
              ^^^^^^^^^^
  File "/repo/celery/result.py", line 735, in get

    return (self.join_native if self.supports_native_join else self.join)(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join
    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: add() takes from 1 to 2 positional arguments but 3 were given

1 _chain tasks ['Signature', 'group', 'group', 'Signature']
 EagerResult 30e75fc7-ffd3-4207-815b-9af17b432e0e
 parent ->
   GroupResult 31175196-742c-47d7-80ff-e612751ac26d
   children: 2
     EagerResult 011d4ab2-92f9-4017-beb4-74f127b08c65
     EagerResult 1d379709-b076-43d7-98c3-2daa85d22e53
   parent ->
     GroupResult e1f60a81-dc9b-4fd1-86d6-cdf41ffb3abd
     children: 2
       EagerResult e72279bc-8da5-441c-a7b3-de206e32ccc1
       EagerResult 719d82bd-c466-49e5-a4e7-98915f11f8fc

     parent ->
       EagerResult 9afce20e-d71e-4bc9-82e5-9adf81e07070
restored:
 AsyncResult 30e75fc7-ffd3-4207-815b-9af17b432e0e
 parent ->
   GroupResult 31175196-742c-47d7-80ff-e612751ac26d
   children: 2
     AsyncResult 011d4ab2-92f9-4017-beb4-74f127b08c65
     AsyncResult 1d379709-b076-43d7-98c3-2daa85d22e53
   parent ->
     GroupResult e1f60a81-dc9b-4fd1-86d6-cdf41ffb3abd
     children: 2
  
…[3016 chars omitted]…
c-d446a24a19af
   children: 2
     AsyncResult d81ee097-2eb8-41d9-8bb9-9ed282f6b4fd
     AsyncResult 7a6f2c5b-8638-474e-b181-fd6f8946db9c
   parent ->
     GroupResult 21591cfc-9b2b-4c1a-893d-44510ef2992a
     children: 2
       AsyncResult 8d18839a-f726-4cf6-ad8c-448c772824e9
       AsyncResult 4a3f4e31-2e34-4816-9431-dbcf494364bd
     parent ->
       AsyncResult 3fac62e8-2251-4f66-a77c-4ae9d9332c3e

390:        x.apply_async()
396:        self.app.conf.task_always_eager = True
420:        result = c.apply_async()
478:        result = c.apply_async()
488:        result = c.apply_async()
727:        self.app.conf.task_always_eager = True
731:        self.app.conf.task_always_eager = True
741:                return (self.add.s(4, 4) | self.add.s(8)).apply_async()
771:        assert chain(app=self.app).apply_async() is None
1055:        x.apply_async()
1059:        x.apply_async()
1064:        g2.apply_async()

1113:        res = x.apply_async()
1175:        x.apply_async()
1216:        self.app.conf.task_always_eager = True
1224:        self.app.conf.task_always_eager = True
1230:        self.app.conf.task_always_eager = True
1241:            res_obj = group_sig.apply_async()
1251:            res_obj = group_sig.apply_async()
1268:            res_obj = group_sig.apply_async()
1270:  
…[655 chars omitted]…
y_async()
1393:        # tasks are never actually run due to our mocking of `apply_async()`
1414:            res_obj = group_sig.apply_async()
1416:        # are never actually run due to our mocking of `apply_async()`
1437:        self.app.conf.task_always_eager = True
1759:        self.app.conf.task_always_eager = True
1773:                return chord([self.add.s(4, 4)], finalize.s()).apply_async()

d = dict(x)
        d['subtask_type'] = "chunks_subclass"
        isinstance(chunks_subclass.from_dict(d), chunks_subclass)
        isinstance(chunks_subclass.from_dict(d).clone(), chunks_subclass)

    def test_chunks(self):
        x = self.add.chunks(range(100), 10)
        assert dict(chunks.from_dict(dict(x), app=self.app)) == dict(x)

        assert x.group()
        assert len(x.group().tasks) == 10

        x.group = Mock()
        gr = x.group.return_value = Mock()

        x.apply_async()
        gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
        gr.apply_async.reset_mock()
        x()
        gr.apply_async.assert_called_with((), {}, route_name=self.add.name)

        self.app.conf.task_always_eager = True
        chunks.apply_chunks(app=self.app, **x['kwargs'])


class test_chain(CanvasCase):

    def test_chain_of_chain_with_a_single_task(
…[8511 chars omitted]…
ver actually run due to our mocking of `apply_async()`
        assert mock_apply_async.call_count == gchild_count
        assert len(res_obj.children) == len(group_sig.tasks)
        # We must have set the chord size for the group of tasks which makes up
        # the header of the `child_chord`, just before we apply the last task.
        mock_set_chord_size.assert_called_once_with(ANY, gchild_count)

celery/app/trace.py:109:IGNORED = states.IGNORED
celery/app/trace.py:110:REJECTED = states.REJECTED
celery/events/state.py:75:    'rejected': states.REJECTED,

"""Built-in task states.

.. _states:

States
------

See :ref:`task-states`.

.. _statesets:

Sets

----

.. state:: READY_STATES

READY_STATES
~~~~~~~~~~~~

Set of states meaning the task result is ready (has been executed).

.. state:: UNREADY_STATES

UNREADY_STATES
~~~~~~~~~~~~~~

Set of states meaning the task result is not ready (hasn't been executed).

.. state:: EXCEPTION_STATES

EXCEPTION_STATES
~~~~~~~~~~~~~~~~

Set of states meaning the task returned an exception.

.. state:: PROPAGATE_STATES

PROPAGATE_STATES
~~~~~~~~~~~~~~~~

Set of exception states that should propagate exceptions to the user.

.. state:: ALL_STATES

ALL_STATES
~~~~~~~~~~

Set of all possible states.

Misc
----

"""

__all__ = (
    'PENDING', 'RECEIVED', 'STARTED', 'SUCCESS', 'FAILURE',
    'REVOKED', 'RETRY', 'IGNORED', 'READY_STATES', 'UNREADY_STATES',
    'EXCEPTION_STATES', 'PROPAGATE_STATES', 'prece
…[2022 chars omitted]…
REJECTED'
#: Task is waiting for retry.
RETRY = 'RETRY'
IGNORED = 'IGNORED'

READY_STATES = frozenset({SUCCESS, FAILURE, REVOKED})
UNREADY_STATES = frozenset({PENDING, RECEIVED, STARTED, REJECTED, RETRY})
EXCEPTION_STATES = frozenset({RETRY, FAILURE, REVOKED})
PROPAGATE_STATES = frozenset({FAILURE, REVOKED})

ALL_STATES = frozenset({
    PENDING, RECEIVED, STARTED, SUCCESS, FAILURE, RETRY, REVOKED,
})

protocol 1
Traceback (most recent call last):
  File "/tmp/repro_eager_groups2.py", line 33, in <module>
    res = c.apply_async()
          ^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1045, in apply_async
    return self.apply(args, kwargs, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1311, in apply
    if res.state in (states.IGNORED, states.REJECTED):
       ^^^^^^^^^
AttributeError: 'GroupResult' object has no attribute 'state'. Did you mean: 'save'?

/repo/celery/backends/base.py:752: RuntimeWarning: Results are not stored in backend and should not be retrieved when task_always_eager is enabled, unless task_store_eager_result is enabled.
  self._ensure_not_eager()

protocol 1
 EagerResult 7dcdb67e-2e06-444d-8221-2b7bf0f6382b SUCCESS
 parent ->
   GroupResult 236f55eb-e38b-46be-9f9b-5efdceb0a036 None
   children: 2
     EagerResult fae70e6c-f53e-4955-b3e8-e5bff7910954 SUCCESS
     EagerResult 8a58a65b-1c0d-4e15-9532-2fd29dac02a3 SUCCESS
   parent ->
     EagerResult d4ca35fc-641e-44cc-922c-fdc8b36c999b SUCCESS

get [[1, 1], [1, 1]]
tuple (('7dcdb67e-2e06-444d-8221-2b7bf0f6382b', (('236f55eb-e38b-46be-9f9b-5efdceb0a036', (('d4ca35fc-641e-44cc-922c-fdc8b36c999b', None), None)), [(('fae70e6c-f53e-4955-b3e8-e5bff7910954', None), None), (('8a58a65b-1c0d-4e15-9532-2fd29dac02a3', None), None)])), None)
restored
 AsyncResult 7dcdb67e-2e06-4
…[898 chars omitted]…
03-a267-483a-96d3-1e0a90e75a1a', None), None)])), None)
restored
 AsyncResult 5be87f6b-e779-4112-a188-7557ddaea23a PENDING
 parent ->
   GroupResult 9f75d536-3b05-43c1-86bc-d0df0cf154ec None
   children: 2
     AsyncResult a3ef2df6-1177-4ba4-852c-11ecd841907d PENDING
     AsyncResult a4fc1f03-a267-483a-96d3-1e0a90e75a1a PENDING
   parent ->
     AsyncResult 3cc8c9c9-6261-4bde-a068-46dfc3c1030c PENDING

t/unit/tasks/test_chord.py:281:    def test_apply(self):
t/unit/tasks/test_canvas.py:276:    def test_apply_async_when_not_registered(self):
t/unit/tasks/test_canvas.py:355:    def test_apply(self):
t/unit/tasks/test_canvas.py:418:    def test_apply_async(self):
t/unit/tasks/test_canvas.py:582:        assert c.apply().get() == ['x0y0', 'x1y1', 'foo', 'y', 'z']
t/unit/tasks/test_canvas.py:591:        assert c.apply().get() == [
t/unit/tasks/test_canvas.py:676:    def test_apply_options(self):
t/unit/tasks/test_canvas.py:703:    def test_apply_options_none(self):
t/unit/tasks/test_canvas.py:749:    def test_apply(self):
t/unit/tasks/test_canvas.py:751:        res = x.apply()
t/unit/tasks/test_canvas.py:761:        res = x.apply(kwargs={'x': 1, 'y': 1}).get()
t/unit/tasks/test_canvas.py:765:        x = chain(self.add.s(1, 2)).apply()

t/unit/tasks/test_canvas.py:826:        assert x.apply
…[6539 chars omitted]…
not_ready(self):
        class NeverReady(TSR):
            is_ready = False

        with self._chord_context(NeverReady, interval=10, max_retries=30) \
                as (cb, retry, _):
            cb.type.apply_async.assert_not_called()
            # did retry
            retry.assert_called_with(countdown=10, max_retries=30)

    def test_when_not_ready_with_configured_chord_retry_interval(self):

class test_chord(ChordCase):

    def test_eager(self):
        from celery import chord

        @self.app.task(shared=False)
        def addX(x, y):
            return x + y

        @self.app.task(shared=False)
        def sumX(n):
            return sum(n)

        self.app.conf.task_always_eager = True
        x = chord(addX.s(i, i) for i in range(10))
        body = sumX.s()
        result = x(body)
        assert result.get() == sum(i + i for i in range(10))

    def test_apply(self):
        self.app.conf.task_always_eager = False
        from celery import chord

        m = Mock()
        m.app.conf.task_always_eager = False
        m.AsyncResult = AsyncResult
        prev, chord.run = chord.run, m
        try:
            x = chord(self.add.s(i, i) for i in range(10))
            body = self.add.s(2)
            result = x(body)
            assert result.id
            # 
…[3589 chars omitted]…
    assert res
        assert not res.results

    def test_apply_async_with_parent(self):
        _task_stack.push(self.add)
        try:
            self.add.push_request(called_directly=False)
            try:
                assert not self.add.request.children
                x = group([self.add.s(4, 4), self.add.s(8, 8)])
                res = x()
                assert self.add.request.children

/repo/celery/backends/base.py:752: RuntimeWarning: Results are not stored in backend and should not be retrieved when task_always_eager is enabled, unless task_store_eager_result is enabled.
  self._ensure_not_eager()

protocol 1
 EagerResult 44ad934a-41c0-49cb-8622-7bea7ad266a6 SUCCESS
 parent ->
   GroupResult 88b4ca94-4692-4b72-b847-6627fecc9e3f None
   children: 2
     EagerResult 68c5f2a5-40d3-4659-a68a-e6b330d6144c SUCCESS
     EagerResult b6d6e63c-25df-49dd-9f67-599295781f9c SUCCESS
   parent ->
     GroupResult df62fd58-eaa0-4303-bd29-4e98708df6ca None

     children: 2
       EagerResult 01bf20eb-758f-48e5-9a24-e61f13b7932d SUCCESS
       EagerResult c3a063ee-6a52-4643-a01d-1c01bf43af4b SUCCESS
     parent ->
       EagerResult 36b2f47a-33fa-4b0e-a600-3dac34ed3554 SUCCESS
get [[1, 1], [1, 1]]
tuple (('44ad934a-41c0-49cb-8622-7bea7ad266a6', (('88b4ca94-4692-4b72-b847-6627fecc9e
…[2098 chars omitted]…
t eb4dc4d7-f8f5-468b-89dc-2c03f2fbaecf PENDING
     AsyncResult 3cf7e102-e29e-4276-8499-6c153eb6b1b3 PENDING
   parent ->
     GroupResult a128cd9c-a86f-48b0-946f-e4ca564bcdd7 None
     children: 2
       AsyncResult 0561bd4a-a7f5-4709-a842-4362be4f597b PENDING
       AsyncResult c9e822d6-2692-433d-9a03-4ef912d752e3 PENDING
     parent ->
       AsyncResult ea88afeb-da41-499e-a3a3-8129846d6029 PENDING

1 repro.ignore EagerResult IGNORED get= None side_effects= []
 apply: IGNORED None []
1 repro.reject EagerResult REJECTED get= None side_effects= []
 apply: REJECTED None []
2 repro.ignore EagerResult IGNORED get= None side_effects= []
 apply: IGNORED None []
2 repro.reject EagerResult REJECTED get= None side_effects= []
 apply: REJECTED None []
ok

fail 0

582:        assert c.apply().get() == ['x0y0', 'x1y1', 'foo', 'y', 'z']
591:        assert c.apply().get() == [
751:        res = x.apply()
761:        res = x.apply(kwargs={'x': 1, 'y': 1}).get()
765:        x = chain(self.add.s(1, 2)).apply()
826:        assert x.apply().get() == 3
1050:        res = x.apply()
1112:        x.apply()
1212:        res = x.apply(kwargs={'x': 1, 'y': 1}).get()

protocol 1 tasks ['Signature', 'group', '_chord']
 AsyncResult 5cc946a9-4e3c-49b7-965b-b23c5848723a
 parent
   GroupResult be5941ac-6956-4cad-a552-27bdf8f2baac
   children 2
     AsyncResult a0520990-dd19-4e15-bf61-3760da2678d8
     AsyncResult 26ead182-3480-4f94-a235-c9e96b1bdbe3
   parent
     GroupResult 1a814b30-88b2-47d5-aa23-51d3ac53fbdc
     children 2
       AsyncResult 92e08ce9-8203-4832-bf43-4624bc969037
       AsyncResult 9373c4a8-0ccb-46aa-8c6f-e42775a7e43f

     parent
       AsyncResult 9a1b7a6d-7d48-46fc-849b-c8ba44922b87
restored
 AsyncResult 5cc946a9-4e3c-49b7-965b-b23c5848723a
 parent
   GroupResult be5941ac-6956-4cad-a552-27bdf8f2baac
   children 2
     AsyncResult a0520990-dd19-4e15-bf61-3760da2678d8
     AsyncResult 26ead182-3480-4f94-a235-c9e96b1bdbe3
   parent
     GroupResult 1a814b30-88b2-47d5-aa23-51d3ac53fbdc
     children 2
       AsyncResult 92e08ce9-8203-4
…[784 chars omitted]…
42ea-984f-c0e2268f7a49
   children 2
     AsyncResult 7e8d5250-cefc-4fb7-bad8-3dc1ce1fb157
     AsyncResult e124df30-8384-442c-bdd2-86926521f107
   parent
     GroupResult e2b20a18-52e4-4bad-b6ef-3bda6d447fd8
     children 2
       AsyncResult 56f7f897-8d8e-4e90-8071-5dc6764d4949
       AsyncResult 41ce5bb1-df80-4e14-9bce-ded70a8e0ec2
     parent
       AsyncResult 9ed7762a-a1a2-4a60-8122-8ef21b68a784

fail 0

80:    from celery.canvas import (chain, chord, chunks, group, maybe_signature, signature, subtask, xmap,  # noqa
163:        'celery.canvas': [
ok

celery/app/task.py:10:from celery import current_app, states
celery/app/trace.py:19:from celery import current_app, group, signals, states
celery/backends/asynchronous.py:15:from celery import states
celery/backends/base.py:23:from celery import current_app, group, maybe_signature, states
celery/backends/cassandra.py:4:from celery import states
celery/backends/database/models.py:7:from celery import states
celery/backends/database/__init__.py:5:from celery import states
celery/backends/elasticsearch.py:7:from celery import states
celery/backends/mongodb.py:8:from celery import states
celery/backends/redis.py:13:from celery import states
celery/backends/rpc.py:13:from celery import states
celery/canvas.py:26:from celery import states

celery/contrib/testing/manager.py:12:from celery import states
celery/events/cursesmon.py:12:from celery import VERSION_BANNER, states
celery/events/state.py:31:from celery import states
celery/worker/request.py:17:from celery import current_app, signals, states

............u........................................................... [ 21%]
....................................................x..x................ [ 42%]
............................uu.uu.......uuu.uuu.uuu.uuu.uuu.uuu.uuu.uuu. [ 56%]
........................................s............................... [ 77%]
........................................................................ [ 99%]
..                                                                       [100%]
=============================== warnings summary ===============================
t/unit/tasks/test_canvas.py::test_chain::test_chain_errbacks_propagated_to_chord_body
  /repo/celery/canvas.py:1270: CPendingDeprecationWarning: task_allow_error_cb_on_chord_header=False is pending deprecation in a future release of Celery.
  Please test the new behavior by setting task_allow_error_cb_on_chord_header to True and repor
…[936 chars omitted]…
new behavior by setting task_allow_error_cb_on_chord_header to True and report any concerns you might have in our issue tracker before we make a final decision regarding how errbacks should behave when used with chords.

    chord_sig.link_error(errback_sig)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
330 passed, 1 skipped, 2 xfailed, 3 warnings, 29 subtests passed in 6.83s

1 ['Signature', 'group', 'group']
 GroupResult b9b41872-20c9-47f2-81a1-052aabaea2ff
 children 2
   EagerResult 45878f82-1eeb-46c8-bf0a-101415538651
   EagerResult 9035bdf2-470f-4111-a060-c13e969d4ab4
 parent
   GroupResult 74df4f69-9d69-4c06-b392-a954e7eb6113
   children 2
     EagerResult 70dae11b-a460-46cb-8971-e389b55fda6a
     EagerResult 97694a63-0b60-4dbd-beee-e4480517d7da
   parent
     EagerResult 77f78b74-c194-4b3e-a46b-c68a5bca33c0

restored
 GroupResult b9b41872-20c9-47f2-81a1-052aabaea2ff
 children 2
   AsyncResult 45878f82-1eeb-46c8-bf0a-101415538651
   AsyncResult 9035bdf2-470f-4111-a060-c13e969d4ab4
 parent
   GroupResult 74df4f69-9d69-4c06-b392-a954e7eb6113
   children 2
     AsyncResult 70dae11b-a460-46cb-8971-e389b55fda6a
     AsyncResult 97694a63-0b60-4dbd-beee-e4480517d7da
   parent
     AsyncResult 77f78b74-c194-4b3e-a46b-c68a5bca33c0

 1 ['Signature', 'Signature
…[2212 chars omitted]…
29-797c-40b7-9ad8-ac6abdbe905f
 parent
   GroupResult a795582a-12d2-4329-a1ff-0537be7fc92d
   children 2
     AsyncResult a65c8863-aae8-45cb-9768-0534aa5fe4db
     AsyncResult 0c165318-fd46-435f-8a7d-57550031866d
   parent
     GroupResult c010aa5f-b338-47a3-a670-66018986f317
     children 2
       AsyncResult a62df100-a71f-4f0e-b8ab-93dd087a73bf
       AsyncResult adbb1844-3883-4bf6-abd3-7c72460e9a9a

AsyncResult final
 parent
   GroupResult g2
   children ['g2-0', 'g2-1', 'g2-2', 'g2-3']
     AsyncResult g2-0
     AsyncResult g2-1
     AsyncResult g2-2
     AsyncResult g2-3
   parent
     GroupResult g1
     children ['g1-0', 'g1-1', 'g1-2']
       AsyncResult g1-0

       AsyncResult g1-1
       AsyncResult g1-2
restored
 AsyncResult final
 parent
   GroupResult g2
   children ['g2-0', 'g2-1', 'g2-2', 'g2-3']
     AsyncResult g2-0
     AsyncResult g2-1
     AsyncResult g2-2
     AsyncResult g2-3
   parent
     GroupResult g1
     children ['g1-0', 'g1-1', 'g1-2']
       AsyncResult g1-0
       AsyncResult g1-1
       AsyncResult g1-2
equal tree True

474:    def freeze(self, _id=None, group_id=None, chord=None,
1097:    def freeze(self, _id=None, group_id=None, chord=None,
1850:    def _freeze_group_tasks(self, _id=None, group_id=None, chord=None,
1901:    def freeze(self, _id=None, group_id=None, chord=None,
2105:    def freeze(self, _id=None, group_id=None, chord=None,
            if isinstance(task, group):
                stack.extendleft(task.tasks)
            else:
                new_tasks.append(task)
                yield task.freeze(group_id=group_id,
                                  chord=chord, root_id=root_id,
                                  parent_id=parent_id,

                                  group_index=group_index)
                group_index += 1

    def __repr__(self):
        if self.tasks:
            return remove_repeating_from_task(
                self.tasks[0]['task'],
                f'group({self.
…[5991 chars omitted]…
roup tasks.
                The generator yields tuples of the form ``(task, AsyncResult, group_id)``.
        """
        for index, task in enumerate(tasks):
            if isinstance(task, CallableSignature):
                # local sigs are always of type Signature, and we
                # clone them to make sure we don't modify the originals.
                task = task.clone()
            else:

# remove task_id and use that as the group_id,
        # if we don't remove it then every task will have the same id...
        options = {**self.options, **{
            k: v for k, v in options.items()
            if k not in self._IMMUTABLE_OPTIONS or k not in self.options
        }}
        options['group_id'] = group_id = (
            options.pop('task_id', uuid()))
        return options, group_id, options.get('root_id')

    def _freeze_group_tasks(self, _id=None, group_id=None, chord=None,
                            root_id=None, parent_id=None, group_index=None):

        """Freeze the tasks in the group.

        Note:
            If the group tasks are created from a generator, the tasks generator would
            not be exhausted, and the tasks would be frozen lazily.

        Returns:
            tuple: A tuple of the group id, and the AsyncResult of each of the
…[3509 chars omitted]…
 and will return it in the new_tasks argument.

        Arguments:
            new_tasks (list): The list to append the flattened tasks to.
            group_id (str): The group_id to use for the tasks.
            chord (Chord): The chord to use for the tasks.
            root_id (str): The root_id to use for the tasks.
            parent_id (str): The parent_id to use for the tasks.

        Yields:

protocol 1 tasks ['Signature', 'group', '_chord']
result tree
 AsyncResult ad01f990-fb97-4b5d-884c-86d8ba5dcfc1
 parent
   GroupResult 4d0c0e2c-4c78-41cd-adf7-24fd302006df
   children type list len? 2
     AsyncResult 0d82f9f6-5738-4c33-a77e-06bed15bab12
     AsyncResult 20c9f6b6-669d-4867-ae25-2e9bdbb3f80c
   parent
     GroupResult 70cc9395-c94d-4499-ae2d-749be1839513
     children type _regen len? 2
       AsyncResult 7344422a-e4e4-4be9-8644-5ef2a28e64ba

       AsyncResult e635d592-8d8e-452b-af2d-f8c6eac70ed9
     parent
       AsyncResult 96bb412d-ff37-44f6-87e5-adff2374b681
as_tuple
(('ad01f990-fb97-4b5d-884c-86d8ba5dcfc1', (('4d0c0e2c-4c78-41cd-adf7-24fd302006df', (('70cc9395-c94d-4499-ae2d-749be1839513', (('96bb412d-ff37-44f6-87e5-adff2374b681', None), None)), [(('7344422a-e4e4-4be9-8644-5ef2a28e64ba', None), None), (('e635d592-8d8e-452b-af2d-f8c6eac70ed9', None), None)])), [(
…[3770 chars omitted]…
-44bc-ac5f-432d9e872d13', (('0a7d73ac-8821-4144-ab22-6fc36cedb663', (('b2545658-75cd-4823-9040-9da2ba6666c1', (('bd461ec2-296d-426e-af14-ed0ee36888ea', None), None)), [(('3fddd755-5082-4b42-b95d-e1d40503a421', None), None), (('00eba6b5-fb77-4075-8c5a-921f0d4133b2', None), None)])), [(('bd4b991e-5343-4b2c-afdb-948fa677de5d', None), None), (('0edd8e55-15be-479e-a146-66099c6a8a1a', None), None)])), None)

protocol 1 tasks ['Signature', 'group', '_chord']
result tree
 EagerResult 3a5956ef-8eaf-4b0c-aafa-4c99f44c7f77
 parent
   GroupResult 68b7d91d-0e56-43b8-8c79-c3f88aab4bb7
   children type list len? 2
     EagerResult 07982db3-f53c-4188-ac41-c4f1a7ccf6f5
     EagerResult bc05cc05-1767-455d-8dcf-e12ccb8c174c
   parent
     GroupResult 8584938c-41e2-431f-9870-daf9ccec064a
     children type list len? 2
       EagerResult 87216c37-81ed-498c-8b62-226ca692e04a

       EagerResult 39e6af33-db79-47be-8a29-bf9b961b08b4
     parent
       EagerResult d54cf1eb-0841-4348-bd98-6fb5ed4e5d3a
as_tuple
(('3a5956ef-8eaf-4b0c-aafa-4c99f44c7f77', (('68b7d91d-0e56-43b8-8c79-c3f88aab4bb7', (('8584938c-41e2-431f-9870-daf9ccec064a', (('d54cf1eb-0841-4348-bd98-6fb5ed4e5d3a', None), None)), [(('87216c37-81ed-498c-8b62-226ca692e04a', None), None), (('39e6af33-db79-47be-8a29-bf9b961b08b4', None), None)])), [(('
…[3762 chars omitted]…
-424d-b079-8118bc79be07', (('a53774e8-c5a2-485e-9806-a6c06610d4bc', (('19cfae72-a7aa-4ace-b813-b099a8cbc13c', (('074c7c52-26a5-4b14-a5c6-f9b5744e4461', None), None)), [(('ed647443-ad6a-43cc-856b-aa478baebdcf', None), None), (('9c17d11f-0da3-4ea6-8078-138f0f7345f9', None), None)])), [(('14507785-072c-4c0c-8a53-361363e00eee', None), None), (('80b03600-3f13-4516-85f8-8bab2e142a7f', None), None)])), None)

celery/canvas.py:1266:                    prev_res.parent = res
celery/canvas.py:1309:                node.parent = last
celery/canvas.py:2132:                    node.parent = header_result
celery/canvas.py:2204:        node.parent = header_result
celery/canvas.py:2302:        bodyres.parent = header_result
celery/result.py:98:        self.parent = parent
celery/result.py:952:        self.parent = parent
celery/result.py:988:                other.parent == self.parent
celery/utils/log.py:93:        logger_.parent = parent_logger

isinstance(other, Signature)):
            # chord | task ->  attach to body
            sig = self.clone()
            sig.body = sig.body | other
            return sig
        elif isinstance(other, group) and len(other.tasks) == 1:
            # chord | group -> chain with chord body.
            # unroll group with one member
            other = maybe_unroll_group(other)
            sig = self.clone()
            sig.body = sig.body | other
            return sig

        else:
            return super().__or__(other)

    def freeze(self, _id=None, group_id=None, chord=None,
               root_id=None, parent_id=None, group_index=None):
        # pylint: disable=redefined-outer-name
        #   XXX chord is also a class in outer scope.
        if not isinstance(self.tasks, group):
            self.tasks = group(self.tasks, app=self.app)
        # first freeze all
…[2455 chars omitted]…
        body.delay([])
            header_result = self.app.GroupResult(*header_result_args)

        bodyres.parent = header_result
        return bodyres

    def clone(self, *args, **kwargs):
        signature = super().clone(*args, **kwargs)
        # need to make copy of body
        try:
            signature.kwargs['body'] = maybe_signature(
                signature.kwargs['body'], clone=True)

LOW async 1 leaf chain _chain actual 0 expected 2 tasks ['Signature', 'group', 'group', '_chain']
LOW async 1 group leaf public actual 1 expected 0 tasks ['_chord']
LOW async 1 group leaf _chain actual 1 expected 3 tasks ['group', 'group', 'group', 'Signature']
LOW async 1 group chain public actual 1 expected 0 tasks ['_chord']
LOW async 1 group chain _chain actual 0 expected 3 tasks ['group', 'group', 'group', '_chain']
LOW async 1 group chord public actual 1 expected 0 tasks ['_chord']
LOW async 1 group chord _chain actual 1 expected 3 tasks ['group', 'group', 'group', '_chord']
LOW async 1 chain chain _chain actual 0 expected 2 tasks ['_chain', 'group', 'group', '_chain']
LOW async 1 chord leaf public actual 1 expected 0 tasks ['_chord']
LOW async 1 chord chain public actual 1 expected 0 tasks ['_chord']
LOW async 1 chord chain _chain actual 0 expected 2 tasks ['_chord', 'group', '
…[528 chars omitted]…

 chord _chain actual 1 expected 3 tasks ['group', 'group', 'group', '_chord']
LOW async 2 chain chain _chain actual 0 expected 2 tasks ['_chain', 'group', 'group', '_chain']
LOW async 2 chord leaf public actual 1 expected 0 tasks ['_chord']
LOW async 2 chord chain public actual 1 expected 0 tasks ['_chord']
LOW async 2 chord chain _chain actual 0 expected 2 tasks ['_chord', 'group', 'group', '_chain']

1 _chain ['Signature', 'group', 'group', '_chain']
 AsyncResult 7f10d06d-0b46-4add-9b22-34ea899229cc
 parent
   AsyncResult ad7b1ada-8a25-4b23-856d-775a959c065d
as_tuple
(('7f10d06d-0b46-4add-9b22-34ea899229cc', (('ad7b1ada-8a25-4b23-856d-775a959c065d', None), None)), None)
restored
 AsyncResult 7f10d06d-0b46-4add-9b22-34ea899229cc
 parent
   AsyncResult ad7b1ada-8a25-4b23-856d-775a959c065d

 1 public ['Signature', 'group', '_chord']

 AsyncResult 7eef827d-f627-4a68-8e67-76e3976f1e07
 parent
   AsyncResult 0b690613-0adc-4b14-86c8-d02ea984504d
   parent
     GroupResult beef36e2-f80c-41cb-88b7-108d2b1b8cfc
     children 2
       AsyncResult 970589e4-ce0f-4501-bbe0-7169af69717a
       AsyncResult 5faa82cb-ba49-4a9c-92b3-a2849ef8ade8
     parent
       GroupResult d17fafd5-5de8-4bf3-9d8f-269512a14c9c
       children 2
         AsyncResult ed2e3b36-a6e1-4834-afdc-4c6f2cc5720f
         As
…[2860 chars omitted]…
1f
     children 2
       AsyncResult bf093d7e-b8d0-4133-a478-83b4513481ce
       AsyncResult f3c6bfc2-569f-4262-aa0d-9218695b40bb
     parent
       GroupResult 6f4c4f77-7876-4b22-a5f1-3f776f693494
       children 2
         AsyncResult 7ff0dfac-6e2f-430f-85f0-9856db8f4193
         AsyncResult f3688534-61ed-4a61-a297-be92d86d3c18
       parent
         AsyncResult b5ea97c5-57f1-421d-80e7-b68137c5106b

tasks:
0 Signature x.ident(1) id None
1 group group((x.ident(21), ident(22))) id None
2 group group((x.ident(31), ident(32))) id None
3 _chain x.ident(8) | ident(9) id None
inner tasks [('Signature', x.ident(8)), ('Signature', x.ident(9))]

prepared tasks (pop order):
Signature 210a1536-3d33-4c23-9cfc-209c87b9d296 x.ident(9)
_chord 52e4e2ee-0108-42e3-899c-59a3b3b2f063 %celery.chord(group([x.ident(21), ident(22)]), kwargs={}, header=group([x.ident(31), ident(32)]), body=x.ident(8))
Signature 34322c1a-76e0-49e8-8569-3e7a048b4957 x.ident(1)
results:

AsyncResult 210a1536-3d33-4c23-9cfc-209c87b9d296 parent AsyncResult 42f08550-657a-4636-96ad-64c6a2283cd6
AsyncResult 42f08550-657a-4636-96ad-64c6a2283cd6 parent GroupResult a4f31cfd-4ae4-4944-9f5f-0b71689e0e0a
AsyncResult 34322c1a-76e0-49e8-8569-3e7a048b4957 parent None None
c.id 210a1536-3d33-4c23-9cfc-209c87b9d296
returned 210a1536-3d33-4c23-9cfc-209c87b9d296
 AsyncResult 210a1536-3d33-4c23-9cfc-209c87b9d296
 parent
   AsyncResult 42f08550-657a-4636-96ad-64c6a2283cd6

tasks:
0 Signature x.ident(1) id None
1 group group((x.ident(21), ident(22))) id None
2 group group((x.ident(31), ident(32))) id None
3 _chain x.ident(8) | ident(9) id None
inner tasks [('Signature', x.ident(8)), ('Signature', x.ident(9))]

prepared tasks (pop order):
Signature 592023d2-99e1-4253-bdce-b1c892c5fde4 x.ident(9)
_chord f97c6b16-a328-4a0f-8f03-4319a6ff5a0f %celery.chord(group([x.ident(21), ident(22)]), kwargs={}, header=group([x.ident(31), ident(32)]), body=x.ident(8))
Signature a243a67b-922f-49d3-9119-15258e4d321d x.ident(1)
results:

AsyncResult 592023d2-99e1-4253-bdce-b1c892c5fde4 parent AsyncResult da9323da-96f6-4d55-a78e-e64845d89630
AsyncResult da9323da-96f6-4d55-a78e-e64845d89630 parent GroupResult e294dabf-68c1-440c-86e0-d1a4bf43f700
AsyncResult a243a67b-922f-49d3-9119-15258e4d321d parent None None
r 592023d2-99e1-4253-bdce-b1c892c5fde4 parent da9323da-96f6-4d55-a78e-e64845d89630 True
p da9323da-96f6-4d55-a78e-e64845d89630 parent None None

tasks:
0 Signature x.ident(1) id None
1 group group((x.ident(21), ident(22))) id None
2 group group((x.ident(31), ident(32))) id None
3 _chain x.ident(8) | ident(9) id None
inner tasks [('Signature', x.ident(8)), ('Signature', x.ident(9))]

prepared tasks (pop order):
Signature 30f9f6da-29d4-4fb9-bd0a-73916b7944ec x.ident(9)
_chord 6a0fedc4-d1ec-4ba7-b8b2-6ee80bcf0fcd %celery.chord(group([x.ident(21), ident(22)]), kwargs={}, header=group([x.ident(31), ident(32)]), body=x.ident(8))
Signature 2814a2ac-7101-4ddc-a02d-76e9ccc5b223 x.ident(1)
results:

AsyncResult 30f9f6da-29d4-4fb9-bd0a-73916b7944ec parent AsyncResult 80a0da82-5f23-43f5-a423-dcad101b4ebb
AsyncResult 80a0da82-5f23-43f5-a423-dcad101b4ebb parent GroupResult 695b83cc-b299-4e5e-b017-0398ebe43d3b
AsyncResult 2814a2ac-7101-4ddc-a02d-76e9ccc5b223 parent None None
r0 30f9f6da-29d4-4fb9-bd0a-73916b7944ec r1 80a0da82-5f23-43f5-a423-dcad101b4ebb r2 2814a2ac-7101-4ddc-a02d-76e9ccc5b223
r0.parent is r1? False
r1.parent id before 695b83cc-b299-4e5e-b017-0398ebe43d3b type <class 'celery.result.GroupResult'>
r0.parent.parent None is r1.parent? False
r1.parent after 695b83cc-b299-4e5e-b017-0398ebe43d3b type <class 'celery.result.GroupResult'>

1 public ['Signature', 'group', '_chord']
 AsyncResult 30f10fa0-5be7-4444-b56d-d226e0514bb5
 parent
   AsyncResult fe957012-1e04-4bd9-98a0-8c6342e929ff
   parent
     GroupResult f60416af-333c-4d01-abfa-f3b6256d2efe
     children 2
       AsyncResult 42061d28-20d7-4d37-92f7-e2d9a0af7c33
       AsyncResult 41dccb18-40b6-43cd-8196-6ed223366b6e
     parent
       GroupResult f07200dc-5a90-4b6e-89ef-b80817e79f98
       children 2

         AsyncResult a6071b87-c70e-48f8-983c-309981a56066
         AsyncResult 34fbeac3-0e91-44e3-82fb-9c3ea43fad5c
       parent
         AsyncResult 13202f1a-32a2-49c0-bef6-b62e4b215c13
restored
 AsyncResult 30f10fa0-5be7-4444-b56d-d226e0514bb5
 parent
   AsyncResult fe957012-1e04-4bd9-98a0-8c6342e929ff
   parent
     GroupResult f60416af-333c-4d01-abfa-f3b6256d2efe
     children 2
       AsyncResult 42061d28-20d7-4d37-92f7-e2d9a0af7c33
       AsyncResult 41d
…[1704 chars omitted]…
c2765f-677a-4581-a0a3-8ef741059752
       parent
         AsyncResult 95e17f30-fffa-4abe-9dfc-2a9995e74bb7

 2 _chain ['Signature', 'group', 'group', 'Signature', 'Signature']
 AsyncResult 8859a542-4b92-4ac1-8230-db39953c2cfd
 parent
   AsyncResult eccd3729-31f4-4f56-8bb7-a236ac809c7b
restored
 AsyncResult 8859a542-4b92-4ac1-8230-db39953c2cfd
 parent
   AsyncResult eccd3729-31f4-4f56-8bb7-a236ac809c7b

1 ['Signature', 'group', 'group', 'Signature', 'Signature']
 AsyncResult 03e9dc0f-25aa-428d-bd42-d525184ecb85
 parent
   AsyncResult daa67bac-bfab-4852-b0fd-647199f3dba7
restored
 AsyncResult 03e9dc0f-25aa-428d-bd42-d525184ecb85
 parent
   AsyncResult daa67bac-bfab-4852-b0fd-647199f3dba7

 2 ['Signature', 'group', 'group', 'Signature', 'Signature']
 AsyncResult 53963b57-5783-4c14-9ced-a64a4e5b14be
 parent

   AsyncResult 36af8724-2d23-43cf-9137-e9e0050cfafd
restored
 AsyncResult 53963b57-5783-4c14-9ced-a64a4e5b14be
 parent
   AsyncResult 36af8724-2d23-43cf-9137-e9e0050cfafd

1 _chain ['Signature', 'group', 'group', '_chain']
 AsyncResult 64d642f8-bb6a-48ba-9bf4-d76b84a84d83
 parent
   AsyncResult efb53958-9646-4a15-9622-a88c58350c2d
as_tuple
(('64d642f8-bb6a-48ba-9bf4-d76b84a84d83', (('efb53958-9646-4a15-9622-a88c58350c2d', None), None)), None)
restored
 AsyncResult 64d642f8-bb6a-48ba-9bf4-d76b84a84d83
 parent
   AsyncResult efb53958-9646-4a15-9622-a88c58350c2d

 1 public ['Signature', 'group', '_chord']

 AsyncResult efb94134-5854-454b-81b0-32178ec7a01b
 parent
   AsyncResult a3ce84c6-3b86-4a82-85c8-8b4003d84449
   parent
     GroupResult f32f3a97-54ee-4932-9d6a-facee4ee6aba
     children 2
       AsyncResult 0baa429a-6e57-4430-b88e-c3d58b3ce7be
       AsyncResult 97a68a72-78ff-4a07-83df-334d430ebd78
     parent
       GroupResult cef7d815-4ea9-468a-8257-fddfcc38076a
       children 2
         AsyncResult 39339376-9ae6-4b6c-80a7-1c2ee2fb335b
         As
…[2860 chars omitted]…
eb
     children 2
       AsyncResult 6936b3f5-986c-4385-8a6f-1781d002802b
       AsyncResult cbe4ad74-f700-4618-a7f7-b28ce6e968d7
     parent
       GroupResult ee87e92f-8820-4ed5-9665-c601e7ee2460
       children 2
         AsyncResult a2ecdd06-0b06-4b28-91c8-fd0dc088fbbf
         AsyncResult ca7b3c2a-fd97-4e75-b50f-e52f2cdb1310
       parent
         AsyncResult 2a8e1cd9-49b4-4ebc-8fed-6500c206da95

1 _chain ['Signature', 'group', 'group', '_chain']
 AsyncResult bd214ee7-ce09-466b-997b-5cc3a0c6ff32
 parent
   AsyncResult b2344959-5120-4170-8723-b1a471162868
   parent
     AsyncResult b2344959-5120-4170-8723-b1a471162868
     parent
       GroupResult 166513a6-a79c-4e9d-b120-9cf1292a3490
       children 2
         AsyncResult 34242bf1-3680-4ba9-991b-f6bf432ac3c3
         AsyncResult c680d668-904e-45c2-8389-e398de348b44
       parent

         GroupResult 866d0529-e449-4f7f-b623-51dd9c50dd90
         children 2
           AsyncResult 506047d6-251b-44c6-a31d-325132889471
           AsyncResult 443e79bc-6948-4d14-b1a6-0dd0ee75fc5a
         parent
           AsyncResult 4b282324-0a2b-4044-9766-5abf2a7365a0
as_tuple
(('bd214ee7-ce09-466b-997b-5cc3a0c6ff32', (('b2344959-5120-4170-8723-b1a471162868', (('b2344959-5120-4170-8723-b1a471162868', (('166513a6-a79c-4e9d-b120-9cf1292a3490', (('
…[5908 chars omitted]…
87
     children 2
       AsyncResult 2e244223-351f-4e90-b216-440df2acf869
       AsyncResult bbfbdd27-aa42-4f58-8ff5-61c2d49f769a
     parent
       GroupResult f5f691e3-2c98-4dad-85fc-d480f59348fc
       children 2
         AsyncResult 3aeabcdc-36b0-443c-9def-c0af83b7f236
         AsyncResult 59181935-c040-44ed-a4ef-e163d9457f1f
       parent
         AsyncResult fc689c23-f703-4ef6-95dc-a1121a759baf

CHORD FREEZE 1ea8737e-81fa-429a-95e8-4f2f6f8245c1 returned AsyncResult b50aea64-0210-4b6f-8859-563c2068661a parent GroupResult 1ea8737e-81fa-429a-95e8-4f2f6f8245c1
CHORD FREEZE 1ea8737e-81fa-429a-95e8-4f2f6f8245c1 returned AsyncResult b50aea64-0210-4b6f-8859-563c2068661a parent GroupResult 1ea8737e-81fa-429a-95e8-4f2f6f8245c1
CHORD FREEZE 9bc1b341-cff4-41f8-81b3-8aaa43f18137 returned AsyncResult b50aea64-0210-4b6f-8859-563c2068661a parent GroupResult 1ea8737e-81fa-429a-95e8-4f2f6f8245c1
results list:
AsyncResult adf2aae0-2071-4be4-855a-d93c407de315 parent AsyncResult b50aea64-0210-4b6f-8859-563c2068661a parent is prior? False
AsyncResult b50aea64-0210-4b6f-8859-563c2068661a parent GroupResult 1ea8737e-81fa-429a-95e8-4f2f6f8245c1 parent is prior? False
AsyncResult 2d65e43f-0258-4aaf-a7c3-83068d72792e parent None None parent is prior? False
chain:
AsyncResult adf2aae0-2071-4be4-855a-d93c407de315
AsyncResult b50aea64-0210-4b6f-8859-563c2068661a
AsyncResult b50aea64-0210-4b6f-8859-563c2068661a
GroupResult 1ea8737e-81fa-429a-95e8-4f2f6f8245c1

GroupResult 9bc1b341-cff4-41f8-81b3-8aaa43f18137
AsyncResult 2d65e43f-0258-4aaf-a7c3-83068d72792e

1 _chain ['Signature', 'group', 'group', '_chain']
 AsyncResult 33a7995d-96b9-4b04-8d83-ddd9486435b8
 parent
   AsyncResult 241000a2-c2f1-43e4-a42d-be37ac3ea562
   parent
     GroupResult 97f9b539-73cf-49e0-a410-8f3278c33cfd
     children 2
       AsyncResult 74b64cc9-04de-4027-a8eb-3c3561fc02af
       AsyncResult 659e48cc-ced8-406c-8170-2e58b0ea007b
     parent
       GroupResult b7f9a4fe-9826-4e7d-8cd2-7f3fe45b2668
       children 2

         AsyncResult 112b8f3f-89fc-4314-bead-a096cf9e350b
         AsyncResult c602e55b-be4d-4c22-a8d2-06bfcb5f786b
       parent
         AsyncResult c5a21237-35f9-4d11-99f3-db23effe68a7
as_tuple
(('33a7995d-96b9-4b04-8d83-ddd9486435b8', (('241000a2-c2f1-43e4-a42d-be37ac3ea562', (('97f9b539-73cf-49e0-a410-8f3278c33cfd', (('b7f9a4fe-9826-4e7d-8cd2-7f3fe45b2668', (('c5a21237-35f9-4d11-99f3-db23effe68a7', None), None)), [(('112b8f3f-89fc-4314-bead-a096cf
…[4319 chars omitted]…
1cd50fe6ac
   parent
     GroupResult 6888b795-6103-40fb-913a-c44d058fcb94
     children 2
       AsyncResult b53b972c-2d97-4e48-9edf-5957f352911c
       AsyncResult 884ed42b-930a-4dee-b3aa-0f9bbacd4628
     parent
       GroupResult cf7500c7-d3a4-47f2-a1d0-561fc1b5339e
       children 2
         AsyncResult 86910774-79b5-4955-9a67-e4fbcc3b0d75
         AsyncResult 5f05ff5b-5a64-4af4-a084-146e62afd060

LOW async 1 group leaf public actual 1 expected 0 tasks ['_chord']
LOW async 1 group leaf _chain actual 1 expected 3 tasks ['group', 'group', 'group', 'Signature']
LOW async 1 group chain public actual 1 expected 0 tasks ['_chord']
LOW async 1 group chord public actual 1 expected 0 tasks ['_chord']
LOW async 1 group chord _chain actual 1 expected 3 tasks ['group', 'group', 'group', '_chord']
LOW async 1 chord leaf public actual 1 expected 0 tasks ['_chord']
LOW async 1 chord chain public actual 1 expected 0 tasks ['_chord']
LOW async 2 group leaf public actual 1 expected 0 tasks ['_chord']
LOW async 2 group leaf _chain actual 1 expected 3 tasks ['group', 'group', 'group', 'Signature']
LOW async 2 group chain public actual 1 expected 0 tasks ['_chord']
LOW async 2 group chord public actual 1 expected 0 tasks ['_chord']
LOW async 2 group chord _chain actual 1 expected 3 tasks ['group', 'group', 'group', '_chord']

LOW async 2 chord leaf public actual 1 expected 0 tasks ['_chord']
LOW async 2 chord chain public actual 1 expected 0 tasks ['_chord']

1 ['group', 'group', 'group', 'Signature']
 AsyncResult 6a25a57c-dc8b-419e-95a4-69edc5765665
 parent
   GroupResult f28e96e3-3d3e-44ec-a81c-cc69193cf3e1
   children 2
     AsyncResult 6a05c355-3d41-4513-8af3-e21f9536afa9
     AsyncResult 91904c42-b04c-40c6-8c98-dba02884285d
restored
 AsyncResult 6a25a57c-dc8b-419e-95a4-69edc5765665
 parent
   GroupResult f28e96e3-3d3e-44ec-a81c-cc69193cf3e1
   children 2

     AsyncResult 6a05c355-3d41-4513-8af3-e21f9536afa9
     AsyncResult 91904c42-b04c-40c6-8c98-dba02884285d

 2 ['group', 'group', 'group', 'Signature']
 AsyncResult fe181fd2-5e00-4e6e-936b-5f6b2f8f3aa8
 parent
   GroupResult 4e6653ab-13e2-41e3-bed3-c08943f30438
   children 2
     AsyncResult 5bf2c1af-91ca-424a-a001-40abccde5a75
     AsyncResult c9fca2ce-8e39-41a8-a53b-227fe8f515aa
restored
 AsyncResult fe181fd2-5e00-4e6e-936b-5f6b2f8f3aa8
 parent
   GroupResult 4e6653ab-13e2-41e3-bed3-c08943f30438
   children 2
     AsyncResult 5bf2c1af-91ca-424a-a001-40abccde5a75
     AsyncResult c9fca2ce-8e39-41a8-a53b-227fe8f515aa

freeze chord 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89 header tasks 2 returned 028049d8-f86d-4e27-92fa-85f13edc7692 parent GroupResult 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89
freeze chord 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89 header tasks 2 returned 028049d8-f86d-4e27-92fa-85f13edc7692 parent GroupResult 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89
freeze chord f0800b5c-dcf8-4f5e-bbd5-f98a13f6eb32 header tasks 2 returned 028049d8-f86d-4e27-92fa-85f13edc7692 parent GroupResult 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89
freeze chord 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89 header tasks 2 returned 028049d8-f86d-4e27-92fa-85f13edc7692 parent GroupResult 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89
freeze chord f0800b5c-dcf8-4f5e-bbd5-f98a13f6eb32 header tasks 2 returned 028049d8-f86d-4e27-92fa-85f13edc7692 parent GroupResult 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89
freeze chord a4dd10a0-3dc9-4919-9eae-65f4b3ba7ab2 header tasks
…[83 chars omitted]…
bd8-8b11-3ebf70f9ef89
results
AsyncResult 028049d8-f86d-4e27-92fa-85f13edc7692 parent GroupResult 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89
tasks
_chord a4dd10a0-3dc9-4919-9eae-65f4b3ba7ab2

chain AsyncResult 028049d8-f86d-4e27-92fa-85f13edc7692
chain GroupResult 7e9814bb-0295-4bd8-8b11-3ebf70f9ef89
chain GroupResult f0800b5c-dcf8-4f5e-bbd5-f98a13f6eb32
chain GroupResult a4dd10a0-3dc9-4919-9eae-65f4b3ba7ab2

0 AsyncResult f66b64d5-c0af-4e0d-ac34-85e352db0581 parent 6496e602-dd5d-4f97-b974-aeae5adf904c bool parent True
1 GroupResult 6496e602-dd5d-4f97-b974-aeae5adf904c parent None bool parent None

1
 AsyncResult e62cfa5e-d84f-44c5-9d01-27dbf5b3428e
 parent
   GroupResult 5b66241c-f254-4316-ba9c-b9d05c0fcbf4
   children 2
     AsyncResult 3d7fa889-bf1d-4c97-839a-f3fd95708279
     AsyncResult 03dca551-0939-4d99-8017-62acbf652d46
   parent
     GroupResult f0316686-493b-4193-a5a5-cb17c5ee4c77
     children 2
       AsyncResult 308f233b-dc85-401d-8138-2401e917c17a
       AsyncResult 1ea0af2a-dc86-4f54-901f-8ffd5b88c2ea

     parent
       GroupResult 59ea764b-0556-459c-8e8f-785a8025da3b
       children 2
         AsyncResult be18b7f1-9e60-485d-a169-e93563387781
         AsyncResult 3330b668-aab8-4fbe-973b-d15a62678c6f
restored
 AsyncResult e62cfa5e-d84f-44c5-9d01-27dbf5b3428e
 parent
   GroupResult 5b66241c-f254-4316-ba9c-b9d05c0fcbf4
   children 2
     AsyncResult 3d7fa889-bf1d-4c97-839a-f3fd95708279
     AsyncResult 03dca551-0939-4d99-8017-62acbf652d46
   parent
     GroupResult
…[1228 chars omitted]…
1db1600b99
   parent
     GroupResult 7149ecf4-bb9e-42dc-9596-28047ee5c9e3
     children 2
       AsyncResult 032b45f7-aebf-43f9-90fd-49197254ccaa
       AsyncResult 8f632bc9-a784-48ff-8d19-cb29819b14b3
     parent
       GroupResult 3e7dfed6-dd04-4250-9fe7-ed2e49b80a95
       children 2
         AsyncResult 6ba5c37f-a4d8-4469-a3ef-2e3072abc25a
         AsyncResult e45c27df-ed45-449c-9833-9a50ab6c3bdc

............u........................................................... [ 21%]
....................................................x..x................ [ 42%]
............................uu.uu.......uuu.uuu.uuu.uuu.uuu.uuu.uuu.uuu. [ 56%]
........................................s............................... [ 77%]
........................................................................ [ 99%]
..                                                                       [100%]
=============================== warnings summary ===============================
t/unit/tasks/test_canvas.py::test_chain::test_chain_errbacks_propagated_to_chord_body
  /repo/celery/canvas.py:1288: CPendingDeprecationWarning: task_allow_error_cb_on_chord_header=False is pending deprecation in a future release of Celery.
  Please test the new behavior by setting task_allow_error_cb_on_chord_header to True and repor
…[936 chars omitted]…
new behavior by setting task_allow_error_cb_on_chord_header to True and report any concerns you might have in our issue tracker before we make a final decision regarding how errbacks should behave when used with chords.

    chord_sig.link_error(errback_sig)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
330 passed, 1 skipped, 2 xfailed, 3 warnings, 29 subtests passed in 7.65s

fail 0

diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..677f5d8 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -23,6 +23,7 @@ from kombu.utils.objects import cached_property
 from kombu.utils.uuid import uuid
 from vine import barrier
 
+from celery import states
 from celery._state import current_app
 from celery.exceptions import CPendingDeprecationWarning
 from celery.result import GroupResult, allow_join_result

@@ -1213,10 +1214,12 @@ class _chain(Signature):
                 continue
 
             # TODO why isn't this asserting is_last_task == False?
+            body_first_res = None
             if isinstance(task, group) and prev_task:
                 # automatically upgrade group(...) | s to chord(group, s)
                 # for chords we freeze by pretending it's a normal
                 # signature instead of a group.
+                body_first_res = 
…[2745 chars omitted]…
_result
+        node = bodyres
+        seen = set()
+        while node:
+            if node.id in seen:
+                raise RuntimeError('Recursive result parents')
+            seen.add(node.id)
+            if node.parent is None:
+                node.parent = header_result
+                break
+            node = node.parent
         return bodyres
 
     def clone(self, *args, **kwargs):

protocol 1
EagerResult 41d419ac-850a-4e91-967f-1505ee7f5257 children -
GroupResult 9a3d6a51-96a6-4c6f-bac2-653f6576d5c4 children 2
GroupResult d3c01be5-58da-4de9-a6ac-367968caa3de children 2
EagerResult 7b4f2471-9116-4778-ac58-fae65620ff30 children -
restored
AsyncResult 41d419ac-850a-4e91-967f-1505ee7f5257 children -
GroupResult 9a3d6a51-96a6-4c6f-bac2-653f6576d5c4 children 2
GroupResult d3c01be5-58da-4de9-a6ac-367968caa3de children 2
AsyncResult 7b4f2471-9116-4778-ac58-fae65620ff30 children -
protocol 2
EagerResult 069dc8b9-49b1-46ee-ad0b-6b0af57635be children -

GroupResult 015e0d84-24f1-465b-8e34-18c05983ece0 children 2
GroupResult 5fd8659c-c012-4885-8634-5a270875a513 children 2
EagerResult b0f20ffc-6aa2-464b-a84e-158f9d203a8c children -
restored
AsyncResult 069dc8b9-49b1-46ee-ad0b-6b0af57635be children -
GroupResult 015e0d84-24f1-465b-8e34-18c05983ece0 children 2
GroupResult 5fd8659c-c012-4885-8634-5a270875a513 children 2
AsyncResult b0f20ffc-6aa2-464b-a84e-158f9d203a8c children -

if prerun_receivers:
                    send_prerun(sender=task, task_id=uuid, task=task,
                                args=args, kwargs=kwargs)
                loader_task_init(uuid, task)
                if track_started:
                    task.backend.store_result(
                        uuid, {'pid': pid, 'hostname': hostname}, STARTED,
                        request=task_request,
                    )

                # -*- TRACE -*-
                try:

                    if task_before_start:
                        task_before_start(uuid, args, kwargs)

                    R = retval = fun(*args, **kwargs)
                    state = SUCCESS
                except Reject as exc:
                    I, R = Info(REJECTED, exc), ExceptionInfo(internal=True)
                    state, retval = I.state, I.retval
                    I.handle_reject(task, tas
…[2471 chars omitted]…
_ok_t(report_internal_error(task, exc), TraceInfo(FAILURE, exc), 0.0, None)


def _signal_internal_error(task, uuid, args, kwargs, request, exc):
    """Send a special `internal_error` signal to the app for outside body errors."""
    tb = None
    einfo = None
    try:
        _, _, tb = sys.exc_info()
        einfo = ExceptionInfo()
        einfo.exception = get_pickleable_exception(einfo.exception)

ImportError while importing test module '/repo/t/unit/backends/test_mongodb.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_mongodb.py:5: in <module>
    import dns.version
E   ModuleNotFoundError: No module named 'dns'
________________ ERROR collecting t/unit/backends/test_redis.py ________________
t/unit/backends/test_redis.py:569: in <module>
    class MyCredentialProvider(CredentialProvider):

E   TypeError: NoneType takes no arguments
_________________ ERROR collecting t/unit/backends/test_rpc.py _________________
'celery' not found in `markers` configuration option
_________________ ERROR collecting t/unit/backends/test_s3.py ____
…[5469 chars omitted]…
ds/test_rpc.py - Failed: 'celery' not found in `markers` c...
ERROR t/unit/backends/test_s3.py
ERROR t/unit/security/test_certificate.py
ERROR t/unit/security/test_key.py
ERROR t/unit/security/test_security.py
ERROR t/unit/security/test_serialization.py
ERROR t/unit/utils/test_annotations.py
!!!!!!!!!!!!!!!!!!! Interrupted: 14 errors during collection !!!!!!!!!!!!!!!!!!!
9 skipped, 14 errors in 10.17s

1 [('AsyncResult', 'ea7c96f5-e8a3-4320-b841-5849ca52b0a6', None), ('AsyncResult', 'a1112ff1-b347-4de2-96c0-06b3d3ebe5cd', None), ('GroupResult', '88ba29f7-1847-4902-8214-252e8f057421', 2), ('GroupResult', '4c3bf743-be62-477c-9709-dc453c96ec66', 2), ('AsyncResult', 'c987e69d-d4c5-4dc3-b48c-03e03f264c3d', None)]
 restored [('AsyncResult', 'ea7c96f5-e8a3-4320-b841-5849ca52b0a6', None), ('AsyncResult', 'a1112ff1-b347-4de2-96c0-06b3d3ebe5cd', None), ('GroupResult', '88ba29f7-1847-4902-8214-252e8f057421', 2), ('GroupResult', '4c3bf743-be62-477c-9709-dc453c96ec66', 2), ('AsyncResult', 'c987e69d-d4c5-4dc3-b48c-03e03f264c3d', None)]
2 [('AsyncResult', 'd7fa6091-b8aa-4c64-b780-44c8e95766ae', None), ('AsyncResult', 'f1e863ab-6e68-43d2-bb5a-33593db5b291', None), ('GroupResult', 'bb349fc9-1023-4eb7-b5b9-1c40a8ed969f', 2), ('GroupResult', '493fa8d6-3f3c-4c84-ac8b-89846f5a4f80', 2), ('AsyncResult', '33b081d5-fa00-43d9-94e4-05515296a535', None)]
 restored [('AsyncResult', 'd7fa6091-b8aa-4c64-b780-44c8e95766ae', None), ('AsyncResult', 'f1e863ab-6e68-43d2-bb5a-33593db5b291', None), ('GroupResult', 'bb349fc9-1023-4eb7-b5b9-1c40a8ed969f', 2), ('GroupResult', '493fa8d6-3f3c-4c84-ac8b-89846f5a4f80', 2), ('AsyncResult', '33b081d5-fa00-43d9-94e4-05515296a535', None)]
ok

t/unit/tasks/test_canvas.py::test_chord::test_flag_allow_error_cb_on_chord_header
  /repo/t/unit/tasks/test_canvas.py:1928: CPendingDeprecationWarning: task_allow_error_cb_on_chord_header=False is pending deprecation in a future release of Celery.
  Please test the new behavior by setting task_allow_error_cb_on_chord_header to True and report any concerns you might have in our issue tracker before we make a final decision regarding how errbacks should behave when used with chords.
    chord_sig.link_error(errback_sig)

t/unit/tasks/test_stamping.py::test_stamping_mechanism::test_callback_stamping
t/unit/tasks/test_stamping.py::test_stamping_mechanism::test_callback_stamping_link_after_stamp
t/unit/tasks/test_stamping.py::test_stamping_mechanism::test_callback_stamping_link_multiple_visitors
  /repo/celery/canvas.py:402: AlwaysEagerIgnored: task_always_eager has no effect on send_task
…[3212 chars omitted]…
     return self._execute_mock_call(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

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

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
988 passed, 1 skipped, 2 xfailed, 9 warnings, 28765 subtests passed in 35.98s

182:def regen(it):
194:class _regen(UserList, list):

    def __iter__(self):
        yield from self.__consumed
        yield from self.__lookahead_consume()

    def __getitem__(self, index):
        if index < 0:
            return self.data[index]
        # Consume elements up to the desired index prior to attempting to
        # access it from within `__consumed`

        consume_count = index - len(self.__consumed) + 1
        for _ in self.__lookahead_consume(limit=consume_count):
            pass
        return self.__consumed[index]

    def __bool__(self):
        if len(self.__consumed):
            return True

        try:
            next(iter(self))
        except StopIteration:
            return False
        else:
            return True

    @property
    def data(self):
        if not self.__done:
            self.__consumed.extend(self.__it)
            self.__done
…[1368 chars omitted]…
yargs else None,
        ', '.join(f'{k}="{v}"' for k, v in kwonlyargs_optional),
        f'**{varkw}' if varkw else None,
    ]))


if sys.version_info >= (3, 14):
    import annotationlib as _annotationlib

    def _getfullargspec(fun):
        # In Python 3.14+, inspect.getfullargspec evaluates annotations by default
        # (PEP 649), raising NameError for TYPE_CHECKING-only types. We don't need

/usr/local/bin/python: No module named flake8
 celery/canvas.py | 52 ++++++++++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 48 insertions(+), 4 deletions(-)
 M celery/canvas.py

eager False p 1 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
eager False p 1 tasks ['_chord', 'group'] groups 3 [2, 2, 2] restored 3 [2, 2, 2]
eager False p 1 tasks ['Signature', 'group', 'group'] groups 2 [2, 2] restored 2 [2, 2]
eager False p 1 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
eager False p 2 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
eager False p 2 tasks ['_chord', 'group'] groups 3 [2, 2, 2] restored 3 [2, 2, 2]
eager False p 2 tasks ['Signature', 'group', 'group'] groups 2 [2, 2] restored 2 [2, 2]
eager False p 2 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
eager True p 1 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
eager True p 1 tasks ['_chord', 'group'] groups 3 [2, 2, 2] restored 3 [2, 2, 2]
eager True p 1 tasks ['Signature', 'group', 'group'] groups 2 
…[37 chars omitted]…

 1 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
eager True p 2 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
eager True p 2 tasks ['_chord', 'group'] groups 3 [2, 2, 2] restored 3 [2, 2, 2]
eager True p 2 tasks ['Signature', 'group', 'group'] groups 2 [2, 2] restored 2 [2, 2]
eager True p 2 tasks ['Signature', 'Signature'] groups 2 [2, 2] restored 2 [2, 2]
ok

Traceback (most recent call last):
  File "<stdin>", line 18, in <module>
  File "/repo/celery/canvas.py", line 1319, in apply
    last and (last.get(),), **dict(self.options, **options))
              ^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: ign() takes 0 positional arguments but 1 was given

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

26:from celery import states
1329:            if getattr(res, 'state', None) in (states.IGNORED, states.REJECTED):

1 [('EagerResult', None), ('GroupResult', 2), ('GroupResult', 2), ('EagerResult', None)] [('AsyncResult', None), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', None)]
Traceback (most recent call last):
  File "<stdin>", line 19, in <module>
AssertionError

docs/history/changelog-3.1.rst:99:    result list are not properly tupled with the as_tuple() method when it is
docs/history/changelog-3.1.rst:100:    a parent result. This is due to the as_tuple() method performing a logical
docs/history/changelog-4.2.rst:285:- **Group Result**: Modify GroupResult.as_tuple() to include parent (fixes #4106) (#4205)
docs/history/changelog-4.2.rst:397:- Matched the behavior of `GroupResult.as_tuple()` to that of `AsyncResult.as_tuple()`.
docs/history/whatsnew-4.0.rst:2144:        >>> tup = result.as_tuple()

/repo
total 332
drwxr-xr-x 1 root root   4096 Nov  2 23:27 .
drwxr-xr-x 1 root root   4096 Nov  2 23:25 ..
-rw-r--r-- 1 root root    343 Nov  2 23:24 .bumpversion.cfg
-rw-r--r-- 1 root root    306 Nov  2 23:24 .dockerignore
-rw-r--r-- 1 root root    227 Nov  2 23:24 .editorconfig
drwxr-xr-x 1 root root   4096 Nov  2 23:25 .git
drwxr-xr-x 1 root root   4096 Nov  2 23:24 .github
-rw-r--r-- 1 root root    437 Nov  2 23:24 .gitignore
-rw-r--r-- 1 root root   1073 Nov  2 23:24 .pre-commit-config.yaml
-rw-r--r-- 1 root root    616 Nov  2 23:24 .readthedocs.yaml

-rw-r--r-- 1 root root  44762 Nov  2 23:24 CONTRIBUTING.rst
-rw-r--r-- 1 root root   8619 Nov  2 23:24 CONTRIBUTORS.txt
-rw-r--r-- 1 root root 128887 Nov  2 23:24 Changelog.rst
-rw-r--r-- 1 root root   2717 Nov  2 23:24 LICENSE
-rw-r--r-- 1 root root    709 Nov  2 23:24 MANIFEST.in
-rw-r--r-- 1 root root   5866 Nov  2 23:24 Makefile

…[2354 chars omitted]…
./requirements/README.rst
./requirements/constraints.txt
./requirements/default.txt
./requirements/dev.txt
./requirements/docs.txt
./requirements/pkgutils.txt
./requirements/security.txt
./requirements/test-ci-base.txt
./requirements/test-ci-default.txt
./requirements/test-integration.txt
./requirements/test-pypy3.txt
./requirements/test.txt
./setup.cfg
./setup.py
./t/__init__.py
./t/skip.py
./tox.ini

celery/canvas.py:356:    def apply(self, args=None, kwargs=None, **options):
celery/canvas.py:370:    def apply_async(self, args=None, kwargs=None, route_name=None, **options):
celery/canvas.py:935:class _chain(Signature):
celery/canvas.py:1036:    def apply_async(self, args=None, kwargs=None, **options):
celery/canvas.py:1294:    def apply(self, args=None, kwargs=None, **options):
celery/canvas.py:1322:class chain(_chain):
celery/canvas.py:1398:    def apply_async(self, args=None, kwargs=None, **opts):
celery/canvas.py:1455:    def apply_async(self, args=None, kwargs=None, **opts):
celery/canvas.py:1471:    def apply_chunks(cls, task, it, n, app=None):
celery/canvas.py:1616:    def apply_async(self, args=None, kwargs=None, add_to_parent=True,
celery/canvas.py:1652:    def apply(self, args=None, kwargs=None, **options):
celery/canvas.py:2153:    def apply_async(self, args=None, kwargs

…[542 chars omitted]…
      [r.as_tuple() for r in self.results]
celery/result.py:1113:def result_from_tuple(r, app=None):
celery/result.py:1123:            parent = result_from_tuple(parent, app)
celery/result.py:1127:                id, [result_from_tuple(child, app) for child in nodes],
t/integration/test_canvas.py
t/smoke/tests/test_canvas.py
t/unit/tasks/test_canvas.py
t/unit/tasks/test_result.py
t/unit/test_canvas.py

"""Composing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
"""

import itertools
import operator
import types
import warnings
from abc import ABCMeta, abstractmethod

from collections import deque
from collections.abc import MutableSequence
from copy import deepcopy
from functools import partial as _partial
from functools import reduce
from operator import itemgetter
from types import GeneratorType

from kombu.utils.functional import fxrange, reprcall
from kombu.utils.objects import cached_property
from kombu.utils.uuid import uuid
from vine import barrier

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
from c
…[8630 chars omitted]…
 4)`:

        .. code-block:: pycon

            >>> res = chain(add.s(2, 2), add.s(4))()
            >>> res.get()
            8

        Calling a chain will return the result of the last task in the chain.
        You can get to the other tasks by following the ``result.parent``'s:

        .. code-block:: pycon

            >>> res.parent.get()
            4

        Using a generator expression:

else:
            # Also supports using task class/instance instead of string name.
            try:
                task_name = task.name
            except AttributeError:
                task_name = task
            else:
                self._type = task

            super().__init__(
                task=task_name, args=tuple(args or ()),
                kwargs=kwargs or {},

                options=dict(options or {}, **ex),
                subtask_type=subtask_type,
                immutable=immutable,
            )

    def __call__(self, *partial_args, **partial_kwargs):
        """Call the task directly (in the current process)."""
        args, kwargs, _ = self._merge(partial_args, partial_kwargs, None)
        return self.type(*args, **kwargs)

    def delay(self, *partial_args, **partial_kwargs):
        """Shortcut to :meth:`apply_async` using star arguments."""
 
…[8630 chars omitted]…
f.is_eager = is_eager
        self.sig = sig
        super().__init__(self, exc, when, **kwargs)

    def humanize(self):
        if isinstance(self.when, numbers.Number):
            return f'in {self.when}s'
        return f'at {self.when}'

    def __str__(self):
        if self.message:
            return self.message
        if self.excs:
            return f'Retry {self.humanize()}: {self.excs}'

1010	        signature = super().clone(*args, **kwargs)
  1011	        signature.kwargs['tasks'] = [
  1012	            to_signature(sig, app=self._app, clone=True)
  1013	            for sig in signature.kwargs['tasks']
  1014	        ]
  1015	        return signature
  1016	
  1017	    def unchain_tasks(self):
  1018	        """Return a list of tasks in the chain.
  1019	
  1020	        The tasks list would be cloned from the chain's tasks.
  1021	        All of the chain callbacks would be added to the last task in the (cloned) chain.

  1022	        All of the tasks would be linked to the same error callback
  1023	        as the chain itself, to ensure that the correct error callback is called
  1024	        if any of the (cloned) tasks of the chain fail.
  1025	        """
  1026	        # Clone chain's tasks assigning signatures from link_error
  1027	        # to each task an
…[8630 chars omitted]…
 else (res, None)
  1122	        if parent:
  1123	            parent = result_from_tuple(parent, app)
  1124	
  1125	        if nodes is not None:
  1126	            return app.GroupResult(
  1127	                id, [result_from_tuple(child, app) for child in nodes],
  1128	                parent=parent,
  1129	            )
  1130	
  1131	        return Result(id, parent=parent)
  1132	    return r

celery/app/task.py:13:from celery.exceptions import Ignore, ImproperlyConfigured, MaxRetriesExceededError, Reject, Retry
celery/app/task.py:15:from celery.result import EagerResult, denied_join_result
celery/app/task.py:864:            raise Reject(exc, requeue=False)
celery/app/task.py:882:            celery.result.EagerResult: pre-evaluated result.
celery/app/task.py:951:        return EagerResult(task_id, retval, state, traceback=tb, name=self.name)
celery/app/task.py:1042:            ~@Ignore: This is always raised when called in asynchronous context.
celery/app/task.py:1227:            raise Ignore('Replaced by new task')
celery/app/trace.py:23:from celery.exceptions import BackendGetMetaError, Ignore, InvalidTaskError, Reject, Retry
celery/app/trace.py:73:LOG_IGNORED = """\
celery/app/trace.py:78:LOG_REJECTED = """\
celery/app/trace.py:92:log_policy_reject = log_policy_t(LOG_R
…[8631 chars omitted]…

gs)
  1466	        return group((xstarmap(task, part, app=self._app)
  1467	                      for part in _chunks(iter(it), n)),
  1468	                     app=self._app)
  1469	
  1470	    @classmethod
  1471	    def apply_chunks(cls, task, it, n, app=None):
  1472	        return cls(task, it, n, app=app)()
  1473	
  1474	
  1475	def _maybe_group(tasks, app):
bash: line 16: rg: command not found

860	
   861	        try:
   862	            S.apply_async()
   863	        except Exception as exc:
   864	            raise Reject(exc, requeue=False)
   865	        if throw:
   866	            raise ret
   867	        return ret
   868	
   869	    def apply(self, args=None, kwargs=None,
   870	              link=None, link_error=None,
   871	              task_id=None, retries=None, throw=None,

   872	              logfile=None, loglevel=None, headers=None, **options):
   873	        """Execute this task locally, by blocking until the task returns.
   874	
   875	        Arguments:
   876	            args (Tuple): positional arguments passed on to the task.
   877	            kwargs (Dict): keyword arguments passed on to the task.
   878	            throw (bool): Re-raise task exceptions.
   879	                Defaults to the :setting:`task_eager_propagates` setting.
   880	
  
…[8630 chars omitted]…
 an exception, this will be the exception
   494	            instance.
   495	        """
   496	        return self._get_task_meta()['result']
   497	    info = result
   498	
   499	    @property
   500	    def traceback(self):
   501	        """Get the traceback of a failed task."""
   502	        return self._get_task_meta().get('traceback')
   503	
   504	    @property
   505	    def state(self):

1045	        return self.run(args, kwargs, app=app, **(
  1046	            dict(self.options, **options) if options else self.options))
  1047	
  1048	    def run(self, args=None, kwargs=None, group_id=None, chord=None,
  1049	            task_id=None, link=None, link_error=None, publisher=None,
  1050	            producer=None, root_id=None, parent_id=None, app=None,
  1051	            group_index=None, **options):
  1052	        """Executes the chain.
  1053	
  1054	        Responsible for executing the chain in the correct order.
  1055	        In a case of a chain of a single task, the task is executed directly
  1056	        and the result is returned for that task specifically.

  1057	        """
  1058	        # pylint: disable=redefined-outer-name
  1059	        #   XXX chord is also a class in outer scope.
  1060	        args = args if args else ()
  1061	        kwargs = k
…[8630 chars omitted]…
     (x, c[1]),
            (x, c[2]),
        ]
        x._cache = None
        x.ready = Mock()
        x.ready.return_value = False
        with pytest.raises(IncompleteStream):
            list(x.iterdeps())
        list(x.iterdeps(intermediate=True))

    def test_eq_not_implemented(self):
        assert self.app.AsyncResult('1') != object()

    @pytest.mark.usefixtures('depends_on_current_app')

"""Built-in task states.

.. _states:

States
------

See :ref:`task-states`.

.. _statesets:

Sets

----

.. state:: READY_STATES

READY_STATES
~~~~~~~~~~~~

Set of states meaning the task result is ready (has been executed).

.. state:: UNREADY_STATES

UNREADY_STATES
~~~~~~~~~~~~~~

Set of states meaning the task result is not ready (hasn't been executed).

.. state:: EXCEPTION_STATES

EXCEPTION_STATES
~~~~~~~~~~~~~~~~

Set of states meaning the task returned an exception.

.. state:: PROPAGATE_STATES

PROPAGATE_STATES
~~~~~~~~~~~~~~~~

Set of exception states that should propagate exceptions to the user.

.. state:: ALL_STATES

ALL_STATES
~~~~~~~~~~

Set of all possible states.

Misc
----

"""

__all__ = (
    'PENDING', 'RECEIVED', 'STARTED', 'SUCCESS', 'FAILURE',
    'REVOKED', 'RETRY', 'IGNORED', 'READY_STATES', 'UNREADY_STATES',
    'EXCEPTION_STATES', 'PROPAGATE_STATES', 'prece
…[8630 chars omitted]…
le(x == self.app)
1065:    def test_GroupResult_with_parent(self):
1067:        result = self.app.GroupResult(
1071:        second_result = result_from_tuple(result.as_tuple(), self.app)
1075:    def test_GroupResult_as_tuple(self):
1077:        result = self.app.GroupResult(
1083:        (result_id, parent_tuple), group_results = result.as_tuple()
1085:        assert parent_tuple == parent.as_tuple()

def prepare_steps(self, args, kwargs, tasks,
                      root_id=None, parent_id=None, link_error=None, app=None,
                      last_task_id=None, group_id=None, chord_body=None,
                      clone=True, from_dict=Signature.from_dict,
                      group_index=None):
        """Prepare the chain for execution.

        To execute a chain, we first need to unpack it correctly.
        During the unpacking, we might encounter other chains, groups, or chords
        which we need to unpack as well.

        For example:

        chain(signature1, chain(signature2, signature3)) --> Upgrades to chain(signature1, signature2, signature3)
        chain(group(signature1, signature2), signature3) --> Upgrades to chord([signature1, signature2], signature3)

        The responsibility of this method is to ensure that the chain is
        correctly unpacked, a
…[8630 chars omitted]…
t_call(self):
        x = Signature('foo', (1, 2), {'arg1': 33}, app=self.app)
        x.type = Mock(name='type')
        x(3, 4, arg2=66)
        x.type.assert_called_with(3, 4, 1, 2, arg1=33, arg2=66)

    def test_link_on_scalar(self):
        x = Signature('TASK', link=Signature('B'))
        assert x.options['link']
        x.link(Signature('C'))
        assert isinstance(x.options['link'], list)

1200	
  1201	            # first task gets partial args from chain
  1202	            if clone:
  1203	                if is_first_task:
  1204	                    task = task.clone(args, kwargs)
  1205	                else:
  1206	                    task = task.clone()
  1207	            elif is_first_task:
  1208	                task.args = tuple(args) + tuple(task.args)
  1209	
  1210	            if isinstance(task, _chain):
  1211	                # splice (unroll) the chain

  1212	                steps_extend(task.tasks)
  1213	                continue
  1214	
  1215	            # TODO why isn't this asserting is_last_task == False?
  1216	            if isinstance(task, group) and prev_task:
  1217	                # automatically upgrade group(...) | s to chord(group, s)
  1218	                # for chords we freeze by pretending it's a normal
  1219	                # signatur
…[8630 chars omitted]…
e(self):
1643:    def test_chord_size_nested_group_chain_group_tail_many(self):
1652:    def test_chord_size_nested_implicit_group_chain_group_tail_single(self):
1659:    def test_chord_size_nested_implicit_group_chain_group_tail_many(self):
1758:    def test_chain_always_eager(self):
1873:    def test_from_dict_deep_deserialize_chain(self, subtests):
1975:    def test_chord_upgrade_on_chaining(self):

t/unit/conftest.py:62:        'broker_url': 'memory://',
t/unit/conftest.py:66:        'result_backend': 'cache+memory://',
t/unit/conftest.py:124:def app(celery_app):
        for task in c.tasks:
            assert task.options['link_error'] == [s('error')]

    def test_apply_options_none(self):
        class static(Signature):

            def clone(self, *args, **kwargs):
                return self

            def _apply_async(self, *args, **kwargs):
                self.args = args
                self.kwargs = kwargs

        c = static(self.add, (2, 2), type=self.add, app=self.app, priority=5)

        c.apply_async(priority=4)
        assert c.kwargs['priority'] == 4

        c.apply_async(priority=None)
        assert c.kwargs['priority'] == 5

    def test_reverse(self):
        x = self.add.s(2, 2) | self.add.s(2)
        assert isinstance(signature(x), _chain)
        as
…[8629 chars omitted]…
roup_id, root_id, app)
        return app.GroupResult(group_id, [
            sig.apply(args=args, kwargs=kwargs, **options) for sig, _, _ in tasks
        ])

    def set_immutable(self, immutable):
        for task in self.tasks:
            task.set_immutable(immutable)

    def stamp(self, visitor=None, append_stamps=False, **headers):
        visitor_headers = None
        if visitor is not None:

raise Reject(exc, requeue=True)
                        return trace_ok_t(R, I, T, Rstr)

            push_task(task)
            root_id = task_request.root_id or uuid
            task_priority = task_request.delivery_info.get('priority') if \
                inherit_parent_priority else None
            push_request(task_request)
            try:
                # -*- PRE -*-
                if prerun_receivers:
                    send_prerun(sender=task, task_id=uuid, task=task,

                                args=args, kwargs=kwargs)
                loader_task_init(uuid, task)
                if track_started:
                    task.backend.store_result(
                        uuid, {'pid': pid, 'hostname': hostname}, STARTED,
                        request=task_request,
                    )

                # -*- TRACE -*-
                try:

…[8629 chars omitted]…
O_PORT') or 27017,
            'database': os.environ.get('MONGO_DB') or 'celery_unittests',
            'taskmeta_collection': (
                os.environ.get('MONGO_TASKMETA_COLLECTION') or
                'taskmeta_collection'
            ),
            'user': os.environ.get('MONGO_USER'),
            'password': os.environ.get('MONGO_PASSWORD'),
        }
    }


@pytest.fixture(scope='session')

t/unit/tasks/test_trace.py:290:            raise Ignore()
t/unit/tasks/test_trace.py:293:        assert info.state == states.IGNORED
t/unit/tasks/test_trace.py:300:            raise Reject()
t/unit/tasks/test_trace.py:303:        assert info.state == states.REJECTED
t/unit/worker/test_request.py:304:            raise Ignore()
t/unit/worker/test_request.py:315:            raise Reject()
t/unit/worker/test_request.py:331:            raise Reject('rejected')
t/unit/worker/test_request.py:366:            raise Reject('rejected')
t/unit/worker/test_request.py:385:            raise Reject(requeue=False)
t/unit/worker/test_request.py:398:            raise Reject(requeue=True)
t/integration/test_canvas.py:252:    def test_group_chord_group_chain(self, manager):

PROTOCOL 1 CANVAS x.t('before') | group((t('g1a'), t('g1b'))) | %t((t('g2a'), t('g2b')), 'after')
result <class 'celery.result.AsyncResult'> 43e37c1e-0a8c-41db-90a7-ad0f7d05fd7f
parent <class 'celery.result.GroupResult'> 081def56-4fb3-43c4-8ccf-f106c9f4c36a
tuple (('43e37c1e-0a8c-41db-90a7-ad0f7d05fd7f', (('081def56-4fb3-43c4-8ccf-f106c9f4c36a', (('f17c8516-3247-449a-bd83-b8a42695067c', (('8bef8962-b639-48cc-a0db-b2b51ebd1bdc', None), None)), [(('d5d5202a-c157-4a9b-a6e5-1a936bc7cb16', None), None), (('14c55c62-75fd-46f0-863e-0da589a59689', None), None)])), [(('590d9ac4-f7d2-4cd8-acee-b91a66cab21b', None), None), (('3960cb7c-0b47-42a1-95cf-56eaeb84c749', None), None)])), None)
restored:
 AsyncResult 43e37c1e-0a8c-41db-90a7-ad0f7d05fd7f parent= 081def56-4fb3-43c4-8ccf-f106c9f4c36a children= []
   GroupResult 081def56-4fb3-43c4-8ccf-f106c9f4c36a parent= f17c8516-3247-449a-bd83-b8a426950
…[1936 chars omitted]…
-4583-b437-a2ab19e74c32 parent= None children= []
       AsyncResult 4f8f4b23-6128-4f02-84dc-f2716295f953 parent= None children= []
       AsyncResult d0948de7-eae1-4062-81ba-817cb9243209 parent= None children= []
     AsyncResult 73a30d53-0ffd-4984-a936-5070fa0b2cce parent= None children= []

     AsyncResult 8cb14ba3-b3dd-4f63-b67b-7598ad2c2ab8 parent= None children= []
prepared _frozen?
frozen: None

505:    def state(self):
930:class GroupResult(ResultSet):
1098:    def state(self):
            Waiting for tasks within a task may lead to deadlocks.
            Please see :ref:`task-synchronous-subtasks`.

        Arguments:
            timeout (float): The number of seconds to wait for results
                before the operation times out.
            propagate (bool): If any of the tasks raises an exception,
                the exception will be re-raised when this flag is set.
            interval (float): Time to wait (in seconds) before retrying to

                retrieve a result from the set.  Note that this does not have
                any effect when using the amqp result store backend,
                as it does not use polling.
            callback (Callable): Optional callback to be called for every
                result received.  Must have signature ``(task_id, v
…[3866 chars omitted]…
app/trace.py:114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
celery/app/trace.py:588:                    I, R = Info(REJECTED, exc), ExceptionInfo(internal=True)
celery/events/state.py:75:    'rejected': states.REJECTED,
celery/states.py:71:    'REJECTED',
celery/states.py:139:REJECTED = 'REJECTED'
celery/states.py:145:UNREADY_STATES = frozenset({PENDING, RECEIVED, STARTED, REJECTED, RETRY})

return other.results == self.results
        return NotImplemented

    def __repr__(self):
        return f'<{type(self).__name__}: [{", ".join(r.id for r in self.results)}]>'

    @property
    def supports_native_join(self):
        try:
            return self.results[0].supports_native_join
        except IndexError:
            pass

    @property
    def app(self):
        if self._app is None:
            self._app = (self.results[0].app if self.results else
                         current_app._get_current_object())
        return self._app

    @app.setter
    def app(self, app):
        self._app = app

    @property
    def backend(self):
        return self.app.backend if self.app else self.results[0].backend


@Thenable.register
class GroupResult(ResultSet):
    """Like :class:`ResultSet`, but with an associated id.

    This type is returned by :class:`~cele
…[2798 chars omitted]…
lf.id`."""
        return str(self.id)

    def __hash__(self):
        """`hash(self) -> hash(self.id)`."""
        return hash(self.id)

    def __repr__(self):
        return f'<{type(self).__name__}: {self.id}>'

369:    def ready(self):
377:    def successful(self):
381:    def failed(self):
644:    def successful(self):
653:    def failed(self):
676:    def ready(self):
1056:    def ready(self):

result = self.app.AsyncResult(result)
        try:
            self.results.remove(result)
        except ValueError:
            raise KeyError(result)

    def discard(self, result):
        """Remove result from the set if it is a member.

        Does nothing if it's not a member.
        """
        try:

            self.remove(result)
        except KeyError:
            pass

    def update(self, results):
        """Extend from iterable of results."""
        self.results.extend(r for r in results if r not in self.results)

    def clear(self):
        """Remove all results from this set."""
        self.results[:] = []  # don't create new list.

    def successful(self):
        """Return true if all tasks successful.

        Returns:
            bool: true if all of the tasks finished
                successfully (i.e. didn't raise an exception).
        """
   
…[6648 chars omitted]…

    raise KeyError('|'.join(short_opts or [] + long_opts or []))


def _patch_eventlet():
    import eventlet.debug

    eventlet.monkey_patch()
    blockdetect = float(os.environ.get('EVENTLET_NOBLOCK', 0))
    if blockdetect:
        eventlet.debug.hub_blocking_detection(blockdetect, blockdetect)


def _patch_gevent():
    import gevent.monkey
    import gevent.signal

    gevent.monkey.patch_all()

1800	                # seems sensible enough. If that's somehow not the case we'll
  1801	                # end up messing up chord counts and there are all sorts of
  1802	                # awful race conditions to think about. We'll hope it's not!
  1803	                sig, res, group_id = current_task
  1804	                chord_obj = chord if chord is not None else sig.options.get("chord")
  1805	                # We need to check the chord size of each contributing task so
  1806	                # that when we get to the final one, we can correctly set the
  1807	                # size in the backend and the chord can be sensible completed.
  1808	                chord_size += _chord._descend(sig)
  1809	                if chord_obj is not None and next_task is None:
  1810	                    # Per above, sanity check that we only saw one group
  1811	                    app

…[8630 chars omitted]…
at it points to the
  1285	                # group result object.
  1286	                node = res
  1287	                while node.parent:
  1288	                    node = node.parent
  1289	                prev_res = node
  1290	        # Use the last task's actual ID, not the input parameter.
  1291	        self.id = results[0].id if results else last_task_id
  1292	        return tasks, results

940	        tasks = d['kwargs']['tasks']
   941	        if tasks:
   942	            if isinstance(tasks, tuple):  # aaaargh
   943	                tasks = d['kwargs']['tasks'] = list(tasks)
   944	            tasks = [maybe_signature(task, app=app) for task in tasks]
   945	        return cls(tasks, app=app, **d['options'])
   946	
   947	    def __init__(self, *tasks, **options):
   948	        tasks = (regen(tasks[0]) if len(tasks) == 1 and is_list(tasks[0])
   949	                 else tasks)
   950	        super().__init__('celery.chain', (), {'tasks': tasks}, **options
   951	                         )

   952	        self._use_link = options.pop('use_link', None)
   953	        self.subtask_type = 'chain'
   954	        self._frozen = None
   955	
   956	    def __call__(self, *args, **kwargs):
   957	        if self.tasks:
   958	            return self.apply_async(args, kwa
…[8629 chars omitted]…
 # If a chain already exists, we need to extend it with the next
   927	        # tasks in the chain.
   928	        # Issue #5354.
   929	        # WARNING: Be careful not to mutate `options['chain']`.
   930	        return ChainMap({'chain': options['chain'] + tasks},
   931	                        options)
   932	
   933	
   934	@Signature.register_type(name='chain')
   935	class _chain(Signature):

1 1 [('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0)] groups 0 children []
1 2 [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0), ('AsyncResult', 0)] groups 2 children [2, 2]
1 3 [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 3), ('GroupResult', 3), ('AsyncResult', 0), ('AsyncResult', 0)] groups 2 children [3, 3]
1 4 [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 4), ('GroupResult', 4), ('AsyncResult', 0), ('AsyncResult', 0)] groups 2 children [4, 4]
2 1 [('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0), ('AsyncResult', 0)] groups 0 children []
2 2 [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0), ('AsyncResult', 0)] groups 2 children [2, 2]
2 3 [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 3), ('GroupResult', 3), ('AsyncResult', 0), ('AsyncResult', 0)] groups 2 children [3, 3]
2 4 [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 4), ('GroupResult', 4), ('AsyncResult', 0), ('AsyncResult', 0)] groups 2 children [4, 4]

1835	        options['group_id'] = group_id = (
  1836	            options.pop('task_id', uuid()))
  1837	        return options, group_id, options.get('root_id')
  1838	
  1839	    def _freeze_group_tasks(self, _id=None, group_id=None, chord=None,
  1840	                            root_id=None, parent_id=None, group_index=None):
  1841	        """Freeze the tasks in the group.
  1842	
  1843	        Note:
  1844	            If the group tasks are created from a generator, the tasks generator would
  1845	            not be exhausted, and the tasks would be frozen lazily.
  1846	

  1847	        Returns:
  1848	            tuple: A tuple of the group id, and the AsyncResult of each of the group tasks.
  1849	        """
  1850	        # pylint: disable=redefined-outer-name
  1851	        #  XXX chord is also a class in outer scope.
  1852	        opts = self.options
  1853	        t
…[8629 chars omitted]…
test_apply_contains_chords_containing_empty_chain(self):
1309:    def test_apply_contains_chords_containing_chain_with_empty_tail(self):
1334:    def test_apply_contains_chords_containing_group(self):
1358:    def test_apply_contains_chords_containing_empty_group(self):
1379:    def test_apply_contains_chords_containing_chord(self):
1404:    def test_apply_contains_chords_containing_empty_chord(self):

1 False x.task('before1') | task('before2') | group((task('a1'), task('a2'), task('a3'))) | %(task([task('b1'), task('b2')], 'after1') | task('after2'))
r AsyncResult 74738cf8-7c7e-4f6a-95ab-0f145f2d1a05 parent AsyncResult ea59a6f4-8a90-445d-9552-96da817b9c6c children 0
rr AsyncResult 74738cf8-7c7e-4f6a-95ab-0f145f2d1a05 parent AsyncResult ea59a6f4-8a90-445d-9552-96da817b9c6c children 0
parent0 AsyncResult 74738cf8-7c7e-4f6a-95ab-0f145f2d1a05 parent AsyncResult ea59a6f4-8a90-445d-9552-96da817b9c6c children 0
parent1 AsyncResult ea59a6f4-8a90-445d-9552-96da817b9c6c parent GroupResult 45a1b760-2154-46a8-b259-837f15b06540 children 0
parent2 GroupResult 45a1b760-2154-46a8-b259-837f15b06540 parent GroupResult ea450560-4e72-4f2f-b3f1-789dd5eafdf5 children 2
parent3 GroupResult ea450560-4e72-4f2f-b3f1-789dd5eafdf5 parent AsyncResult b999789c-7db9-44cd-90b0-fcd2dc646baa children 3
parent4 A
…[2966 chars omitted]…
parent1 GroupResult b6fc8259-925a-4e50-b117-ef852b1b5cf2 parent AsyncResult 09c70d5a-8d36-4254-8c66-b7aece46f4d6 children 3
parent2 AsyncResult 09c70d5a-8d36-4254-8c66-b7aece46f4d6 parent AsyncResult 8825ecb5-6d66-4752-93e5-1031c1052319 children 0
parent3 AsyncResult 8825ecb5-6d66-4752-93e5-1031c1052319 parent None None children 0

tuple group fanout lengths
group b6fc8259-925a-4e50-b117-ef852b1b5cf2 3

759:    def __or__(self, other):
960:    def __or__(self, other):
1598:    def __or__(self, other):
2077:    def __or__(self, other):
   400	        #   Works on this, as it's a property
   401	        return _apply(args, kwargs, **options)
   402	
   403	    def _merge(self, args=None, kwargs=None, options=None, force=False):
   404	        """Merge partial args/kwargs/options with existing ones.
   405	
   406	        If the signature is immutable and ``force`` is False, the existing
   407	        args/kwargs will be returned as-is and only the options will be merged.

   408	
   409	        Stamped headers are considered immutable and will not be merged regardless.
   410	
   411	        Arguments:
   412	            args (Tuple): Partial args to be prepended to the existing args.
   413	            kwargs (Dict): Partial kwargs to be merged with existing kwargs.
   414	           
…[6279 chars omitted]…
, t(4)], 'after') | t('after2'))")]
parents [('AsyncResult', '81efb05b-9764-49e5-a32c-1a7bb20bb924', 0), ('AsyncResult', '96f91a02-d72f-4510-aef3-7cc9defcd380', 0), ('GroupResult', '8e398b68-a467-4f0d-8f48-2999890637e6', 2), ('GroupResult', 'f24f769d-e8cd-4952-a187-f1915b840c22', 2), ('AsyncResult', '93b0c781-43c8-44ea-b489-271c65bf2ec9', 0), ('AsyncResult', '739963fa-cd3e-4a96-82be-950ed86bc781', 0)]

625	                        task.backend.mark_as_done(
   626	                            uuid, retval, task_request, publish_result,
   627	                        )
   628	                    except EncodeError as exc:
   629	                        I, R, state, retval = on_error(task_request, exc)
   630	                        # MEMORY LEAK FIX: Clear traceback frames to prevent memory retention (Issue #8882)
   631	                        traceback_clear(exc)
   632	                    else:
   633	                        Rstr = saferepr(R, resultrepr_maxsize)
   634	                        T = monotonic() - time_start
   635	                        if task_on_success:
   636	                            task_on_success(retval, uuid, args, kwargs)

   637	                        if success_receivers:
   638	                            send_success(sender=task, result=retval)
   
…[6322 chars omitted]…
                I, _, _, _ = on_error(task_request, exc)
   684	        return trace_ok_t(R, I, T, Rstr)
   685	
celery/app/trace.py:114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
celery/app/trace.py:347:                 IGNORE_STATES=IGNORE_STATES):
celery/app/trace.py:650:                if state not in IGNORE_STATES:
grep: celery/app/__pycache__/trace.cpython-312.pyc: binary file matches

2fc35be base
[build_sphinx]
source-dir = docs/
build-dir = docs/_build
all_files = 1

[flake8]
# classes can be lowercase, arguments and variables can be uppercase
# whenever it makes the code more readable.
max-line-length = 117
extend-ignore =
    # incompatible with black https://github.com/psf/black/issues/315#issuecomment-395457972

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

[bdist_rp
…[405 chars omitted]…
def _freeze_group_tasks(self, _id=None, group_id=None, chord=None,
1892:        return self.app.GroupResult(*self._freeze_group_tasks(
1899:    def _freeze_tasks(self, tasks, group_id, chord, root_id, parent_id):
1913:    def _freeze_unroll(self, new_tasks, group_id, chord, root_id, parent_id):
2267:        header_result_args = header._freeze_group_tasks(group_id=group_id, chord=body, root_id=root_id)

530	                The task executed successfully.  The :attr:`result` attribute
   531	                then contains the tasks return value.
   532	        """
   533	        return self._get_task_meta()['status']
   534	    status = state  # XXX compat
   535	
   536	    @property
   537	    def task_id(self):
   538	        """Compat. alias to :attr:`id`."""
   539	        return self.id
   540	
   541	    @task_id.setter

   542	    def task_id(self, id):
   543	        self.id = id
   544	
   545	    @property
   546	    def name(self):
   547	        return self._get_task_meta().get('name')
   548	
   549	    @property
   550	    def args(self):
   551	        return self._get_task_meta().get('args')
   552	
   553	    @property
   554	    def kwargs(self):
   555	        return self._get_task_meta().get('kwargs')
   556	
   557	    @property
   558	    def worker(self):
   5
…[8629 chars omitted]…
er
  1980	    tasks as a list.
  1981	
  1982	    Example:
  1983	
  1984	        The chord:
  1985	
  1986	        .. code-block:: pycon
  1987	
  1988	            >>> res = chord([add.s(2, 2), add.s(4, 4)])(sum_task.s())
  1989	
  1990	        is effectively :math:`\Sigma ((2 + 2) + (4 + 4))`:
  1991	
  1992	        .. code-block:: pycon
  1993	
  1994	            >>> res.get()
  1995	            12

Ignore ['first', 'next'] EagerResult SUCCESS next next 5effcbbf-d3f8-4dbb-811a-b4de197bb5b5
get: next
Reject ['first', 'next'] EagerResult SUCCESS next next b0c38c6b-c28b-4d5d-8b4e-81c562ba709e
get: next

Ignore IGNORED Ignore('stop') None
get no arg None
frozenset({'FAILURE', 'REVOKED'})
Reject REJECTED reject requeue=False: stop None
get no arg None
frozenset({'FAILURE', 'REVOKED'})

PROTOCOL 1
BAD 0 1 2 2 canvas %celery.chord([x.t('a0'), x.t('a1')], kwargs={}, header=(x.t('b0'), x.t('b1')), body=x.t('post')) raw [('AsyncResult', 0), ('GroupResult', 2)] restored [('AsyncResult', 0), ('GroupResult', 2)]
BAD 0 1 2 3 canvas %celery.chord([x.t('a0'), x.t('a1')], kwargs={}, header=(x.t('b0'), x.t('b1'), x.t('b2')), body=x.t('post')) raw [('AsyncResult', 0), ('GroupResult', 2)] restored [('AsyncResult', 0), ('GroupResult', 2)]
BAD 0 1 3 2 canvas %celery.chord([x.t('a0'), x.t('a1'), x.t('a2')], kwargs={}, header=(x.t('b0'), x.t('b1')), body=x.t('post')) raw [('AsyncResult', 0), ('GroupResult', 3)] restored [('AsyncResult', 0), ('GroupResult', 3)]
BAD 0 1 3 3 canvas %celery.chord([x.t('a0'), x.t('a1'), x.t('a2')], kwargs={}, header=(x.t('b0'), x.t('b1'), x.t('b2')), body=x.t('post')) raw [('AsyncResult', 0), ('GroupResult', 3)] restored [('AsyncResult', 0), ('GroupResult
…[2396 chars omitted]…
ader=[x.t('b0'), x.t('b1')], body=x.t('post') | t('post')) raw [('AsyncResult', 0), ('GroupResult', 3)] restored [('AsyncResult', 0), ('GroupResult', 3)]
BAD 0 2 3 3 canvas %celery.chord([x.t('a0'), x.t('a1'), x.t('a2')], kwargs={}, header=[x.t('b0'), x.t('b1'), x.t('b2')], body=x.t('post') | t('post')) raw [('AsyncResult', 0), ('GroupResult', 3)] restored [('AsyncResult', 0), ('GroupResult', 3)]
done

320	    """
   321	    # Cleared Tb, but einfo still has a reference to Traceback.
   322	    # exc cleans up the Traceback at the last moment that can be revealed.
   323	    tb = None
   324	    if exc is not None:
   325	        if hasattr(exc, '__traceback__'):
   326	            tb = exc.__traceback__
   327	        else:
   328	            _, _, tb = sys.exc_info()
   329	    else:
   330	        _, _, tb = sys.exc_info()
   331	

   332	    while tb is not None:
   333	        try:
   334	            # MEMORY LEAK FIX: tb.tb_frame.clear() clears ALL frame data including
   335	            # local variables, which is more efficient than accessing f_locals separately.
   336	            # Removed redundant tb.tb_frame.f_locals access that was creating unnecessary references.
   337	            tb.tb_frame.clear()
   338	        except RuntimeError:
   339	            # Ignore t
…[7854 chars omitted]…
assert x, result_from_tuple(x == self.app)
t/unit/tasks/test_result.py:1071:        second_result = result_from_tuple(result.as_tuple(), self.app)
t/unit/tasks/test_result.py:1075:    def test_GroupResult_as_tuple(self):
t/unit/tasks/test_result.py:1083:        (result_id, parent_tuple), group_results = result.as_tuple()
t/unit/tasks/test_result.py:1085:        assert parent_tuple == parent.as_tuple()

1 [('AsyncResult', '38461961-6c5f-44e1-b17d-d04881b633d5', []), ('GroupResult', '59648380-cfa4-4ad3-b2ec-757c39510dc0', ['23500d03-b6df-4785-a486-28682bc5aadc', '5a9126a7-7cb0-4d1b-b2ad-8db59c910dd6']), ('GroupResult', '8acd4bee-932f-45fe-8d7d-7c2475469ce4', ['baf78b65-b6cd-4c4b-a5a2-1b17cc2a47f5', '90cdeb5c-d829-4dc3-ac13-eb2951366257', 'f5b305bb-b0e0-48e1-8037-b45a3c7dac78']), ('AsyncResult', '885c9b3e-c886-41fa-a8d5-29c208b92e35', [])]
1 [('AsyncResult', '38461961-6c5f-44e1-b17d-d04881b633d5', []), ('GroupResult', '59648380-cfa4-4ad3-b2ec-757c39510dc0', ['23500d03-b6df-4785-a486-28682bc5aadc', '5a9126a7-7cb0-4d1b-b2ad-8db59c910dd6']), ('GroupResult', '8acd4bee-932f-45fe-8d7d-7c2475469ce4', ['baf78b65-b6cd-4c4b-a5a2-1b17cc2a47f5', '90cdeb5c-d829-4dc3-ac13-eb2951366257', 'f5b305bb-b0e0-48e1-8037-b45a3c7dac78']), ('AsyncResult', '885c9b3e-c886-41fa-a8d5-29c208b92e35', [])]
Traceback (most recent call last):
  File "<stdin>", line 9, in <module>
TypeError: 'list' object is not callable

515	                try:
   516	                    state = r.state
   517	                except BackendGetMetaError:
   518	                    pass
   519	                else:
   520	                    if state == SUCCESS:
   521	                        info(LOG_IGNORED, {
   522	                            'id': task_request.id,
   523	                            'name': get_task_name(task_request, name),
   524	                            'description': 'Task already completed successfully.'
   525	                        })
   526	                        _root_id = task_request.root_id or uuid

   527	                        _priority = task_request.delivery_info.get('priority') if \
   528	                            inherit_parent_priority else None
   529	                        try:
   530	                            _meta = r._get_task_meta()
   531	                    
…[4625 chars omitted]…
8	                    except EncodeError as exc:
   629	                        I, R, state, retval = on_error(task_request, exc)
   630	                        # MEMORY LEAK FIX: Clear traceback frames to prevent memory retention (Issue #8882)
114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
347:                 IGNORE_STATES=IGNORE_STATES):
650:                if state not in IGNORE_STATES:

11:from celery.result import AsyncResult, EagerResult, GroupResult
330:    @pytest.mark.parametrize('_id, group_id, chord, root_id, parent_id, group_index', [
331:        ('_id', 'group_id', 'chord', 'root_id', 'parent_id', 1),
333:    def test_freezing_args_set_in_options(self, _id, group_id, chord, root_id, parent_id, group_index):
340:            parent_id=parent_id,
349:        assert options['parent_id'] == parent_id
421:        assert result.parent
422:        assert result.parent.parent
423:        assert result.parent.parent.parent is None
479:        assert isinstance(result, GroupResult)
490:        assert isinstance(result, GroupResult)
755:        assert res.parent.get() == 16

756:        assert res.parent.parent.get() == 8
757:        assert res.parent.parent.parent is None
767:        assert x.parent is None
795:    def test_chord_sets_result_parent(self):
803:        as
…[3104 chars omitted]…
atches
celery/result.py:123:    def as_tuple(self):
celery/result.py:125:        return (self.id, parent and parent.as_tuple()), None
celery/result.py:1005:    def as_tuple(self):
celery/result.py:1007:            (self.id, self.parent and self.parent.as_tuple()),
celery/result.py:1008:            [r.as_tuple() for r in self.results]
grep: celery/__pycache__/result.cpython-312.pyc: binary file matches

2088	            sig = self.clone()
  2089	            sig.body = sig.body | other
  2090	            return sig
  2091	        else:
  2092	            return super().__or__(other)
  2093	
  2094	    def freeze(self, _id=None, group_id=None, chord=None,
  2095	               root_id=None, parent_id=None, group_index=None):
  2096	        # pylint: disable=redefined-outer-name
  2097	        #   XXX chord is also a class in outer scope.
  2098	        if not isinstance(self.tasks, group):
  2099	            self.tasks = group(self.tasks, app=self.app)

  2100	        # first freeze all tasks in the header
  2101	        header_result = self.tasks.freeze(
  2102	            parent_id=parent_id, root_id=root_id, chord=self.body)
  2103	        self.id = self.tasks.id
  2104	        # secondly freeze all tasks in the body: those that should be called after the header
  2105	
  2106	    
…[8629 chars omitted]…
tasks)
  1591	        super().__init__('celery.group', (), {'tasks': tasks}, **options
  1592	                         )
  1593	        self.subtask_type = 'group'
  1594	
  1595	    def __call__(self, *partial_args, **options):
  1596	        return self.apply_async(partial_args, **options)
  1597	
  1598	    def __or__(self, other):
  1599	        if (
  1600	            isinstance(other, group) and

grep: celery/app/__pycache__/amqp.cpython-312.pyc: binary file matches
grep: celery/app/__pycache__/defaults.cpython-312.pyc: binary file matches
grep: celery/__pycache__/canvas.cpython-312.pyc: binary file matches
celery/app/amqp.py:259:        self.task_protocols = {
celery/app/amqp.py:267:        return self.task_protocols[self.app.conf.task_protocol]
celery/app/defaults.py:295:        protocol=Option(2, type='int', old={'celery_task_protocol'}),
celery/canvas.py:1064:        if use_link is None and app.conf.task_protocol == 1:
celery/canvas.py:1164:        if use_link is None and app.conf.task_protocol == 1:
docs/faq.rst:598:that don't contain the proper fields (as per the task message protocol).
docs/history/changelog-3.0.rst:297:- Worker/``statedb``: Now uses pickle protocol 2 (Python 2.5+)
docs/history/changelog-3.1.rst:44:- Now supports Task protocol 2 from the future 4.0 rele
…[273 chars omitted]…

tocol 1
docs/history/changelog-4.1.rst:192:  return empty value in protocol 2 (Issue #3687).
docs/history/changelog-5.0.rst:94:- Custom headers now propagate when using the protocol 1 hybrid messages (#6374).
docs/history/whatsnew-4.0.rst:348:This version introduces a brand new task message protocol,
docs/userguide/configuration.rst:402:Set the default task message protocol version used to send tasks.

390	        x.apply_async()
   391	        gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
   392	        gr.apply_async.reset_mock()
   393	        x()
   394	        gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
   395	
   396	        self.app.conf.task_always_eager = True
   397	        chunks.apply_chunks(app=self.app, **x['kwargs'])
   398	
   399	
   400	class test_chain(CanvasCase):
   401	

   402	    def test_chain_of_chain_with_a_single_task(self):
   403	        s = self.add.s(1, 1)
   404	        assert chain([chain(s)]).tasks == list(chain(s).tasks)
   405	
   406	    @pytest.mark.parametrize("chain_type", (_chain, chain_subclass))
   407	    def test_clone_preserves_state(self, chain_type):
   408	        x = chain_type(self.add.s(i, i) for i in range(10))
   409	        assert x.clone().tasks == x.tasks
   410	        assert x
…[8629 chars omitted]…
           task = task.clone()
  1207	            elif is_first_task:
  1208	                task.args = tuple(args) + tuple(task.args)
  1209	
  1210	            if isinstance(task, _chain):
  1211	                # splice (unroll) the chain
  1212	                steps_extend(task.tasks)
  1213	                continue
  1214	
  1215	            # TODO why isn't this asserting is_last_task == False?

1 [('AsyncResult', '383d4a29-6423-4c2a-8297-feba67f9263a', 0), ('GroupResult', '898bbbe3-ef04-410d-b5da-4d0761a194d6', 2), ('GroupResult', 'd43ce351-b20b-4027-954e-9bff3e7509c1', 3), ('AsyncResult', '38f0b07c-b632-44e5-a5e6-63a6a9de957e', 0)] [('AsyncResult', '383d4a29-6423-4c2a-8297-feba67f9263a', 0), ('GroupResult', '898bbbe3-ef04-410d-b5da-4d0761a194d6', 2), ('GroupResult', 'd43ce351-b20b-4027-954e-9bff3e7509c1', 3), ('AsyncResult', '38f0b07c-b632-44e5-a5e6-63a6a9de957e', 0)]
2 [('AsyncResult', '46ac6f0c-d396-47a7-8925-31f2d4d2fbb9', 0), ('GroupResult', 'b8be600f-0e78-449c-a943-d0e924312461', 2), ('GroupResult', 'a4d1f5d3-8967-47cc-a9d1-be5e1ceae134', 3), ('AsyncResult', 'f0bfbbc7-d3e4-4f02-a822-f07b2ffb1a94', 0)] [('AsyncResult', '46ac6f0c-d396-47a7-8925-31f2d4d2fbb9', 0), ('GroupResult', 'b8be600f-0e78-449c-a943-d0e924312461', 2), ('GroupResult', 'a4d1f5d3-8967-47cc-a9d1-be5e1ceae134', 3), ('AsyncResult', 'f0bfbbc7-d3e4-4f02-a822-f07b2ffb1a94', 0)]

814:- Replacing a task with a chain which contains a group now returns a result instead of hanging (#9484)
1002:- Replacing a task with a chain which contains a group now returns a result instead of hanging (#9484)
* master
5.6.2

constructor x.task('before1') | task('before2') | group((task('a1'), task('a2'), task('a3'))) | %(task([task('b1'), task('b2')], 'after1') | task('after2')) tasks [('Signature', "x.task('before1')"), ('Signature', "x.task('before2')"), ('group', "group((x.task('a1'), task('a2'), task('a3')))"), ('_chord', "%(x.task([task('b1'), task('b2')], 'after1') | task('after2'))")]
apply EagerResult (([([((('before1',), 'before2'), 'a1'), ((('before1',), 'before2'), 'a2'), ((('before1',), 'before2'), 'a3')], 'b1'), ([((('before1',), 'before2'), 'a1'), ((('before1',), 'before2'), 'a2'), ((('before1',), 'before2'), 'a3')], 'b2')], 'after1'), 'after2') state SUCCESS
  0 EagerResult a07f74dd-0e45-4fb1-be8f-15bfe5630824 state SUCCESS result (([([((('before1',), 'before2'), 'a1'), ((('before1',), 'before2'), 'a2'), ((('before1',), 'before2'), 'a3')], 'b1'), ([((('before1',), 'before2'), 'a1'), ((('b
…[4647 chars omitted]…
esult ('before1',) children 0
tuple (('927e2cb2-caed-4e59-b29f-83c298b73eee', (('fcb07ec6-b4d1-403a-bcab-b4e06adf8ce3', (('4c301910-f315-4414-9592-27b805ba92a5', (('c0a742aa-6a83-4ba3-9182-14619cbbefaa', None), None)), None)), [(('dc47dd4c-b789-4c3c-8d2f-9ab99dff5e3b', None), None), (('5d76a52b-21d5-45d1-989f-4517c8f61188', None), None), (('45fc4b3b-0491-4ee5-99a2-d33ff0a9fbfc', None), None)])), None)

1 1 1 _chain ['Signature', 'group', '_chord'] 
 ('AsyncResult', 'd9f80c4c-c768-4c22-9473-36c5f85d8643', [], ('GroupResult', '427678f6-1d96-492a-aef3-94c3f57edf21', [('AsyncResult', '550f75ae-ce32-4401-8cc8-8709f0d35a56', [], None), ('AsyncResult', '184ac818-7233-4eea-9fda-1c4e131b5190', [], None)], ('GroupResult', 'd2bfef44-7bbb-4ea8-8d93-ae86fd0a2ede', [('AsyncResult', 'cff4ad77-9da0-4fc3-91cf-4c13e2a822d0', [], None), ('AsyncResult', '5edff198-2af6-42e3-b867-3cd4bc626a34', [], None), ('AsyncResult', '223e7817-f925-4af8-923b-720910b2edee', [], None)], ('AsyncResult', 'f59e7583-dc80-4bdb-86ad-b2d8f3cd3a64', [], None))))
restored same shape? True

 1 1 2 _chain ['Signature', 'group', '_chord'] 
 ('AsyncResult', 'de0f140e-697b-4454-9dd2-19c032f53682', [], ('AsyncResult', 'd58af9df-4cfb-4c8d-945d-fecda880adc0', [], ('GroupResult', '7d6649cc-79a7-4c48-b0c6-4ed840d15988', [('AsyncResult'
…[4486 chars omitted]…
lt', '7503dc56-34bb-4ebd-91a5-81b434d98b67', [('AsyncResult', 'c62f835e-8ee3-438b-bc3b-c41fd5ca67ca', [], None), ('AsyncResult', '6dab65a9-9bdc-40e0-a825-1241e3848097', [], None), ('AsyncResult', 'cbe091e9-7eb6-46a5-868c-e1390d87c627', [], None)], ('AsyncResult', 'c6388460-c9fb-4fc2-b6c6-b155d2a0330f', [], ('AsyncResult', 'c5bace01-63d0-4566-83b4-a77a13b5e41b', [], None))))))
restored same shape? True

1360	            an iterable, then that'll be used as the list of signatures
  1361	            to chain instead.  This means that you can use a generator
  1362	            expression.
  1363	
  1364	    Returns:
  1365	        ~celery.chain: A lazy signature that can be called to apply the first
  1366	            task in the chain.  When that task succeeds the next task in the
  1367	            chain is applied, and so on.
  1368	    """
  1369	
  1370	    # could be function, but must be able to reference as :class:`chain`.
  1371	    def __new__(cls, *tasks, **kwargs):

  1372	        # This forces `chain(X, Y, Z)` to work the same way as `X | Y | Z`
  1373	        if not kwargs and tasks:
  1374	            if len(tasks) != 1 or is_list(tasks[0]):
  1375	                tasks = tasks[0] if len(tasks) == 1 else tasks
  1376	                # if is_list(tasks) and len(tasks) == 
…[1020 chars omitted]…
s chunks_subclass(chunks):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.subtask_type = "chunks_subclass"


class test_Signature(CanvasCase):
    def test_getitem_property_class(self):
        assert Signature.task
        assert Signature.args
        assert Signature.kwargs
        assert Signature.options
1371:    def __new__(cls, *tasks, **kwargs):

docs/history/changelog-3.1.rst:99:    result list are not properly tupled with the as_tuple() method when it is
docs/history/changelog-3.1.rst:100:    a parent result. This is due to the as_tuple() method performing a logical
docs/history/changelog-4.2.rst:285:- **Group Result**: Modify GroupResult.as_tuple() to include parent (fixes #4106) (#4205)
docs/history/changelog-4.2.rst:397:- Matched the behavior of `GroupResult.as_tuple()` to that of `AsyncResult.as_tuple()`.
docs/history/whatsnew-4.0.rst:2144:        >>> tup = result.as_tuple()

70	  (Issue #3262).
    71	
    72	- **Canvas**: Chord now forwards partial arguments to its subtasks.
    73	
    74	    Fix contributed by Tayfun Sen.
    75	
    76	- **App**: Arguments to app such as ``backend``, ``broker``, etc
    77	  are now pickled and sent to the child processes on Windows.
    78	
    79	    Fix contributed by Jeremy Zafran.
    80	
    81	- **Deployment**: Generic init scripts now supports being symlinked

    82	  in runlevel directories (Issue #3208).
    83	
    84	- **Deployment**: Updated CentOS scripts to work with CentOS 7.
    85	
    86	    Contributed by Joe Sanford.
    87	
    88	- **Events**: The curses monitor no longer crashes when the
    89	  result of a task is empty.
    90	
    91	    Fix contributed by Dongweiming.
    92	
    93	- **Worker**: ``repr(worker)`` would crash when called early
    94	  in the startup process (Issue #251
…[4219 chars omitted]…
aseAsyncResult``, use ``AsyncResult`` for instance checks
  2149	  instead.
  2150	
  2151	- Removed ``TaskSetResult``, use ``GroupResult`` instead.
  2152	
  2153	    - ``TaskSetResult.total`` -> ``len(GroupResult)``
  2154	
  2155	    - ``TaskSetResult.taskset_id`` -> ``GroupResult.id``
  2156	
  2157	- Removed ``ResultSet.subtasks``, use ``ResultSet.results`` instead.
  2158	
  2159	
  2160	TaskSet

895:    def __len__(self):
979:    def __bool__(self):
   700	    def revoke(self, connection=None, terminate=False, signal=None,
   701	               wait=False, timeout=None):
   702	        """Send revoke signal to all workers for all tasks in the set.
   703	
   704	        Arguments:
   705	            terminate (bool): Also terminate the process currently working
   706	                on the task (if any).
   707	            signal (str): Name of signal to send to process if terminate.
   708	                Default is TERM.
   709	            wait (bool): Wait for replies from worker.

   710	                The ``timeout`` argument specifies the number of seconds
   711	                to wait.  Disabled by default.
   712	            timeout (float): Time in seconds to wait for replies when
   713	                the ``wait`` argument is enabled.
   714	        """
   715	  
…[2629 chars omitted]…
ulfilled(self, result):
   120	        self.backend.remove_pending_result(self)
   121	        return result
   122	
   123	    def as_tuple(self):
   124	        parent = self.parent
   125	        return (self.id, parent and parent.as_tuple()), None
   126	
   127	    def as_list(self):
   128	        """Return as a list of task IDs."""
   129	        results = []
   130	        parent = self.parent

880	                callback(task_id, value)
   881	            else:
   882	                acc[order_index[task_id]] = value
   883	        return acc
   884	
   885	    def _iter_meta(self, **kwargs):
   886	        return (meta for _, meta in self.backend.get_many(
   887	            {r.id for r in self.results}, max_iterations=1, **kwargs
   888	        ))
   889	
   890	    def _failed_join_report(self):
   891	        return (res for res in self.results

   892	                if res.backend.is_cached(res.id) and
   893	                res.state in states.PROPAGATE_STATES)
   894	
   895	    def __len__(self):
   896	        return len(self.results)
   897	
   898	    def __eq__(self, other):
   899	        if isinstance(other, ResultSet):
   900	            return other.results == self.results
   901	        return NotImplemented
   902	
   903	    def __repr__(self):
   904
…[705 chars omitted]…
f.id or self.results)
   981	    __nonzero__ = __bool__  # Included for Py2 backwards compatibility
   982	
   983	    def __eq__(self, other):
   984	        if isinstance(other, GroupResult):
   985	            return (
   986	                other.id == self.id and
   987	                other.results == self.results and
   988	                other.parent == self.parent
979:    def __bool__(self):

1 False direct

 1 False single_list

 1 False from_group
BAD pre post 1 1 canvastype chain ['group'] RAW [('GroupResult', 9)] REST [('GroupResult', 9)]
BAD pre post 1 2 canvastype chain ['group'] RAW [('GroupResult', 10)] REST [('GroupResult', 10)]
BAD pre post 1 3 canvastype chain ['group'] RAW [('GroupResult', 11)] REST [('GroupResult', 11)]
BAD pre post 2 1 canvastype chain ['group'] RAW [('GroupResult', 10)] REST [('GroupResult', 10)]
BAD pre post 2 2 canvastype chain ['group'] RAW [('GroupResult', 11)] REST [('GroupResult', 11)]
BAD pre post 2 3 canvastype chain ['group'] RAW [('GroupResult', 12)] REST [('GroupResult', 12)]
BAD pre post 3 1 canvastype chain ['group'] RAW [('GroupResult', 11)] REST [('GroupResult', 11)]

BAD pre post 3 2 canvastype chain ['group'] RAW [('GroupResult', 12)] REST [('GroupResult', 12)]
BAD pre post 3 3 canvastype chain ['group'] RAW [('GroupResult'
…[7814 chars omitted]…
)]
BAD pre post 2 3 canvastype chain ['group'] RAW [('GroupResult', 12)] REST [('GroupResult', 12)]
BAD pre post 3 1 canvastype chain ['group'] RAW [('GroupResult', 11)] REST [('GroupResult', 11)]
BAD pre post 3 2 canvastype chain ['group'] RAW [('GroupResult', 12)] REST [('GroupResult', 12)]
BAD pre post 3 3 canvastype chain ['group'] RAW [('GroupResult', 13)] REST [('GroupResult', 13)]

 2 True pipe

before 1 (True, True) [('AsyncResult', 0), ('GroupResult', 4), ('GroupResult', 3), ('AsyncResult', 0)]
tuple? (('a8569fed-2e95-436d-b26f-5051bbedb38e', (('91c15d7a-cdc7-46e6-9345-4e862f27cdfa', (('de052f09-0e7b-4231-8674-1bb62acd4643', (('31147c6e-0789-4704-b505-76b38f1b6ec8', None), None)), [(('0ea16b56-d98c-4985-9beb-ce320838de2a', None), None), (('ca86c08a-13f7-40d6-a849-5b00ae75d013', None), None), (('7e8b4055-5608-4c30-86ba-50c2ab4a519f', None), None)])), [(('7426078e-228f-4e0c-aa4d-9726c14ea41e', None), None), (('de5cfe88-471e-4a1e-a092-8a24b3070bf2', None), None), (('63fdb8ba-fadd-4eb4-9458-59ad27ee436b', None), None), (('1423ba5a-3ff6-466c-b74d-001b34bd64dd', None), None)])), None)
after [('AsyncResult', 0), ('GroupResult', 4), ('GroupResult', 3), ('AsyncResult', 0)]
before 1 (True, False) [('AsyncResult', 0), ('GroupResult', 4), ('GroupResult', 3), ('AsyncResult', 0)]
tuple? 
…[3418 chars omitted]…
16cbcf6c', None), None), (('61db6183-853e-4c3f-bcc0-f8ab23993372', None), None)])), [(('84dedbbd-ca43-4b47-b6bc-4c6ecf66ff25', None), None), (('2d1603cb-1402-4501-9a44-695324ed4ec9', None), None), (('cc5cfc5c-ce2c-4a3d-8bbd-1d4b4913bc81', None), None), (('c9f3e26e-9e4a-439b-b183-bb9bd31a8b11', None), None)])), None)
after [('AsyncResult', 0), ('GroupResult', 4), ('GroupResult', 3), ('AsyncResult', 0)]

headers = self._stamp_headers(visitor_headers, append_stamps, **headers)
            self.body.stamp(visitor, append_stamps, **headers)

    def apply_async(self, args=None, kwargs=None, task_id=None,
                    producer=None, publisher=None, connection=None,
                    router=None, result_cls=None, **options):
        args = args if args else ()
        kwargs = kwargs if kwargs else {}
        args = (tuple(args) + tuple(self.args)
                if args and not self.immutable else self.args)
        body = kwargs.pop('body', None) or self.kwargs['body']
        kwargs = dict(self.kwargs['kwargs'], **kwargs)

        body = body.clone(**options)
        app = self._get_app(body)
        tasks = (self.tasks.clone() if isinstance(self.tasks, group)
                 else group(self.tasks, app=app, task_id=self.options.get('task_id', uuid())))
        if ap
…[6581 chars omitted]…
          # 1. link the error callback to each task in the header
            # 2. link the error callback to the body
            # 3. return the error callback
            # In summary, up to 4 lines of code + updating the method docstring.
            warnings.warn(
                "task_allow_error_cb_on_chord_header=False is pending deprecation in "
                "a future release of Celery.\n"

celery/backends/asynchronous.py:15:from celery import states
celery/backends/cassandra.py:4:from celery import states
celery/backends/database/models.py:7:from celery import states
celery/backends/database/__init__.py:5:from celery import states
celery/backends/elasticsearch.py:7:from celery import states
celery/backends/mongodb.py:8:from celery import states
celery/backends/redis.py:13:from celery import states
celery/backends/rpc.py:13:from celery import states
celery/contrib/testing/manager.py:12:from celery import states
celery/events/state.py:31:from celery import states
REJECTED IGNORED

Ignore None => IGNORED Ignore None
Reject None => REJECTED Reject None
Ignore False => IGNORED Ignore None
Reject False => REJECTED Reject None
Ignore True => IGNORED Ignore None
Reject True => REJECTED Reject None

293eec0604f7beee2a478287522347c35f200623b9873329e4b1db58b71a9778  -
  File "<stdin>", line 8
    define=1
IndentationError: unexpected unindent

1 direct tuple (('70273194-77c6-42c2-ad58-41f3d07dff22', (('28730bdd-c89b-4afc-a6ee-cf0239b6924c', (('102ffd94-4b26-40e1-b629-a6b4e9cafeb0', (('4313359d-f60b-42ce-a73a-ce9f3c22bce4', None), None)), [(('93335dd0-cd43-4157-9709-bb34d500476c', None), None), (('a2ac6f27-4c7f-46fc-b23f-b28d7f24fbcd', None), None), (('de77ff7a-464a-4bc2-9da2-67241ffebf6c', None), None)])), [(('06d2f4de-aad7-41ec-a308-bdcbd4bde905', None), None), (('4854a73c-108a-485c-bc8f-60875c2a2d8e', None), None), (('bc339608-15d6-48f5-9d1c-7b8320620b37', None), None), (('4ae0820e-c01f-4520-9b9d-c3ff138faa98', None), None)])), None)
restored [('AsyncResult', '70273194-77c6-42c2-ad58-41f3d07dff22', None), ('GroupResult', '28730bdd-c89b-4afc-a6ee-cf0239b6924c', 4), ('GroupResult', '102ffd94-4b26-40e1-b629-a6b4e9cafeb0', 3), ('AsyncResult', '4313359d-f60b-42ce-a73a-ce9f3c22bce4', None)]

 1 chainclass tuple (('30e1d7a7-7d
…[3874 chars omitted]…
1de9a4e54e3', None), None), (('d5c3727c-d022-4fd2-8f76-e00ed7552fa7', None), None), (('40b76f8b-977d-4f45-95ae-476bd71efe15', None), None)])), None)
restored [('AsyncResult', '70ee9fb9-f411-4e38-b7bf-e6e0fae92a3b', None), ('GroupResult', 'de09ec0d-7937-4e77-846f-3a3e208f3d7f', 4), ('GroupResult', '1aa2cf75-1311-41a7-a1f1-3a90a4ad155c', 3), ('AsyncResult', 'f6e730ad-a0d3-4c55-b960-f709c34d4711', None)]

celery/contrib/pytest.py:173:def celery_app(request,
188:def AAA_reset_CELERY_LOADER_env():
190:    assert not os.environ.get('CELERY_LOADER')
207:        request.instance.CELERY_TEST_CONFIG = celery_config
t/integration/test_canvas.py:244:        prev = chain_add.app.conf.task_always_eager
t/integration/test_canvas.py:245:        chain_add.app.conf.task_always_eager = True
t/integration/test_canvas.py:249:        chain_add.app.conf.task_always_eager = prev
t/integration/test_canvas.py:1778:        prev = chord_add.app.conf.task_always_eager
t/integration/test_canvas.py:1779:        chord_add.app.conf.task_always_eager = True
t/integration/test_canvas.py:1783:        chord_add.app.conf.task_always_eager = prev
t/integration/test_mem_leak_in_exception_handling.py:27:            task_always_eager=True,
t/unit/app/test_app.py:338:            assert app.conf.task_always_eager == 4

t/unit/
…[2616 chars omitted]…
e
t/unit/tasks/test_stamping.py:922:        self.app.conf.task_always_eager = True
t/unit/tasks/test_stamping.py:1193:        self.app.conf.task_always_eager = True
t/unit/tasks/test_stamping.py:1224:        self.app.conf.task_always_eager = True
t/unit/tasks/test_tasks.py:1778:        self.app.conf.task_always_eager = True
t/unit/tasks/test_tasks.py:1787:        self.app.conf.task_always_eager = True

protocol=1, canvas=repro.task('before') | group((task('first-a'), task('first-b'))) | group((task('second-a'), task('second-b'))) | task('after')
result: [('AsyncResult', 'f8a3a10b-3f0e-423b-9884-9c6a64deab90', 0), ('GroupResult', '335899af-d160-4f94-b242-41bfe44c13de', 2), ('GroupResult', '71b72a02-eec8-4ce0-a2ca-51645bc3fa0c', 2), ('AsyncResult', 'ccba99f1-55c9-4132-8877-73243a2e5827', 0)]
tuple: (('f8a3a10b-3f0e-423b-9884-9c6a64deab90', (('335899af-d160-4f94-b242-41bfe44c13de', (('71b72a02-eec8-4ce0-a2ca-51645bc3fa0c', (('ccba99f1-55c9-4132-8877-73243a2e5827', None), None)), [(('6fdf14fb-2c96-4c65-94b4-68de88a099a4', None), None), (('d3fdf2d2-f3b7-43fc-b721-2531bf0b9aae', None), None)])), [(('84253835-5a5d-406c-8ff9-3266d24fc835', None), None), (('5e7019f5-ee8c-4aa8-833b-03fee4e77d6a', None), None)])), None)
restored: [('AsyncResult', 'f8a3a10b-3f0e-423b-9884-9c6a64deab90', 0), ('G
…[846 chars omitted]…
7dee25cb6591', None), None)])), [(('2a68508d-14f3-42ee-90c7-e46dae1f4992', None), None), (('db803a08-954a-47bc-93d3-4571f92f3fe0', None), None)])), None)
restored: [('AsyncResult', 'e691a672-2bd1-43a9-a721-f9b3b3038bea', 0), ('GroupResult', 'f06817e6-f2cc-4776-9ae0-9f446a3ef7a8', 2), ('GroupResult', '9a97fc78-c217-43e9-8979-ccac3017882a', 2), ('AsyncResult', '3e219f7d-4b5f-409a-82ac-f4fab400a5c7', 0)]

done

"""Internal state.

This is an internal module containing thread state
like the ``current_app``, and ``current_task``.

This module shouldn't be used directly.
"""

import os
import sys
import threading
import weakref

from celery.local import Proxy
from celery.utils.threads import LocalStack

__all__ = (
    'set_default_app', 'get_current_app', 'get_current_task',
    'get_current_worker_task', 'current_app', 'current_task',
    'connect_on_app_finalize',
)

#: Global default app used when no current app.
default_app = None

#: Function returning the app provided or the default app if none.
#:
#: The environment variable :envvar:`CELERY_TRACE_APP` is used to
#: trace app leaks.  When enabled an exception is raised if there
#: is no active app.
app_or_default = None

#: List of all app instances (weakrefs), mustn't be used directly.
_apps = weakref.WeakSet()

#: Global set of functio
…[1092 chars omitted]…
es
celery/backends/database/__init__.py:5:from celery import states
celery/backends/elasticsearch.py:7:from celery import states
celery/backends/mongodb.py:8:from celery import states
celery/backends/redis.py:13:from celery import states
celery/backends/rpc.py:13:from celery import states
celery/contrib/testing/manager.py:12:from celery import states
celery/events/state.py:31:from celery import states

"""Composing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
"""

import itertools
import operator
import types
import warnings
from abc import ABCMeta, abstractmethod

from collections import deque
from collections.abc import MutableSequence
from copy import deepcopy
from functools import partial as _partial
from functools import reduce
from operator import itemgetter
from types import GeneratorType

from kombu.utils.functional import fxrange, reprcall
from kombu.utils.objects import cached_property
from kombu.utils.uuid import uuid
from vine import barrier

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
from c
…[519 chars omitted]…

    was a single task without pre-knowledge."""
    # Issue #1656
    try:
        size = len(group.tasks)
    except TypeError:
        try:
            size = group.tasks.__length_hint__()
        except (AttributeError, TypeError):
            return group
        else:
            return list(group.tasks)[0] if size == 1 else group
    else:
        return group.tasks[0] if size == 1 else group

1 simple chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]
1 chainmembers chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]
1 innergroups chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]
1 chords chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]
2 simple chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]
2 chainmembers chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]
2 innergroups chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]
2 chords chain 0 rootparent [('AsyncResult', 0), ('AsyncResult', 0)]

1 1 1 x.t('p0') | group((t('a'), t('b'))) | %t((t('c'), t('d')), 'q0') EagerResult raw [('EagerResult', '6887b909-8686-4a6d-8b60-5bf7937c8fc3', 0), ('GroupResult', '5a402690-a0b3-490b-9cad-319e6de74536', 2), ('EagerResult', '93c878e0-038a-479d-9780-7475b44e9b76', 0)] rest [('AsyncResult', '6887b909-8686-4a6d-8b60-5bf7937c8fc3', 0), ('GroupResult', '5a402690-a0b3-490b-9cad-319e6de74536', 2), ('AsyncResult', '93c878e0-038a-479d-9780-7475b44e9b76', 0)]
1 2 2 x.t('p0') | t('p1') | group((t('a'), t('b'))) | %(t([t('c'), t('d')], 'q0') | t('q1')) EagerResult raw [('EagerResult', '90a62316-8ba4-4580-8f48-d82857fd34b0', 0), ('GroupResult', '4d680291-f891-4d21-943b-6e458ca5a830', 2), ('EagerResult', '095fde2b-c5ba-49fb-9bfa-86c7f40d7f02', 0), ('EagerResult', 'dcc0e1f1-224e-4c10-aea0-42f2d160ca74', 0)] rest [('AsyncResult', '90a62316-8ba4-4580-8f48-d82857fd34b0', 0), ('GroupResult', '4d680291-f
…[798 chars omitted]…
c4ea-bc09-40b3-82c1-c4e3e0fa250c', 2), ('EagerResult', '9b2f0dad-cc8b-4180-b8dd-2b8ed2429ad3', 0), ('EagerResult', 'ee4d213f-97ab-4d21-8cc4-c2a1074cc2b5', 0)] rest [('AsyncResult', '2f1cddce-db1c-421f-a003-98893b098a81', 0), ('GroupResult', 'b820c4ea-bc09-40b3-82c1-c4e3e0fa250c', 2), ('AsyncResult', '9b2f0dad-cc8b-4180-b8dd-2b8ed2429ad3', 0), ('AsyncResult', 'ee4d213f-97ab-4d21-8cc4-c2a1074cc2b5', 0)]

assert x.id == '2'

    @pytest.mark.usefixtures('depends_on_current_app')
    def test_reduce_direct(self):
        x = AsyncResult('1', app=self.app)
        fun, args = x.__reduce__()
        assert fun(*args) == x

    def test_children(self):
        x = self.app.AsyncResult('1')
        children = [EagerResult(str(i), i, states.SUCCESS) for i in range(3)]
        x._cache = {'children': children, 'status': states.SUCCESS}

        x.backend = Mock()
        assert x.children
        assert len(x.children) == 3

    def test_propagates_for_parent(self):
        x = self.app.AsyncResult(uuid())
        x.backend = Mock(name='backend')
        x.backend.get_task_meta.return_value = {}
        x.backend.wait_for_pending.return_value = 84
        x.parent = EagerResult(uuid(), KeyError('foo'), states.FAILURE)
        with pytest.raises(KeyError):
            x.get(propagate=Tr
…[3388 chars omitted]…
self.app.AsyncResult(self.task1['id'])
        r3 = self.app.AsyncResult(self.task2['id'])
        assert r1 == r2
        assert r1 != r3
        assert r1 == r2.id
        assert r1 != r3.id

    @pytest.mark.usefixtures('depends_on_current_app')
    def test_reduce_restore(self):
        r1 = self.app.AsyncResult(self.task1['id'])
        fun, args = r1.__reduce__()
        assert fun(*args) == r1

2fc35be base
docs/history/changelog-4.2.rst:285:- **Group Result**: Modify GroupResult.as_tuple() to include parent (fixes #4106) (#4205)
docs/history/changelog-4.2.rst:397:- Matched the behavior of `GroupResult.as_tuple()` to that of `AsyncResult.as_tuple()`.
t/unit/tasks/test_result.py:1075:    def test_GroupResult_as_tuple(self):

t/unit/tasks/test_chord.py:381:        assert self.app.backend.apply_chord.call_count == 2
t/unit/tasks/test_chord.py:391:        self.app.backend.set_chord_size.assert_called_once_with(result.parent.id, group_size)
t/unit/tasks/test_canvas.py:421:        assert result.parent
t/unit/tasks/test_canvas.py:422:        assert result.parent.parent
t/unit/tasks/test_canvas.py:423:        assert result.parent.parent.parent is None
t/unit/tasks/test_canvas.py:693:        c.apply_async(chord='some_chord_id')
t/unit/tasks/test_canvas.py:795:    def test_chord_sets_result_parent(self):
t/unit/tasks/test_canvas.py:1257:    def test_apply_contains_chord(self):
t/unit/tasks/test_canvas.py:1274:        # the header of the `child_chord`, just before we apply the last task.
t/unit/tasks/test_canvas.py:1277:    def test_apply_contains_chords_containing_chain(self):
t/unit/tasks/test_canvas.py:1299:    
…[7198 chars omitted]…

 task_id=None)
2050:    def test_chord_run_body_freeze_prevents_task_id_empty_error(self):
2059:        test_chord = chord(header, body)
2067:        result = test_chord.run(header, body, (), task_id=external_task_id)
2077:    def test_chord_run_body_freeze_with_no_external_task_id(self):
2081:        test_chord = chord(header, body)
2087:        result = test_chord.run(header, body, (), task_id=None)

194:class _regen(UserList, list):
195-    # must be subclass of list so that json can encode.
196-
197-    def __init__(self, it):
198-        # pylint: disable=super-init-not-called
199-        # UserList creates a new list and sets .data, so we don't
200-        # want to call init here.
201-        self.__it = it
202-        self.__consumed = []
203-        self.__done = False
204-
205-    def __reduce__(self):

206-        return list, (self.data,)
207-
208-    def map(self, func):
209-        self.__consumed = [func(el) for el in self.__consumed]
210-        self.__it = map(func, self.__it)
211-
212-    def __length_hint__(self):
213-        return self.__it.__length_hint__()
214-
215-    def __lookahead_consume(self, limit=None):
216-        if not self.__done and (limit is None or limit > 0):
217-            it = iter(self.__it)
218-            try:
219-                now = nex
…[5221 chars omitted]…
elif kind == param.KEYWORD_ONLY:
346-                kwonlyargs.append(name)
347-                if param.default is not param.empty:
348-                    kwonlydefaults[name] = param.default
349-            elif kind == param.VAR_KEYWORD:
350-                varkw = name
351-        return inspect.FullArgSpec(
352-            args=args,
353-            varargs=varargs,
354-            varkw=varkw,

810	        Raises:
   811	
   812	            celery.exceptions.Retry:
   813	                To tell the worker that the task has been re-sent for retry.
   814	                This always happens, unless the `throw` keyword argument
   815	                has been explicitly set to :const:`False`, and is considered
   816	                normal operation.
   817	        """
   818	        request = self.request
   819	        retries = request.retries + 1
   820	        if max_retries is not None:
   821	            self.override_max_retries = max_retries

   822	        max_retries = self.max_retries if max_retries is None else max_retries
   823	
   824	        # Not in worker or emulated by (apply/always_eager),
   825	        # so just raise the original exception.
   826	        if request.called_directly:
   827	            # raises orig stack if PyErr_Occurred,
   828	    
…[1461 chars omitted]…
   861	        try:
   862	            S.apply_async()
   863	        except Exception as exc:
   864	            raise Reject(exc, requeue=False)
   865	        if throw:
   866	            raise ret
   867	        return ret
   868	
   869	    def apply(self, args=None, kwargs=None,
   870	              link=None, link_error=None,
756:    def retry(self, args=None, kwargs=None, exc=None, throw=True,

t/unit/tasks/test_result.py:38:    elif task['state'] == states.RETRY:
t/unit/tasks/test_result.py:74:        self.task4 = mock_task('task3', states.RETRY, KeyError('red'))
t/unit/tasks/test_result.py:367:        res = self.app.AsyncResult(self.task4['id'])  # has RETRY state
t/unit/tasks/test_result.py:378:        res = self.app.AsyncResult(self.task4['id'])  # has RETRY state
t/unit/tasks/test_result.py:400:        # RETRY state should also exist
t/unit/tasks/test_result.py:1002:        res = EagerResult('x', 'x', states.RETRY)
t/unit/tasks/test_result.py:1004:        assert res.state == states.RETRY
t/unit/tasks/test_result.py:1005:        assert res.status == states.RETRY
t/unit/tasks/test_result.py:1008:        res = EagerResult('x', 'x', states.RETRY)
t/unit/tasks/test_trace.py:405:        assert info.state == states.RETRY

1 no [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 3), ('AsyncResult', 0), ('AsyncResult', 0)]
1 yes [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 3), ('AsyncResult', 0), ('AsyncResult', 0)]
2 no [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 3), ('AsyncResult', 0), ('AsyncResult', 0)]
2 yes [('AsyncResult', 0), ('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 3), ('AsyncResult', 0), ('AsyncResult', 0)]

1 explicitapp x.t('p0') | group((t('a'), t('b'))) | group((t('c'), t('d'))) | t('p1') ['Signature', 'group', 'group', 'Signature'] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
1 withoutapp x.t('p0') | group((t('a'), t('b'))) | %t((t('c'), t('d')), 'p1') ['Signature', 'group', '_chord'] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
2 explicitapp x.t('p0') | group((t('a'), t('b'))) | group((t('c'), t('d'))) | t('p1') ['Signature', 'group', 'group', 'Signature'] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
2 withoutapp x.t('p0') | group((t('a'), t('b'))) | %t((t('c'), t('d')), 'p1') ['Signature', 'group', '_chord'] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]

c1 = chain(
            add.s(1, 2), fail.s(), add.s(3, 4),
        )
        res = c1()

        with pytest.raises(ExpectedException):
            res.get(propagate=True)

        with pytest.raises(ExpectedException):
            res.parent.get(propagate=True)

    @flaky

    def test_chain_inside_group_receives_arguments(self, manager):
        c = (
            add.s(5, 6) |
            group((add.s(1) | add.s(2), add.s(3)))
        )
        res = c()
        assert res.get(timeout=TIMEOUT) == [14, 14]

    @flaky
    def test_eager_chain_inside_task(self, manager):
        from .tasks import chain_add

        prev = chain_add.app.conf.task_always_eager
        chain_add.app.conf.task_always_eager = True

        chain_add.apply_async(args=(4, 8), throw=True).get()

        chain_add.app.conf.task_always_eager = prev

    @flaky
    def test_group_chord_group_chain(self
…[5583 chars omitted]…
Expected at least 2 group ids in redis'
    assert actual[0] == actual[1], 'Expected group ids to be equal'


class test_link_error:
    @flaky
    def test_link_error_eager(self):
        exception = ExpectedException("Task expected to fail", "test")
t/integration/conftest.py:49:def celery_worker():
t/integration/conftest.py:108:def celery_worker_pool():
grep: t/conftest.py: No such file or directory

done

get 1 [[[[[['before'], 'a1'], [['before'], 'a2']], 'b1'], [[[['before'], 'a1'], [['before'], 'a2']], 'b2'], [[[['before'], 'a1'], [['before'], 'a2']], 'b3']], 'after']
1 result raw [('AsyncResult', '0f92f945-d5a7-415b-aab8-2d35eaa685a3', 0, 'SUCCESS'), ('GroupResult', '9819e53b-37bf-40c7-961d-d37abc909454', 3, None), ('GroupResult', 'f747384b-2d8c-4fdb-ae11-3cebc4428f2e', 2, None), ('AsyncResult', 'afeedaeb-55c8-4fdb-a5d2-347c0fd19c4b', 0, 'SUCCESS')]
1 result restored [('AsyncResult', '0f92f945-d5a7-415b-aab8-2d35eaa685a3', 0, 'SUCCESS'), ('GroupResult', '9819e53b-37bf-40c7-961d-d37abc909454', 3, None), ('GroupResult', 'f747384b-2d8c-4fdb-ae11-3cebc4428f2e', 2, None), ('AsyncResult', 'afeedaeb-55c8-4fdb-a5d2-347c0fd19c4b', 0, 'SUCCESS')]
get 2 [[[[[['before'], 'a1'], [['before'], 'a2']], 'b1'], [[[['before'], 'a1'], [['before'], 'a2']], 'b2'], [[[['before'], 'a1'], [['before'], 'a2']
…[196 chars omitted]…
-110c-42d0-aced-dd0fe660641d', 2, None), ('AsyncResult', 'ba4e6de7-f604-4255-ac89-65d58ea73dcf', 0, 'SUCCESS')]
2 result restored [('AsyncResult', '12478915-c5e9-43af-8045-3306cbcfebda', 0, 'SUCCESS'), ('GroupResult', 'e7332521-6edb-46ef-be72-e5998a40ade7', 3, None), ('GroupResult', 'dfadc58a-110c-42d0-aced-dd0fe660641d', 2, None), ('AsyncResult', 'ba4e6de7-f604-4255-ac89-65d58ea73dcf', 0, 'SUCCESS')]

1 generator-chain chain [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
1 generator-chain-as-iterable chain [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
1 group-as-iterable chain [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
2 generator-chain chain [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
2 generator-chain-as-iterable chain [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
2 group-as-iterable chain [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]

BAD 1 chain raw [('AsyncResult', 'efe47095-3236-4a4a-979d-24627e4f6847', 0), ('GroupResult', 'd5af10f9-b513-4a33-bc1a-14b20990a135', 2), ('GroupResult', 'e5de4f19-c79d-4195-8182-b20f1ccc87c7', 2), ('AsyncResult', '62f9202f-0632-4066-a342-108070ff90d3', 0)] rest [('AsyncResult', 'efe47095-3236-4a4a-979d-24627e4f6847', 0), ('GroupResult', 'd5af10f9-b513-4a33-bc1a-14b20990a135', 2), ('GroupResult', 'e5de4f19-c79d-4195-8182-b20f1ccc87c7', 2), ('AsyncResult', '62f9202f-0632-4066-a342-108070ff90d3', 0)] tuples equal True
BAD 2 chain raw [('AsyncResult', '0bd01ce8-f59d-47d9-acd4-c40f2d0478c0', 0), ('GroupResult', '0a7ef8e8-b408-4d15-b212-a4a2e0c4590b', 2), ('GroupResult', 'ab5d79ba-ef7f-41a0-acb5-5087327347ca', 2), ('AsyncResult', 'd12c035c-b909-413e-a0bd-20a14a57b398', 0)] rest [('AsyncResult', '0bd01ce8-f59d-47d9-acd4-c40f2d0478c0', 0), ('GroupResult', '0a7ef8e8-b408-4d15-b212-a4a2e0c4590b', 2), ('GroupResult', 'ab5d79ba-ef7f-41a0-acb5-5087327347ca', 2), ('AsyncResult', 'd12c035c-b909-413e-a0bd-20a14a57b398', 0)] tuples equal True
done

finally:
        _state._tls = prev_tls


@contextmanager
def setup_default_app(app, use_trap=False):
    """Setup default app for testing.

    Ensures state is clean after the test returns.
    """
    prev_current_app = _state.get_current_app()
    prev_default_app = _state.default_app

    prev_finalizers = set(_state._on_app_finalizers)
    prev_apps = weakref.WeakSet(_state._apps)

    try:
        if use_trap:
            with set_trap(app):
                yield
        else:
            yield
    finally:
        _state.set_default_app(prev_default_app)
        _state._tls.current_app = prev_current_app
        if app is not prev_current_app:
            app.close()
        _state._on_app_finalizers = prev_finalizers
        _state._apps = prev_apps

"""Create Celery app instances used for testing."""
import weakref
from contextlib import contextmanager
from copy import deepcopy

from kombu.utils.imports import symbol_by_name

from celery import Celery, _state

#: Contains the default configuration values for the test app.
DEFAULT_TEST_CONFIG = {
    'worker_hijack_root_logger': False,

    'worker_log_color': False,
    'accept_content': {'json'},
    'enable_utc': True,
    'timezone': 'UTC',
    'broker_url': 'memory://',
    'result_backend': 'cache+memory://',
    'broker_heartbeat': 0,
}


class Trap:
    """Trap that pretends to be an app but raises an exception instead.

    This to protect from code that does not properly pass app instances,
    then falls back to the current_app.
    """

    def __getattr__(self, name):
        # Workaround to allow unittest.mock to patch this object
        # in Python 3.8 and above.
  
…[1810 chars omitted]…
ref.WeakSet(_state._apps)

    try:
        if use_trap:
            with set_trap(app):
                yield
        else:
            yield
    finally:
        _state.set_default_app(prev_default_app)
        _state._tls.current_app = prev_current_app
        if app is not prev_current_app:
            app.close()
        _state._on_app_finalizers = prev_finalizers
        _state._apps = prev_apps

.. _changelog:

================
 Change history
================

This document contains change notes for bugfix & new features
in the main branch & 5.6.x series, please see :ref:`whatsnew-5.6` for
an overview of what's new in Celery 5.6.

.. _version-5.6.2:

5.6.2
=====

:release-date: 2026-01-04
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix recursive WorkController instantiation in DjangoWorkerFixup + AttributeError when pool_cls is a string (#10045)
- Bugfix: Revoked tasks now immediately update backend status to REVOKED (#9869)
- Prepare for release: v5.6.2 (#10049)

.. _version-5.6.1:

5.6.1
=====

:release-date: 2025-12-29
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix Redis Sentinel ACL authentication support (#10013)
- Fix: Broker heartbeats not sent during graceful shutdown (#9986)
- docs #5410 -- Document confirm_publish broker transport
…[8630 chars omitted]…
est_request.py                      | 1870 +++++++++
 t/unit/worker/test_revoke.py                       |   10 +
 t/unit/worker/test_state.py                        |  222 +
 t/unit/worker/test_strategy.py                     |  358 ++
 t/unit/worker/test_worker.py                       | 1282 ++++++
 tox.ini                                            |  137 +
 820 files changed, 164904 insertions(+)

295:        protocol=Option(2, type='int', old={'celery_task_protocol'}),
        default_exchange=Option(None, type='string'),  # taken from queue
        default_exchange_type=Option('direct'),
        default_routing_key=Option(None, type='string'),  # taken from queue
        default_rate_limit=Option(type='string'),
        default_priority=Option(None, type='string'),
        eager_propagates=Option(
            False, type='bool', old={'celery_eager_propagates_exceptions'},
        ),
        ignore_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,
2

580	        #   _merge_dictionaries(headers, visitor_headers, aggregate_duplicates=True)
   581	        #   headers["foo"] == ["bar1", "bar2"] -> The stamp is now a list
   582	        #   _merge_dictionaries(headers, visitor_headers, aggregate_duplicates=False)
   583	        #   headers["foo"] == "bar2" -> "bar1" is lost, but the stamp is according to the visitor
   584	
   585	        headers = headers.copy()
   586	
   587	        if "stamped_headers" not in headers:
   588	            headers["stamped_headers"] = list(headers.keys())
   589	
   590	        # Merge headers with visitor headers
   591	        if visitor_headers is not None:

   592	            visitor_headers = visitor_headers or {}
   593	            if "stamped_headers" not in visitor_headers:
   594	                visitor_headers["stamped_headers"] = list(visitor_headers.keys())
   595	
   596	            # S
…[4253 chars omitted]…
(self, other):
   783	        # Python 3.9 introduces | as the merge operator for dicts.
   784	        # We override the in-place version of that operator
   785	        # so that canvases continue to work as they did before.
   786	        return self.__or__(other)
   787	
   788	    def election(self):
   789	        type = self.type
   790	        app = type.app
['ignored', 'follow'] SUCCESS 42 42

1 _chain _chain ['Signature', 'group', 'group', 'Signature'] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
tuple (('0c5ade34-46d0-4cbe-9975-a2e1024c5949', (('be0ecb17-ba69-4686-9483-e6a66282a454', (('08de01cd-f2f6-4d3b-a3a6-7c4b0e096543', (('1188371b-0d63-4171-95e3-254106e22fd7', None), None)), [(('0c3a6b38-43d4-419c-a2ff-2e1141c2dc20', None), None), (('cc69d88a-75b3-4840-a5b4-c507cb2778b3', None), None)])), [(('aa632e03-f600-42de-b1c8-490b799b307a', None), None), (('64666a9c-1fd6-4437-a1e6-c64065000e51', None), None)])), None)
rest [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
1 chain chain ['Signature', 'group', 'group', 'Signature'] [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]
tuple (('50a6af65-feaf-43a1-a1ba-daff0472df25', (('3ef97e09-9147-44ef-9d75-274d94bcbee8', (('8f37d7a9-
…[1314 chars omitted]…
020-ba9d-1eb50a0dd080', (('4308b8e6-3d93-4101-93e2-8a505603a0b7', None), None)), [(('ced021e0-7f40-431a-b93a-997e6a4b2037', None), None), (('c2b3bdd7-758b-439d-bbcc-c289cd69d2f6', None), None)])), [(('69106474-43e0-46fa-84c0-13776eac3578', None), None), (('e1094b09-e88f-4150-84a9-2ac52371fd87', None), None)])), None)
rest [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 2), ('AsyncResult', 0)]

45	    _set_task_join_will_block(False)
    46	    try:
    47	        yield
    48	    finally:
    49	        _set_task_join_will_block(reset_value)
    50	
    51	
    52	@contextmanager
    53	def denied_join_result():
    54	    reset_value = task_join_will_block()
    55	    _set_task_join_will_block(True)
    56	    try:

    57	        yield
    58	    finally:
    59	        _set_task_join_will_block(reset_value)
    60	
    61	
    62	class ResultBase:
    63	    """Base class for results."""
    64	
    65	    #: Parent result (if part of a chain)
    66	    parent = None
    67	
    68	
    69	@Thenable.register
    70	class AsyncResult(ResultBase):
    71	    """Query task state.
    72	
    73	    Arguments:
    74	        id (str): See :attr:`id`.
    75	        backend (Backend): See :attr:`backend`.
    76	    """
    77	
    78	    app = None
    79	
    80	    #: Error raised for timeouts.
    81	    TimeoutError = TimeoutError
    82	
    83	    #: The task's UUID.
    84	    id = None
    85	
    86	    #: The task result backend to use.
    87	    backend = None
    88	
    89	    def __init__(self, id, backend=None,
    90	                 task_name=None,            # deprecated
895:    def __len__(self):
979:    def __bool__(self):

1 False before groups [] after []
1 True before groups [] after []
2 False before groups [] after []
2 True before groups [] after []

before tasks <class 'celery.utils.functional._regen'> <class 'celery.utils.functional._regen'>
canvas tasks [('Signature', <class 'NoneType'>, "x.t('pre')"), ('group', <class 'celery.utils.functional._regen'>, "group(<_regen: [x.t('a0'), t('a1')]>)"), ('group', <class 'celery.utils.functional._regen'>, "group(<_regen: [x.t('b0'), t('b1'), t('b2')]>)"), ('Signature', <class 'NoneType'>, "x.t('post')")]
AsyncResult <class 'NoneType'>  2334bcaf-b8fe-4d40-addc-7aa9f60d95ca
GroupResult <class 'celery.utils.functional._regen'> 3 36037490-6448-4226-8bcd-f65ecd55bc30
GroupResult <class 'celery.utils.functional._regen'> 2 d099ec6e-e196-4157-8d10-0ae890e1ca95
AsyncResult <class 'NoneType'>  e704536d-8c30-498e-a322-5e5a10129ef3
after tuple
AsyncResult <class 'NoneType'>  2334bcaf-b8fe-4d40-addc-7aa9f60d95ca
GroupResult <class 'celery.utils.functional._regen'> 3 36037490-6448-4226-8bcd-f65ecd55bc30
GroupResult <class 'celery.utils.functional._regen'> 2 d099ec6e-e196-4157-8d10-0ae890e1ca95
AsyncResult <class 'NoneType'>  e704536d-8c30-498e-a322-5e5a10129ef3
Traceback (most recent call last):

  File "<stdin>", line 24, in <module>
TypeError: 'NoneType' object is not subscriptable

r AsyncResult
tuple (('7164d883-94c4-4fde-929b-08fd9ba3433d', (('3c3667bb-cfbb-4405-b432-f9491179d09a', None), None)), None)
raw [('AsyncResult', '7164d883-94c4-4fde-929b-08fd9ba3433d', 0), ('AsyncResult', '3c3667bb-cfbb-4405-b432-f9491179d09a', 0)]
rest [('AsyncResult', '7164d883-94c4-4fde-929b-08fd9ba3433d', 0), ('AsyncResult', '3c3667bb-cfbb-4405-b432-f9491179d09a', 0)]

1 0 0 ['group', 'group'] [('GroupResult', 2, 'bc65ac5e-0b51-4161-9ee3-f759ffed3352'), ('GroupResult', 2, '1206b59a-e7ac-49bb-854b-bf41bfe1f3c9')]
1 0 1 ['group', 'group', 'Signature'] [('AsyncResult', 0, 'fed59a17-b93b-4c27-a857-94703a0d2caa'), ('GroupResult', 2, '809ec180-142a-403e-81e3-de0bea49f477')]
1 0 2 ['group', 'group', 'Signature', 'Signature'] [('AsyncResult', 0, '6e9e3ba2-7b8c-48d5-a395-31e85f3011f7'), ('AsyncResult', 0, '11a2aa0b-17b2-4b55-8388-da7b74b2b4c9')]
1 0 3 ['group', 'group', 'Signature', 'Signature', 'Signature'] [('AsyncResult', 0, '3006faae-ab66-4bdf-813a-0dadadfc5a19'), ('AsyncResult', 0, 'ceca1018-0440-4897-bb0d-fde2472b8f5d'), ('AsyncResult', 0, '00744302-5516-4059-ae40-e04f821a63d4')]
1 1 0 ['Signature', 'group', 'group'] [('GroupResult', 2, '035f7626-250c-48ba-b75c-6d105e58297d'), ('GroupResult', 2, 'c3deee49-6bad-4957-ad00-52a20412a2ca'), ('AsyncResult', 
…[6898 chars omitted]…
[('AsyncResult', 0, '20c5fc03-7657-41ca-b61d-6fe47db813ac'), ('AsyncResult', 0, '57c5bbd9-17ad-4236-8bfd-dabb0fa4cfdb')]
2 3 3 ['Signature', 'Signature', 'Signature', 'group', 'group', 'Signature', 'Signature', 'Signature'] [('AsyncResult', 0, '7e3cb106-24a8-4919-ae62-649cacd40e41'), ('AsyncResult', 0, '8ed07fd6-16e6-4c4e-811f-3ee09ae3a0dc'), ('AsyncResult', 0, '015a6596-ab5a-4eb0-9a35-c6a54cc5dbd0')]

1 same_group canvas [('Signature', None), ('group', None), ('group', None), ('Signature', None)] 
 raw [('AsyncResult', 'e8ed8118-93b7-483e-a1ce-c69958229bf4', 0), ('GroupResult', '22d5ddec-0d66-4791-8ad0-20ec0d531192', 2), ('GroupResult', '57d22e69-4b8b-41cb-9820-eec63a235110', 2), ('AsyncResult', '0e850d03-ccbd-46cc-84be-0df1f31c998b', 0)] 
 rest [('AsyncResult', 'e8ed8118-93b7-483e-a1ce-c69958229bf4', 0), ('GroupResult', '22d5ddec-0d66-4791-8ad0-20ec0d531192', 2), ('GroupResult', '57d22e69-4b8b-41cb-9820-eec63a235110', 2), ('AsyncResult', '0e850d03-ccbd-46cc-84be-0df1f31c998b', 0)]
1 same_task_sigs canvas [('Signature', None), ('group', None), ('group', None), ('Signature', None)] 
 raw [('AsyncResult', 'cc46e97c-4dfc-47b3-a4a2-50e0eda34dc7', 0), ('GroupResult', 'a16c01fc-0923-4723-8829-eafcda402b90', 2), ('GroupResult', 'e6a47c78-8109-4000-9bf2-7313f7d98a5a', 2), ('AsyncResult', '
…[3490 chars omitted]…
c5-8cd2-4033-af74-adb4070f427e', 2), ('GroupResult', 'e93a26a7-10d1-484c-b5df-1f031698318d', 2), ('AsyncResult', '0d50002a-1902-448c-9e8b-173f58070bd7', 0)] 
 rest [('AsyncResult', 'e1d819a1-2eca-42f7-b3a4-cc2d1539ff78', 0), ('GroupResult', '60ee76c5-8cd2-4033-af74-adb4070f427e', 2), ('GroupResult', 'e93a26a7-10d1-484c-b5df-1f031698318d', 2), ('AsyncResult', '0d50002a-1902-448c-9e8b-173f58070bd7', 0)]

canvas tasks [('Signature', "x.t('before1')"), ('Signature', "x.t('before2')"), ('group', "group((x.t('a1'), t('a2'), t('a3')))"), ('group', "group((x.t('b1'), t('b2')))"), ('Signature', "x.t('after1')"), ('Signature', "x.t('after2')")]
immediate raw [('AsyncResult', '821251ad-db85-421b-8938-fa7086c5de5b', 0), ('AsyncResult', '0dd3b1a1-681f-4736-9ab3-31940b22b713', 0)] (('821251ad-db85-421b-8938-fa7086c5de5b', (('0dd3b1a1-681f-4736-9ab3-31940b22b713', None), None)), None)
get value [[[[[[[['before1'], 'before2'], 'a1'], [[['before1'], 'before2'], 'a2'], [[['before1'], 'before2'], 'a3']], 'b1'], [[[[['before1'], 'before2'], 'a1'], [[['before1'], 'before2'], 'a2'], [[['before1'], 'before2'], 'a3']], 'b2']], 'after1'], 'after2']
later raw [('AsyncResult', '821251ad-db85-421b-8938-fa7086c5de5b', 0), ('AsyncResult', '0dd3b1a1-681f-4736-9ab3-31940b22b713', 0)] (('821251ad-db85-421b-8938-fa7
…[902 chars omitted]…
, 'a3']], 'b2']], 'after1'], 'after2']
later raw [('AsyncResult', '729cb053-6423-492a-ab6d-91a80621b42d', 0), ('AsyncResult', '98b3b625-7a91-489d-a821-1c2eefea3d25', 0)] (('729cb053-6423-492a-ab6d-91a80621b42d', (('98b3b625-7a91-489d-a821-1c2eefea3d25', None), None)), None)
restored [('AsyncResult', '729cb053-6423-492a-ab6d-91a80621b42d', 0), ('AsyncResult', '98b3b625-7a91-489d-a821-1c2eefea3d25', 0)]

1 0 0 _chord ['Signature', 'Signature', 'Signature'] [('GroupResult', 'af8cccc0-5ea2-4227-a511-1a83e6b82ba4', 3), ('GroupResult', 'af8cccc0-5ea2-4227-a511-1a83e6b82ba4', 3)] [('GroupResult', 'af8cccc0-5ea2-4227-a511-1a83e6b82ba4', 3), ('GroupResult', 'af8cccc0-5ea2-4227-a511-1a83e6b82ba4', 3)]
1 1 0 _chain ['Signature', 'group', 'group'] [('GroupResult', 'bdd15ce6-a2b4-4459-920c-947f42733ad8', 3), ('GroupResult', '7e5ee0cc-ec0e-48c4-b9aa-f74334271795', 3), ('AsyncResult', '7ed12d94-4351-4242-b663-02447a1feff7', 0)] [('GroupResult', 'bdd15ce6-a2b4-4459-920c-947f42733ad8', 3), ('GroupResult', '7e5ee0cc-ec0e-48c4-b9aa-f74334271795', 3), ('AsyncResult', '7ed12d94-4351-4242-b663-02447a1feff7', 0)]
1 1 1 _chain ['Signature', 'group', '_chord'] [('AsyncResult', 'a62ad915-b13e-4562-a3fa-bd6b4bd0337e', 0), ('GroupResult', '4d9077ab-7660-457c-a1e5-02844814a472', 3), ('GroupResult', 'a0c55f16-da
…[8629 chars omitted]…
 'aa544cf9-5062-480f-ba0a-1c4f89cc4dd2', 0), ('AsyncResult', '9fe00919-7909-41d8-b4e5-98ebddb331be', 0), ('GroupResult', '9a395011-59a3-4254-acff-03ae6f22977c', 3), ('GroupResult', 'a26bd0a0-ce8e-4ca9-9c19-8c9fa183e07d', 3), ('AsyncResult', '7d0cb056-1544-4070-bbcc-971f27a00473', 0), ('AsyncResult', 'f9127019-c7e1-4284-b1f9-b40b97eeae97', 0), ('AsyncResult', '09a83325-abb1-4dc9-a3a7-8f84bb2e0153', 0)]

before worker 1 chain ['Signature', 'Signature', 'group', 'group', 'Signature', 'Signature']
inside eager? False
immediate [('AsyncResult', '3a9632b9-9ffd-40e8-b8d8-bc2944d5d49d', 0, '6d8159c8-8ca5-4c5f-a2a9-1c544ce6976f'), ('AsyncResult', '6d8159c8-8ca5-4c5f-a2a9-1c544ce6976f', 0, None)]
tuple (('3a9632b9-9ffd-40e8-b8d8-bc2944d5d49d', (('6d8159c8-8ca5-4c5f-a2a9-1c544ce6976f', None), None)), None)
get [[[[[[[['before1'], 'before2'], 'a1'], [[['before1'], 'before2'], 'a2'], [[['before1'], 'before2'], 'a3']], 'b1'], [[[[['before1'], 'before2'], 'a1'], [[['before1'], 'before2'], 'a2'], [[['before1'], 'before2'], 'a3']], 'b2']], 'after1'], 'after2']
after get [('AsyncResult', '3a9632b9-9ffd-40e8-b8d8-bc2944d5d49d', 0, '6d8159c8-8ca5-4c5f-a2a9-1c544ce6976f'), ('AsyncResult', '6d8159c8-8ca5-4c5f-a2a9-1c544ce6976f', 0, None)]
before worker 2 chain ['Signature', 'Signature', 'group', 'group',
…[360 chars omitted]…
before2'], 'a1'], [[['before1'], 'before2'], 'a2'], [[['before1'], 'before2'], 'a3']], 'b1'], [[[[['before1'], 'before2'], 'a1'], [[['before1'], 'before2'], 'a2'], [[['before1'], 'before2'], 'a3']], 'b2']], 'after1'], 'after2']
after get [('AsyncResult', 'fdbb5aad-8f98-4858-a3dd-6b9d18062056', 0, 'e6a5ccc8-0f52-45c5-a77e-a3448fe9ced3'), ('AsyncResult', 'e6a5ccc8-0f52-45c5-a77e-a3448fe9ced3', 0, None)]

worker ready
PREPARE results:
[('AsyncResult', 'f45d7239-66fb-46a5-8c37-f9f014edd939', 0), ('AsyncResult', 'af5b1789-5121-4ff2-88ee-83817c4ef6cc', 0)]
[('AsyncResult', 'af5b1789-5121-4ff2-88ee-83817c4ef6cc', 0), ('GroupResult', '1ebd936d-ac9c-4907-a8d1-efd12690e82d', 2), ('GroupResult', '6beab4ed-7302-494c-a013-6adc4b4ba5ac', 3), ('AsyncResult', 'bd8811c3-6181-4007-86d5-11371d47beb7', 0), ('AsyncResult', '654a3d67-8aab-4dc0-a54b-040cac5904dd', 0)]
[('AsyncResult', 'bd8811c3-6181-4007-86d5-11371d47beb7', 0), ('AsyncResult', '654a3d67-8aab-4dc0-a54b-040cac5904dd', 0)]
[('AsyncResult', '654a3d67-8aab-4dc0-a54b-040cac5904dd', 0)]
returned [('AsyncResult', 'f45d7239-66fb-46a5-8c37-f9f014edd939', 0), ('AsyncResult', 'af5b1789-5121-4ff2-88ee-83817c4ef6cc', 0)]
tuple (('f45d7239-66fb-46a5-8c37-f9f014edd939', (('af5b1789-5121-4ff2-88ee-83817c4ef6cc', None), None)), None)
worker ready
PREPARE r
…[448 chars omitted]…
5d3bc0de-c093-40a9-a8fa-56016afab52d', 0), ('AsyncResult', '53fa74dc-28b8-4017-af53-2bdf39eedb00', 0)]

[('AsyncResult', '53fa74dc-28b8-4017-af53-2bdf39eedb00', 0)]
returned [('AsyncResult', '4561a8a1-77d7-4e7f-bec5-b287dc3f343c', 0), ('AsyncResult', 'b6de02f8-c90c-4cfc-9e76-99933983bb84', 0)]
tuple (('4561a8a1-77d7-4e7f-bec5-b287dc3f343c', (('b6de02f8-c90c-4cfc-9e76-99933983bb84', None), None)), None)

16:from . import local
    def debug_import(name, locals=None, globals=None,
                     fromlist=None, level=-1, real_import=builtins.__import__):
        glob = globals or getattr(sys, 'emarfteg_'[::-1])(1).f_globals
        importer_name = glob and glob.get('__name__') or 'unknown'
        print(f'-- {importer_name} imports {name}')
        return real_import(name, locals, globals, fromlist, level)
    builtins.__import__ = debug_import

# This is never executed, but tricks static analyzers (PyDev, PyCharm,
# pylint, etc.) into knowing the types of these symbols, and what
# they contain.

STATICA_HACK = True
globals()['kcah_acitats'[::-1].upper()] = False
if STATICA_HACK:  # pragma: no cover
    from celery._state import current_app, current_task
    from celery.app import shared_task
    from celery.app.base import Celery
    from celery.app.task import Task
    from celer
…[2128 chars omitted]…
e closure per task class.

# pylint: disable=redefined-outer-name
# We cache globals and attribute lookups, so disable this warning.
# pylint: disable=broad-except
# We know what we're doing...


__all__ = (
    'TraceInfo', 'build_tracer', 'trace_task',
    'setup_worker_optimizations', 'reset_worker_optimizations',
)

from celery.worker.state import successful_requests

logger = get_logger(__name__)

prepared: [[('AsyncResult', '63d11250-18f5-4372-910a-a57a66917441', 0, '29fec1d8-0631-43cc-b996-16d86e5cd9e8'), ('AsyncResult', '29fec1d8-0631-43cc-b996-16d86e5cd9e8', 0, None)], [('AsyncResult', '29fec1d8-0631-43cc-b996-16d86e5cd9e8', 0, 'fc76a256-f50e-443a-8596-d12d022799f9'), ('GroupResult', 'fc76a256-f50e-443a-8596-d12d022799f9', 2, 'b41bf3a3-5b7e-43cc-bf36-2415937fa6cc'), ('GroupResult', 'b41bf3a3-5b7e-43cc-bf36-2415937fa6cc', 2, '8ae2c724-4ae8-4538-8c37-84ae9a9831b9'), ('AsyncResult', '8ae2c724-4ae8-4538-8c37-84ae9a9831b9', 0, '427db539-2fc9-452a-b1c6-ddaea42697f4'), ('AsyncResult', '427db539-2fc9-452a-b1c6-ddaea42697f4', 0, None)], [('AsyncResult', '8ae2c724-4ae8-4538-8c37-84ae9a9831b9', 0, '427db539-2fc9-452a-b1c6-ddaea42697f4'), ('AsyncResult', '427db539-2fc9-452a-b1c6-ddaea42697f4', 0, None)], [('AsyncResult', '427db539-2fc9-452a-b1c6-ddaea42697f4', 0, None)]]
before sending
…[2630 chars omitted]…
482d-af7f-71a85df98c12', 0, None)]
after get returned [('AsyncResult', '4473f85b-febb-44ab-9b24-bf507bce31b7', 0, '4189797c-f871-482d-af7f-71a85df98c12'), ('AsyncResult', '4189797c-f871-482d-af7f-71a85df98c12', 0, None)]
after get holder [('AsyncResult', '4473f85b-febb-44ab-9b24-bf507bce31b7', 0, '4189797c-f871-482d-af7f-71a85df98c12'), ('AsyncResult', '4189797c-f871-482d-af7f-71a85df98c12', 0, None)]

119:    def test_link_error_eager(self):
133:    def test_link_error_callback_error_callback_retries_eager(self):
151:    def test_link_error_using_signature_eager(self):
175:    def test_simple_chain(self, manager):
180:    def test_single_chain(self, manager):
185:    def test_complex_chain(self, manager):
196:    def test_group_results_in_chain(self, manager):
209:    def test_chain_of_chain_with_a_single_task(self, manager):
213:    def test_chain_on_error(self, manager):
232:    def test_chain_inside_group_receives_arguments(self, manager):
241:    def test_eager_chain_inside_task(self, manager):
252:    def test_group_chord_group_chain(self, manager):

348:    def test_chain_error_handler_with_eta(self, manager):
381:    def test_chain_of_task_a_group_and_a_chord(self, manager):
396:    def test_chain_of_chords_as_groups_chained_to_a_task_with_two_tasks(self,
414:    def test_cha
…[7593 chars omitted]…
                  pool=pool,
                                  loglevel=loglevel,
                                  logfile=logfile,
                                  perform_ping_check=perform_ping_check,
                                  shutdown_timeout=shutdown_timeout,
                                  **kwargs) as worker:
            if perform_ping_check:
                from .tasks import ping

chain x.t('before1') | t('before2') | group((t('a1'), t('a2'), t('a3'))) | group((t('b1'), t('b2'))) | t('after1') | t('after2') ['Signature', 'Signature', 'group', 'group', 'Signature', 'Signature']
run AsyncResult [('AsyncResult', '9a7c46fa-4e50-42c2-8a3e-1d76dd428833', 0), ('AsyncResult', '6aedf871-6a1c-4244-bbec-c7caa389056e', 0)]
before c.run prepare?
apply_async AsyncResult [('AsyncResult', '9a7c46fa-4e50-42c2-8a3e-1d76dd428833', 0), ('AsyncResult', '69351d6e-3ec4-418f-a5b5-87bb39dfb810', 0)]

536:    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
   520	    def start_strategy(self, app, consumer, **kwargs):
   521	        return instantiate(self.Strategy, self, app, consumer, **kwargs)
   522	
   523	    def delay(self, *args, **kwargs):
   524	        """Star argument version of :meth:`apply_async`.
   525	
   526	        Does not support the extra options enabled by :meth:`apply_async`.
   527	
   528	        Arguments:
   529	            *args (Any): Positional arguments passed on to the task.
   530	            **kwargs (Any): Keyword arguments passed on to the task.

   531	        Returns:
   532	            celery.result.AsyncResult: Future promise.
   533	        """
   534	        return self.apply_async(args, kwargs)
   535	
   536	    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
   537	                    
…[8628 chars omitted]…
tasks) or []:
1086:                chain_task.set(replaced_task_nesting=replaced_task_nesting)
1088:        # If the task being replaced is part of a chain, we need to re-create
1091:        for t in reversed(self.request.chain or []):
1092:            chain_task = signature(t, app=self.app)
1093:            chain_task.set(replaced_task_nesting=replaced_task_nesting)
1094:            sig |= chain_task

680	            options.setdefault('priority', self.priority)
   681	
   682	        app = self._get_app()
   683	        if app.conf.task_always_eager:
   684	            with app.producer_or_acquire(producer) as eager_producer:
   685	                serializer = options.get('serializer')
   686	                if serializer is None:
   687	                    if eager_producer.serializer:
   688	                        serializer = eager_producer.serializer
   689	                    else:
   690	                        serializer = app.conf.task_serializer
   691	                body = args, kwargs

   692	                content_type, content_encoding, data = serialization.dumps(
   693	                    body, serializer,
   694	                )
   695	                args, kwargs = serialization.loads(
   696	                    data, content_type, content_encoding,
   697	
…[5843 chars omitted]…
        **options (Any): Extra options to pass on to :meth:`apply_async`.
   809	
   810	        Raises:
   811	
   812	            celery.exceptions.Retry:
   813	                To tell the worker that the task has been re-sent for retry.
   814	                This always happens, unless the `throw` keyword argument
   815	                has been explicitly set to :const:`False`, and is considered

tasks [('Signature', 'x.t'), ('_chord', 'celery.chord'), ('Signature', 'x.t'), ('Signature', 'x.t')]
results
0 [('AsyncResult', '5d835a44-ecdd-4d1b-8931-2f835d65d137', 0), ('AsyncResult', '547710c7-d697-4d40-a99f-920e9f785147', 0)]
1 [('AsyncResult', '547710c7-d697-4d40-a99f-920e9f785147', 0), ('GroupResult', '39bb6fec-24dd-4cb0-917d-20e43af7d552', 2), ('GroupResult', '76ed01f9-0015-4f64-8fda-9379aef07d04', 2), ('AsyncResult', '176e9311-0e57-401a-9fc5-a7a5445af5e1', 0), ('AsyncResult', '2a463f49-bacb-41a8-8111-84585133b39f', 0)]
2 [('AsyncResult', '176e9311-0e57-401a-9fc5-a7a5445af5e1', 0), ('AsyncResult', '2a463f49-bacb-41a8-8111-84585133b39f', 0)]
3 [('AsyncResult', '2a463f49-bacb-41a8-8111-84585133b39f', 0)]

c x.t('p1') | t('p2') | group((t('a1'), t('a2'))) | group((t('b1'), t('b2'))) | t('q1') | t('q2') True False
run args ((), []) {'app': <Celery x at 0x75499aa84110>}
prep args ((), [], (x.t('p1'), x.t('p2'), group((x.t('a1'), t('a2'))), group((x.t('b1'), t('b2'))), x.t('q1'), x.t('q2')), None, None, None, <Celery x at 0x75499aa84110>, None, None, None) {'group_index': None}
prep tasks [('Signature', 'x.t', None), ('_chord', 'celery.chord', None), ('Signature', 'x.t', None), ('Signature', 'x.t', None)]
prep results
res0 AsyncResult 1f4c56e5-01b9-4a7c-8927-e38450a102f4 parent path [('AsyncResult', '1f4c56e5-01b9-4a7c-8927-e38450a102f4', 0), ('AsyncResult', '5185497b-19f4-40f3-9325-c77c91b5a08a', 0)]
res1 AsyncResult 5185497b-19f4-40f3-9325-c77c91b5a08a parent path [('AsyncResult', '5185497b-19f4-40f3-9325-c77c91b5a08a', 0), ('GroupResult', '6530daa6-0dfa-48f4-80fc-bd7b6e66b928', 2), ('Gr
…[517 chars omitted]…
5-01b9-4a7c-8927-e38450a102f4 parent path [('AsyncResult', '1f4c56e5-01b9-4a7c-8927-e38450a102f4', 0), ('AsyncResult', '5185497b-19f4-40f3-9325-c77c91b5a08a', 0)]
APPLY RET AsyncResult 1f4c56e5-01b9-4a7c-8927-e38450a102f4 parent path [('AsyncResult', '1f4c56e5-01b9-4a7c-8927-e38450a102f4', 0), ('AsyncResult', '5185497b-19f4-40f3-9325-c77c91b5a08a', 0)]
r is result? 1f4c56e5-01b9-4a7c-8927-e38450a102f4

in prepare [('AsyncResult', '749d13d6-1caf-47c9-ab20-63dfd41ed369', 0, '8fb85d09-35ea-403b-a5f9-f82fc54ac211'), ('AsyncResult', '8fb85d09-35ea-403b-a5f9-f82fc54ac211', 0, None)]
before helper [('AsyncResult', '749d13d6-1caf-47c9-ab20-63dfd41ed369', 0, '8fb85d09-35ea-403b-a5f9-f82fc54ac211'), ('AsyncResult', '8fb85d09-35ea-403b-a5f9-f82fc54ac211', 0, None)] tasks ['x.t', 'celery.chord', 'x.t']
after helper [('AsyncResult', '749d13d6-1caf-47c9-ab20-63dfd41ed369', 0, '8fb85d09-35ea-403b-a5f9-f82fc54ac211'), ('AsyncResult', '8fb85d09-35ea-403b-a5f9-f82fc54ac211', 0, None)] ChainMap({'chain': [x.t('q2'), %celery.chord(group([x.t('a1'), t('a2')]), kwargs={}, header=group([x.t('b1'), t('b2')]), body=x.t('q1')), x.t('p2')]}, {})
apply signature x.t before [('AsyncResult', '749d13d6-1caf-47c9-ab20-63dfd41ed369', 0, '8fb85d09-35ea-403b-a5f9-f82fc54ac211'), ('AsyncResult', '8fb85d09-35ea-403b-a5f9-f82fc54ac211', 0, None)]
apply signature x.t after [('AsyncResult', '749d13d6-1caf-47c9-ab20-63dfd41ed369', 0, '8fb85d09-35ea-403b-a5f9-f82fc54ac211'), ('AsyncResult', '8fb85d09-35ea-403b-a5f9-f82fc54ac211', 0, None)]
FINAL [('AsyncResult', '749d13d6-1caf-47c9-ab20-63dfd41ed369', 0, '8fb85d09-35ea-403b-a5f9-f82fc54ac211'), ('AsyncResult', '8fb85d09-35ea-403b-a5f9-f82fc54ac211', 0, None)]

473:    def freeze(self, _id=None, group_id=None, chord=None,
1096:    def freeze(self, _id=None, group_id=None, chord=None,
1890:    def freeze(self, _id=None, group_id=None, chord=None,
2094:    def freeze(self, _id=None, group_id=None, chord=None,
   520	        return self.AsyncResult(tid)
   521	
   522	    _freeze = freeze
   523	
   524	    def replace(self, args=None, kwargs=None, options=None):
   525	        """Replace the args, kwargs or options set for this signature.
   526	
   527	        These are only replaced if the argument for the section is

   528	        not :const:`None`.
   529	        """
   530	        signature = self.clone()
   531	        if args is not None:
   532	            signature.args = args
   533	        if kwargs is not None:
   534	            signature.kwargs = kwargs
   535	        if options is not None:
   536	            signature.options =
…[5134 chars omitted]…
something that will never occur for the frozen result.
  1089	            if not tasks:
  1090	                return result_from_apply
  1091	            else:
  1092	                return results_from_prepare[0]
  1093	
  1094	    # in order for a chain to be frozen, each of the members of the chain individually needs to be frozen
  1095	    # TODO figure out why we are always cloning before freeze

460	            args, kwargs, opts = self.args, self.kwargs, self.options
   461	        signature = Signature.from_dict({'task': self.task,
   462	                                         'args': tuple(args),
   463	                                         'kwargs': kwargs,
   464	                                         'options': deepcopy(opts),
   465	                                         'subtask_type': self.subtask_type,
   466	                                         'immutable': self.immutable},
   467	                                        app=self._app)
   468	        signature._type = self._type
   469	        return signature
   470	
   471	    partial = clone

   472	
   473	    def freeze(self, _id=None, group_id=None, chord=None,
   474	               root_id=None, parent_id=None, group_index=None):
   475	        """Finalize the signature by adding a concrete tas
…[4191 chars omitted]…
atter or GraphFormatter(root=self.id, shape='oval'),
   406	        )
   407	        for parent, node in self.iterdeps(intermediate=intermediate):
   408	            graph.add_arc(node)
   409	            if parent:
   410	                graph.add_edge(parent, node)
   411	        return graph
   412	
   413	    __class_getitem__ = classmethod(types.GenericAlias)
   414	
   415	    def __str__(self):

diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..aa5bc47 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -24,6 +24,7 @@ from kombu.utils.uuid import uuid
 from vine import barrier
 
 from celery._state import current_app
+from celery import states
 from celery.exceptions import CPendingDeprecationWarning
 from celery.result import GroupResult, allow_join_result
 from celery.utils import abstract

@@ -1296,6 +1297,8 @@ class _chain(Signature):
         kwargs = kwargs if kwargs else {}
         last, (fargs, fkwargs) = None, (args, kwargs)
         for task in self.tasks:
+            if last and last.state in (states.IGNORED, states.REJECTED):
+                break
             res = task.clone(fargs, fkwargs).apply(
                 last and (last.get(),), **dict(self.options, **options))
             res.parent, last, (fargs, fkwargs) = last, res, (None, None)
…[843 chars omitted]…
.as_tuple() for result in self.results],
         )
 
     @property
@@ -1119,7 +1122,7 @@ def result_from_tuple(r, app=None):
     if not isinstance(r, ResultBase):
         res, nodes = r
         id, parent = res if isinstance(res, (list, tuple)) else (res, None)
-        if parent:
+        if parent is not None:
             parent = result_from_tuple(parent, app)
 
         if nodes is not None:

Ignore IGNORED None ['stop']
Reject REJECTED None ['stop']
protocol 1 [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 3), ('AsyncResult', 0)]
protocol 2 [('AsyncResult', 0), ('GroupResult', 2), ('GroupResult', 3), ('AsyncResult', 0)]

............u........................................................... [ 22%]
....................................................x..x................ [ 45%]
............................uu.uu.......uuu.uuu.uuu.uuu.uuu.uuu.uuu.uuu. [ 59%]
........................................s............................... [ 82%]
........................................................                 [100%]
312 passed, 1 skipped, 2 xfailed, 3 warnings, 29 subtests passed in 6.42s

diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..aa5bc47 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -24,6 +24,7 @@ from kombu.utils.uuid import uuid
 from vine import barrier
 
 from celery._state import current_app
+from celery import states
 from celery.exceptions import CPendingDeprecationWarning
 from celery.result import GroupResult, allow_join_result
 from celery.utils import abstract

@@ -1296,6 +1297,8 @@ class _chain(Signature):
         kwargs = kwargs if kwargs else {}
         last, (fargs, fkwargs) = None, (args, kwargs)
         for task in self.tasks:
+            if last and last.state in (states.IGNORED, states.REJECTED):
+                break
             res = task.clone(fargs, fkwargs).apply(
                 last and (last.get(),), **dict(self.options, **options))
             res.parent, last, (fargs, fkwargs) = last, res, (None, None)
…[891 chars omitted]…
  )
 
     @property
@@ -1119,7 +1122,7 @@ def result_from_tuple(r, app=None):
     if not isinstance(r, ResultBase):
         res, nodes = r
         id, parent = res if isinstance(res, (list, tuple)) else (res, None)
-        if parent:
+        if parent is not None:
             parent = result_from_tuple(parent, app)
 
         if nodes is not None:
all reproducer checks passed for both protocols

Traceback (most recent call last):
  File "<stdin>", line 34, in <module>
AssertionError: []

bad 1 None 1 2 2 2 x.identity('before') | group(<_regen: [identity('a0'), identity('a1')]>) | group(<_regen: [identity('b0'), identity('b1')]>) | identity('after') | identity('after') ['Signature', 'group', 'group', 'Signature', 'Signature'] [('AsyncResult', 0), ('AsyncResult', 0)]

protocol 1 tasks [('Signature', 'x.t'), ('_chord', 'celery.chord'), ('Signature', 'x.t'), ('Signature', 'x.t')]
res 0 [('AsyncResult', 'c256442a-ad61-41dd-8532-c049e32d7480', 0, 128455853476608, 128455853488896), ('AsyncResult', '1b1370c6-8b2f-41c0-9f8a-9d47f1e01453', 0, 128455853488896, None)]
res 1 [('AsyncResult', '1b1370c6-8b2f-41c0-9f8a-9d47f1e01453', 0, 128455850050816, 128455850050528), ('GroupResult', '1205d391-7521-42c9-88d1-1756fb34b54f', 2, 128455850050528, 128455850050048), ('GroupResult', 'd42f81b6-b994-4f0b-9ea8-ca84ce597cc7', 2, 128455850050048, 128455854876352), ('AsyncResult', '15dccc96-b8ba-468c-9b94-fe44efe179a4', 0, 128455854876352, 128455870095792), ('AsyncResult', '39798025-d017-4498-8fcc-d479fdeffda5', 0, 128455870095792, None)]
res 2 [('AsyncResult', '15dccc96-b8ba-468c-9b94-fe44efe179a4', 0, 128455854876352, 128455870095792), ('AsyncResult', '39798025-d017-44
…[1286 chars omitted]…
-7639-47aa-ac2e-9f698ae33e6b', 0, 128455850053888, None)]
res 3 [('AsyncResult', '3093d240-7639-47aa-ac2e-9f698ae33e6b', 0, 128455850053888, None)]
identities?
0 parent is result1 False ids 128455850054464 128455850056288
res1 parent chain
128455850053360 AsyncResult c0fd0184-349f-4076-8cbd-818a055ce1a5 parent 128455850054464
128455850054464 AsyncResult a54a7787-26ad-4f70-a7ca-71e1b281badb parent None

1 x.t('p1') | t('p2') | group((t('a1'), t('a2'))) | group((t('b1'), t('b2'))) | t('q1') | t('q2') ['Signature', 'Signature', 'group', 'group', 'Signature', 'Signature'] [('AsyncResult', 0), ('AsyncResult', 0)] (('bd605f06-1182-4b70-8e39-e64c758530b3', (('5ed4b79d-7eb1-4b50-a2ab-013c62a6263d', None), None)), None)
rest [('AsyncResult', 0), ('AsyncResult', 0)]
2 x.t('p1') | t('p2') | group((t('a1'), t('a2'))) | group((t('b1'), t('b2'))) | t('q1') | t('q2') ['Signature', 'Signature', 'group', 'group', 'Signature', 'Signature'] [('AsyncResult', 0), ('AsyncResult', 0)] (('5583ae09-c90e-4606-a918-1f868629c129', (('d4cc6c08-dca0-4ab9-b1a1-d5db46a94522', None), None)), None)
rest [('AsyncResult', 0), ('AsyncResult', 0)]