import copy
import re
from contextlib import contextmanager
from unittest.mock import ANY, MagicMock, Mock, call, patch, sentinel

import pytest
from kombu.serialization import prepare_accept_content
from kombu.utils.encoding import bytes_to_str, ensure_bytes

import celery
from celery import chord, group, signature, states, uuid
from celery.app.task import Context, Task
from celery.backends.base import (BaseBackend, DisabledBackend, KeyValueStoreBackend, _create_chord_error_with_cause,
                                  _create_fake_task_request, _nulldict)
from celery.exceptions import BackendGetMetaError, BackendStoreError, ChordError, SecurityError, TimeoutError
from celery.result import GroupResult, result_from_tuple
from celery.utils import serialization
from celery.utils.functional import pass1
from celery.utils.serialization import UnpickleableExceptionWrapper
from celery.utils.serialization import find_pickleable_exception as fnpe
from celery.utils.serialization import get_pickleable_exception as gpe
from celery.utils.serialization import subclass_exception


class wrapobject:

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


class paramexception(Exception):

    def __init__(self, param):
        self.param = param


class objectexception:
    class Nested(Exception):
        pass


Oldstyle = None

Unpickleable = subclass_exception(
    'Unpickleable', KeyError, 'foo.module',
)
Impossible = subclass_exception(
    'Impossible', object, 'foo.module',
)
Lookalike = subclass_exception(
    'Lookalike', wrapobject, 'foo.module',
)


class test_nulldict:

    def test_nulldict(self):
        x = _nulldict()
        x['foo'] = 1
        x.update(foo=1, bar=2)
        x.setdefault('foo', 3)


class test_serialization:

    def test_create_exception_cls(self):
        assert serialization.create_exception_cls('FooError', 'm')
        assert serialization.create_exception_cls('FooError', 'm', KeyError)


class test_Backend_interface:

    def setup_method(self):
        self.app.conf.accept_content = ['json']

    def test_accept_precedence(self):

        # default is app.conf.accept_content
        accept_content = self.app.conf.accept_content
        b1 = BaseBackend(self.app)
        assert prepare_accept_content(accept_content) == b1.accept

        # accept parameter
        b2 = BaseBackend(self.app, accept=['yaml'])
        assert len(b2.accept) == 1
        assert list(b2.accept)[0] == 'application/x-yaml'
        assert prepare_accept_content(['yaml']) == b2.accept

        # accept parameter over result_accept_content
        self.app.conf.result_accept_content = ['json']
        b3 = BaseBackend(self.app, accept=['yaml'])
        assert len(b3.accept) == 1
        assert list(b3.accept)[0] == 'application/x-yaml'
        assert prepare_accept_content(['yaml']) == b3.accept

        # conf.result_accept_content if specified
        self.app.conf.result_accept_content = ['yaml']
        b4 = BaseBackend(self.app)
        assert len(b4.accept) == 1
        assert list(b4.accept)[0] == 'application/x-yaml'
        assert prepare_accept_content(['yaml']) == b4.accept

    def test_get_result_meta(self):
        b1 = BaseBackend(self.app)
        meta = b1._get_result_meta(result={'fizz': 'buzz'},
                                   state=states.SUCCESS, traceback=None,
                                   request=None)
        assert meta['status'] == states.SUCCESS
        assert meta['result'] == {'fizz': 'buzz'}
        assert meta['traceback'] is None

        self.app.conf.result_extended = True
        args = ['a', 'b']
        kwargs = {'foo': 'bar'}
        task_name = 'mytask'

        b2 = BaseBackend(self.app)
        request = Context(args=args, kwargs=kwargs,
                          task=task_name,
                          delivery_info={'routing_key': 'celery'})
        meta = b2._get_result_meta(result={'fizz': 'buzz'},
                                   state=states.SUCCESS, traceback=None,
                                   request=request, encode=False)
        assert meta['name'] == task_name
        assert meta['args'] == args
        assert meta['kwargs'] == kwargs
        assert meta['queue'] == 'celery'

    def test_get_result_meta_stamps_attribute_error(self):
        class Request:
            pass
        self.app.conf.result_extended = True
        b1 = BaseBackend(self.app)
        meta = b1._get_result_meta(result={'fizz': 'buzz'},
                                   state=states.SUCCESS, traceback=None,
                                   request=Request())
        assert meta['status'] == states.SUCCESS
        assert meta['result'] == {'fizz': 'buzz'}
        assert meta['traceback'] is None

    def test_get_result_meta_encoded(self):
        self.app.conf.result_extended = True
        b1 = BaseBackend(self.app)
        args = ['a', 'b']
        kwargs = {'foo': 'bar'}

        request = Context(args=args, kwargs=kwargs)
        meta = b1._get_result_meta(result={'fizz': 'buzz'},
                                   state=states.SUCCESS, traceback=None,
                                   request=request, encode=True)
        assert meta['args'] == ensure_bytes(b1.encode(args))
        assert meta['kwargs'] == ensure_bytes(b1.encode(kwargs))

    def test_get_result_meta_with_none(self):
        b1 = BaseBackend(self.app)
        meta = b1._get_result_meta(result=None,
                                   state=states.SUCCESS, traceback=None,
                                   request=None)
        assert meta['status'] == states.SUCCESS
        assert meta['result'] is None
        assert meta['traceback'] is None

        self.app.conf.result_extended = True
        args = ['a', 'b']
        kwargs = {'foo': 'bar'}
        task_name = 'mytask'

        b2 = BaseBackend(self.app)
        request = Context(args=args, kwargs=kwargs,
                          task=task_name,
                          delivery_info={'routing_key': 'celery'})
        meta = b2._get_result_meta(result=None,
                                   state=states.SUCCESS, traceback=None,
                                   request=request, encode=False)
        assert meta['name'] == task_name
        assert meta['args'] == args
        assert meta['kwargs'] == kwargs
        assert meta['queue'] == 'celery'

    def test_get_result_meta_format_date(self):
        import datetime
        self.app.conf.result_extended = True
        b1 = BaseBackend(self.app)
        args = ['a', 'b']
        kwargs = {'foo': 'bar'}

        request = Context(args=args, kwargs=kwargs)
        meta = b1._get_result_meta(result={'fizz': 'buzz'},
                                   state=states.SUCCESS, traceback=None,
                                   request=request, format_date=True)
        assert isinstance(meta['date_done'], str)

        self.app.conf.result_extended = True
        b2 = BaseBackend(self.app)
        args = ['a', 'b']
        kwargs = {'foo': 'bar'}

        request = Context(args=args, kwargs=kwargs)
        meta = b2._get_result_meta(result={'fizz': 'buzz'},
                                   state=states.SUCCESS, traceback=None,
                                   request=request, format_date=False)
        assert isinstance(meta['date_done'], datetime.datetime)


class test_BaseBackend_interface:

    def setup_method(self):
        self.b = BaseBackend(self.app)

        @self.app.task(shared=False)
        def callback(result):
            pass

        self.callback = callback

    def test__forget(self):
        with pytest.raises(NotImplementedError):
            self.b._forget('SOMExx-N0Nex1stant-IDxx-')

    def test_forget(self):
        with pytest.raises(NotImplementedError):
            self.b.forget('SOMExx-N0nex1stant-IDxx-')

    def test_on_chord_part_return(self):
        self.b.on_chord_part_return(None, None, None)

    def test_apply_chord(self, unlock='celery.chord_unlock'):
        self.app.tasks[unlock] = Mock()
        header_result_args = (
            uuid(),
            [self.app.AsyncResult(x) for x in range(3)],
        )
        self.b.apply_chord(header_result_args, self.callback.s())
        assert self.app.tasks[unlock].apply_async.call_count

    def test_chord_unlock_queue(self, unlock='celery.chord_unlock'):
        self.app.tasks[unlock] = Mock()
        header_result_args = (
            uuid(),
            [self.app.AsyncResult(x) for x in range(3)],
        )
        body = self.callback.s()

        self.b.apply_chord(header_result_args, body)
        called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
        assert called_kwargs['queue'] == 'testcelery'

        routing_queue = Mock()
        routing_queue.name = "routing_queue"
        self.app.amqp.router.route = Mock(return_value={
            "queue": routing_queue
        })
        self.b.apply_chord(header_result_args, body)
        assert self.app.amqp.router.route.call_args[0][1] == body.name
        called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
        assert called_kwargs["queue"] == "routing_queue"

        self.b.apply_chord(header_result_args, body.set(queue='test_queue'))
        called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
        assert called_kwargs['queue'] == 'test_queue'

        @self.app.task(shared=False, queue='test_queue_two')
        def callback_queue(result):
            pass

        self.b.apply_chord(header_result_args, callback_queue.s())
        called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
        assert called_kwargs['queue'] == 'test_queue_two'

        with self.Celery() as app2:
            @app2.task(name='callback_different_app', shared=False)
            def callback_different_app(result):
                pass

            callback_different_app_signature = self.app.signature('callback_different_app')
            self.b.apply_chord(header_result_args, callback_different_app_signature)
            called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
            assert called_kwargs['queue'] == 'routing_queue'

            callback_different_app_signature.set(queue='test_queue_three')
            self.b.apply_chord(header_result_args, callback_different_app_signature)
            called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
            assert called_kwargs['queue'] == 'test_queue_three'


class test_exception_pickle:
    def test_BaseException(self):
        assert fnpe(Exception()) is None

    def test_get_pickleable_exception(self):
        exc = Exception('foo')
        assert gpe(exc) == exc

    def test_unpickleable(self):
        assert isinstance(fnpe(Unpickleable()), KeyError)
        assert fnpe(Impossible()) is None


class test_prepare_exception:

    def setup_method(self):
        self.b = BaseBackend(self.app)

    def test_unpickleable(self):
        self.b.serializer = 'pickle'
        x = self.b.prepare_exception(Unpickleable(1, 2, 'foo'))
        assert isinstance(x, KeyError)
        y = self.b.exception_to_python(x)
        assert isinstance(y, KeyError)

    def test_json_exception_arguments(self):
        self.b.serializer = 'json'
        x = self.b.prepare_exception(Exception(object))
        assert x == {
            'exc_message': serialization.ensure_serializable(
                (object,), self.b.encode),
            'exc_type': Exception.__name__,
            'exc_module': Exception.__module__}
        y = self.b.exception_to_python(x)
        assert isinstance(y, Exception)

    def test_json_exception_nested(self):
        self.b.serializer = 'json'
        x = self.b.prepare_exception(objectexception.Nested('msg'))
        assert x == {
            'exc_message': ('msg',),
            'exc_type': 'objectexception.Nested',
            'exc_module': objectexception.Nested.__module__}
        y = self.b.exception_to_python(x)
        assert isinstance(y, objectexception.Nested)

    def test_impossible(self):
        self.b.serializer = 'pickle'
        x = self.b.prepare_exception(Impossible())
        assert isinstance(x, UnpickleableExceptionWrapper)
        assert str(x)
        y = self.b.exception_to_python(x)
        assert y.__class__.__name__ == 'Impossible'
        assert y.__class__.__module__ == 'foo.module'

    def test_regular(self):
        self.b.serializer = 'pickle'
        x = self.b.prepare_exception(KeyError('baz'))
        assert isinstance(x, KeyError)
        y = self.b.exception_to_python(x)
        assert isinstance(y, KeyError)

    def test_unicode_message(self):
        message = '\u03ac'
        x = self.b.prepare_exception(Exception(message))
        assert x == {'exc_message': (message,),
                     'exc_type': Exception.__name__,
                     'exc_module': Exception.__module__}


class KVBackend(KeyValueStoreBackend):
    mget_returns_dict = False

    def __init__(self, app, *args, **kwargs):
        self.db = {}
        super().__init__(app, *args, **kwargs)

    def get(self, key):
        return self.db.get(key)

    def _set_with_state(self, key, value, state):
        self.db[key] = value

    def mget(self, keys):
        if self.mget_returns_dict:
            return {key: self.get(key) for key in keys}
        else:
            return [self.get(k) for k in keys]

    def delete(self, key):
        self.db.pop(key, None)


class DictBackend(BaseBackend):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._data = {'can-delete': {'result': 'foo'}}

    def _restore_group(self, group_id):
        if group_id == 'exists':
            return {'result': 'group'}

    def _get_task_meta_for(self, task_id):
        if task_id == 'task-exists':
            return {'result': 'task'}

    def _delete_group(self, group_id):
        self._data.pop(group_id, None)


class test_BaseBackend_dict:

    def setup_method(self):
        self.b = DictBackend(app=self.app)

        @self.app.task(shared=False, bind=True)
        def bound_errback(self, result):
            pass

        @self.app.task(shared=False)
        def errback(arg1, arg2):
            errback.last_result = arg1 + arg2

        self.bound_errback = bound_errback
        self.errback = errback

    def test_delete_group(self):
        self.b.delete_group('can-delete')
        assert 'can-delete' not in self.b._data

    def test_prepare_exception_json(self):
        x = DictBackend(self.app, serializer='json')
        e = x.prepare_exception(KeyError('foo'))
        assert 'exc_type' in e
        e = x.exception_to_python(e)
        assert e.__class__.__name__ == 'KeyError'
        assert str(e).strip('u') == "'foo'"

    def test_save_group(self):
        b = BaseBackend(self.app)
        b._save_group = Mock()
        b.save_group('foofoo', 'xxx')
        b._save_group.assert_called_with('foofoo', 'xxx')

    def test_add_to_chord_interface(self):
        b = BaseBackend(self.app)
        with pytest.raises(NotImplementedError):
            b.add_to_chord('group_id', 'sig')

    def test_forget_interface(self):
        b = BaseBackend(self.app)
        with pytest.raises(NotImplementedError):
            b.forget('foo')

    def test_restore_group(self):
        assert self.b.restore_group('missing') is None
        assert self.b.restore_group('missing') is None
        assert self.b.restore_group('exists') == 'group'
        assert self.b.restore_group('exists') == 'group'
        assert self.b.restore_group('exists', cache=False) == 'group'

    def test_reload_group_result(self):
        self.b._cache = {}
        self.b.reload_group_result('exists')
        self.b._cache['exists'] = {'result': 'group'}

    def test_reload_task_result(self):
        self.b._cache = {}
        self.b.reload_task_result('task-exists')
        self.b._cache['task-exists'] = {'result': 'task'}

    def test_fail_from_current_stack(self):
        import inspect
        self.b.mark_as_failure = Mock()
        frame_list = []

        def raise_dummy():
            frame_str_temp = str(inspect.currentframe().__repr__)
            frame_list.append(frame_str_temp)
            raise KeyError('foo')
        try:
            raise_dummy()
        except KeyError as exc:
            self.b.fail_from_current_stack('task_id')
            self.b.mark_as_failure.assert_called()
            args = self.b.mark_as_failure.call_args[0]
            assert args[0] == 'task_id'
            assert args[1] is exc
            assert args[2]

            tb_ = exc.__traceback__
            while tb_ is not None:
                if str(tb_.tb_frame.__repr__) == frame_list[0]:
                    assert len(tb_.tb_frame.f_locals) == 0
                tb_ = tb_.tb_next

    def test_prepare_value_serializes_group_result(self):
        self.b.serializer = 'json'
        g = self.app.GroupResult('group_id', [self.app.AsyncResult('foo')])
        v = self.b.prepare_value(g)
        assert isinstance(v, (list, tuple))
        assert result_from_tuple(v, app=self.app) == g

        v2 = self.b.prepare_value(g[0])
        assert isinstance(v2, (list, tuple))
        assert result_from_tuple(v2, app=self.app) == g[0]

        self.b.serializer = 'pickle'
        assert isinstance(self.b.prepare_value(g), self.app.GroupResult)

    def test_is_cached(self):
        b = BaseBackend(app=self.app, max_cached_results=1)
        b._cache['foo'] = 1
        assert b.is_cached('foo')
        assert not b.is_cached('false')

    def test_mark_as_done__chord(self):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()
        request = Mock(name='request')
        b.on_chord_part_return = Mock()
        b.mark_as_done('id', 10, request=request)
        b.on_chord_part_return.assert_called_with(request, states.SUCCESS, 10)

    def test_mark_as_failure__bound_errback_eager(self):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()
        request = Mock(name='request')
        request.delivery_info = {
            'is_eager': True
        }
        request.errbacks = [
            self.bound_errback.subtask(args=[1], immutable=True)]
        exc = KeyError()
        group = self.patching('celery.backends.base.group')
        b.mark_as_failure('id', exc, request=request)
        group.assert_called_with(request.errbacks, app=self.app)
        group.return_value.apply.assert_called_with(
            (request.id, ), parent_id=request.id, root_id=request.root_id)

    def test_mark_as_failure__bound_errback(self):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()
        request = Mock(name='request')
        request.delivery_info = {}
        request.errbacks = [
            self.bound_errback.subtask(args=[1], immutable=True)]
        exc = KeyError()
        group = self.patching('celery.backends.base.group')
        b.mark_as_failure('id', exc, request=request)
        group.assert_called_with(request.errbacks, app=self.app)
        group.return_value.apply_async.assert_called_with(
            (request.id, ), parent_id=request.id, root_id=request.root_id)

    def test_mark_as_failure__errback(self):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()
        request = Mock(name='request')
        request.errbacks = [self.errback.subtask(args=[2, 3], immutable=True)]
        exc = KeyError()
        b.mark_as_failure('id', exc, request=request)
        assert self.errback.last_result == 5

    @patch('celery.backends.base.group')
    def test_class_based_task_can_be_used_as_error_callback(self, mock_group):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()

        class TaskBasedClass(Task):
            def run(self):
                pass

        TaskBasedClass = self.app.register_task(TaskBasedClass())

        request = Mock(name='request')
        request.errbacks = [TaskBasedClass.subtask(args=[], immutable=True)]
        exc = KeyError()
        b.mark_as_failure('id', exc, request=request)
        mock_group.assert_called_once_with(request.errbacks, app=self.app)

    @patch('celery.backends.base.group')
    def test_unregistered_task_can_be_used_as_error_callback(self, mock_group):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()

        request = Mock(name='request')
        request.errbacks = [signature('doesnotexist',
                                      immutable=True)]
        exc = KeyError()
        b.mark_as_failure('id', exc, request=request)
        mock_group.assert_called_once_with(request.errbacks, app=self.app)

    def test_mark_as_failure__chord(self):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()
        request = Mock(name='request')
        request.errbacks = []
        b.on_chord_part_return = Mock()
        exc = KeyError()
        b.mark_as_failure('id', exc, request=request)
        b.on_chord_part_return.assert_called_with(request, states.FAILURE, exc)

    def test_mark_as_revoked__chord(self):
        b = BaseBackend(app=self.app)
        b._store_result = Mock()
        request = Mock(name='request')
        request.errbacks = []
        b.on_chord_part_return = Mock()
        b.mark_as_revoked('id', 'revoked', request=request)
        b.on_chord_part_return.assert_called_with(request, states.REVOKED, ANY)

    def test_chord_error_from_stack_raises(self):
        class ExpectedException(Exception):
            pass

        b = BaseBackend(app=self.app)
        callback = MagicMock(name='callback')
        callback.options = {'link_error': []}
        callback.keys.return_value = []
        task = self.app.tasks[callback.task] = Mock()
        b.fail_from_current_stack = Mock()
        self.patching('celery.group')
        with patch.object(
            b, "_call_task_errbacks", side_effect=ExpectedException()
        ) as mock_call_errbacks:
            b.chord_error_from_stack(callback, exc=ValueError())
        task.backend.fail_from_current_stack.assert_called_with(
            callback.id, exc=mock_call_errbacks.side_effect,
        )

    def test_exception_to_python_when_None(self):
        b = BaseBackend(app=self.app)
        assert b.exception_to_python(None) is None

    def test_not_an_actual_exc_info(self):
        pass

    def test_not_an_exception_but_a_callable(self):
        x = {
            'exc_message': ('echo 1',),
            'exc_type': 'system',
            'exc_module': 'os'
        }

        with pytest.raises(SecurityError,
                           match=re.escape(r"Expected an exception class, got os.system with payload ('echo 1',)")):
            self.b.exception_to_python(x)

    def test_not_an_exception_but_another_object(self):
        x = {
            'exc_message': (),
            'exc_type': 'object',
            'exc_module': 'builtins'
        }

        with pytest.raises(SecurityError,
                           match=re.escape(r"Expected an exception class, got builtins.object with payload ()")):
            self.b.exception_to_python(x)

    def test_exception_to_python_when_attribute_exception(self):
        b = BaseBackend(app=self.app)
        test_exception = {'exc_type': 'AttributeDoesNotExist',
                          'exc_module': 'celery',
                          'exc_message': ['Raise Custom Message']}

        result_exc = b.exception_to_python(test_exception)
        assert str(result_exc) == 'Raise Custom Message'

    def test_exception_to_python_when_type_error(self):
        b = BaseBackend(app=self.app)
        celery.TestParamException = paramexception
        test_exception = {'exc_type': 'TestParamException',
                          'exc_module': 'celery',
                          'exc_message': []}

        result_exc = b.exception_to_python(test_exception)
        del celery.TestParamException
        assert str(result_exc) == "<class 't.unit.backends.test_base.paramexception'>([])"

    def test_wait_for__on_interval(self):
        self.patching('time.sleep')
        b = BaseBackend(app=self.app)
        b._get_task_meta_for = Mock()
        b._get_task_meta_for.return_value = {'status': states.PENDING}
        callback = Mock(name='callback')
        with pytest.raises(TimeoutError):
            b.wait_for(task_id='1', on_interval=callback, timeout=1)
        callback.assert_called_with()

        b._get_task_meta_for.return_value = {'status': states.SUCCESS}
        b.wait_for(task_id='1', timeout=None)

    def test_get_children(self):
        b = BaseBackend(app=self.app)
        b._get_task_meta_for = Mock()
        b._get_task_meta_for.return_value = {}
        assert b.get_children('id') is None
        b._get_task_meta_for.return_value = {'children': 3}
        assert b.get_children('id') == 3

    @pytest.mark.parametrize(
        "message,original_exc,expected_cause_behavior",
        [
            # With exception - should preserve original exception
            (
                "Dependency failed",
                ValueError("original error"),
                "has_cause",
            ),
            # Without exception (None) - should not have __cause__
            (
                "Dependency failed",
                None,
                "no_cause",
            ),
            # With non-exception - should not have __cause__
            (
                "Dependency failed",
                "not an exception",
                "no_cause",
            ),
        ],
        ids=(
            "with_exception",
            "without_exception",
            "with_non_exception",
        )
    )
    def test_create_chord_error_with_cause(
        self, message, original_exc, expected_cause_behavior
    ):
        """Test _create_chord_error_with_cause with various parameter combinations."""
        chord_error = _create_chord_error_with_cause(message, original_exc)

        # Verify basic ChordError properties
        assert isinstance(chord_error, ChordError)
        assert str(chord_error) == message

        # Verify __cause__ behavior based on test case
        if expected_cause_behavior == "has_cause":
            assert chord_error.__cause__ is original_exc
        elif expected_cause_behavior == "no_cause":
            assert not hasattr(chord_error, '__cause__') or chord_error.__cause__ is None

    @pytest.mark.parametrize(
        "task_id,errbacks,task_name,extra_kwargs,expected_attrs",
        [
            # Basic parameters test
            (
                "test-task-id",
                ["errback1", "errback2"],
                "test.task",
                {},
                {
                    "id": "test-task-id",
                    "errbacks": ["errback1", "errback2"],
                    "task": "test.task",
                    "delivery_info": {},
                },
            ),
            # Default parameters test
            (
                "test-task-id",
                None,
                None,
                {},
                {
                    "id": "test-task-id",
                    "errbacks": [],
                    "task": "unknown",
                    "delivery_info": {},
                },
            ),
            # Extra parameters test
            (
                "test-task-id",
                None,
                None,
                {"extra_param": "extra_value"},
                {
                    "id": "test-task-id",
                    "errbacks": [],
                    "task": "unknown",
                    "delivery_info": {},
                    "extra_param": "extra_value",
                },
            ),
        ],
        ids=(
            "basic_parameters",
            "default_parameters",
            "extra_parameters",
        )
    )
    def test_create_fake_task_request(
        self, task_id, errbacks, task_name, extra_kwargs, expected_attrs
    ):
        """Test _create_fake_task_request with various parameter combinations."""
        # Build call arguments
        args = [task_id]
        if errbacks is not None:
            args.append(errbacks)
        if task_name is not None:
            args.append(task_name)

        fake_request = _create_fake_task_request(*args, **extra_kwargs)

        # Verify all expected attributes
        for attr_name, expected_value in expected_attrs.items():
            assert getattr(fake_request, attr_name) == expected_value

    def _create_mock_callback(self, task_name="test.task", spec=None, **options):
        """Helper to create mock callbacks with common setup."""
        from collections.abc import Mapping

        # Create a mock that properly implements the
        # mapping protocol for PyPy env compatibility
        class MockCallback(Mock, Mapping):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self._mapping_data = {}

            def __getitem__(self, key):
                return self._mapping_data[key]

            def __iter__(self):
                return iter(self._mapping_data)

            def __len__(self):
                return len(self._mapping_data)

            def keys(self):
                return self._mapping_data.keys()

            def items(self):
                return self._mapping_data.items()

        callback = MockCallback(spec=spec)
        callback.task = task_name
        callback.options = {"link_error": [], **options}

        return callback

    def _setup_task_backend(self, task_name, backend=None):
        """Helper to set up task with backend in app registry."""
        if backend is None:
            backend = Mock()
            backend.fail_from_current_stack = Mock(return_value="backend_result")

        self.app.tasks[task_name] = Mock()
        self.app.tasks[task_name].backend = backend
        return backend

    @pytest.mark.parametrize(
        "callback_type,task_name,expected_group_handler_called",
        [
            ("group", "test.group.task", True),
            ("regular", "test.task", False),
        ],
        ids=["group_callback", "regular_callback"]
    )
    def test_chord_error_from_stack_callback_dispatch(self, callback_type, task_name, expected_group_handler_called):
        """Test chord_error_from_stack dispatches to correct handler based on callback type."""
        backend = self.b

        # Create callback based on type
        spec = group if callback_type == "group" else None
        callback = self._create_mock_callback(task_name, spec=spec)

        # Setup backend resolution
        mock_backend = self._setup_task_backend(task_name)

        # Mock handlers
        backend._handle_group_chord_error = Mock(return_value="group_result")
        backend._call_task_errbacks = Mock()

        exc = ValueError("test exception")
        result = backend.chord_error_from_stack(callback, exc)

        if expected_group_handler_called:
            backend._handle_group_chord_error.assert_called_once_with(
                group_callback=callback, backend=mock_backend, exc=exc
            )
            assert result == "group_result"
        else:
            mock_backend.fail_from_current_stack.assert_called_once()

    def test_chord_error_from_stack_backend_fallback(self):
        """Test chord_error_from_stack falls back to self when task not found."""
        backend = self.b

        callback = self._create_mock_callback("nonexistent.task")

        # Ensure task doesn't exist
        if "nonexistent.task" in self.app.tasks:
            del self.app.tasks["nonexistent.task"]

        backend._call_task_errbacks = Mock()
        backend.fail_from_current_stack = Mock(return_value="self_result")

        _ = backend.chord_error_from_stack(callback, ValueError("test"))

        # Verify self was used as fallback backend
        backend.fail_from_current_stack.assert_called_once()

    def _create_mock_frozen_group(self, group_id="group-id", task_ids=None, task_names=None):
        """Helper to create mock frozen group with results."""
        if task_ids is None:
            task_ids = ["task-id-1"]
        if task_names is None:
            task_names = ["test.task"] * len(task_ids)

        results = []
        for task_id, task_name in zip(task_ids, task_names):
            result = Mock()
            result.id = task_id
            result.task = task_name
            results.append(result)

        frozen_group = Mock(spec=GroupResult)
        frozen_group.results = results
        frozen_group.id = group_id
        frozen_group.revoke = Mock()
        return frozen_group

    def _setup_group_chord_error_test(self, exc=None, errbacks=None, task_ids=None):
        """Common setup for group chord error tests."""
        if exc is None:
            exc = ValueError("test error")
        if errbacks is None:
            errbacks = []
        if task_ids is None:
            task_ids = ["task-id-1"]

        backend = Mock()
        backend._call_task_errbacks = Mock()
        backend.fail_from_current_stack = Mock()
        backend.mark_as_failure = Mock()

        group_callback = Mock(spec=group)
        group_callback.options = {"link_error": errbacks}

        frozen_group = self._create_mock_frozen_group(task_ids=task_ids)
        group_callback.freeze.return_value = frozen_group

        return self.b, backend, group_callback, frozen_group, exc

    @pytest.mark.parametrize(
        "exception_setup,expected_exc_used",
        [
            ("with_cause", "original"),
            ("without_cause", "direct"),
        ],
        ids=["extracts_cause", "without_cause"]
    )
    def test_handle_group_chord_error_exception_handling(self, exception_setup, expected_exc_used):
        """Test _handle_group_chord_error handles exceptions with and without __cause__."""
        # Setup exceptions based on test case
        if exception_setup == "with_cause":
            original_exc = ValueError("original error")
            exc = ChordError("wrapped error")
            exc.__cause__ = original_exc
            expected_exc = original_exc
        else:
            exc = ValueError("direct error")
            expected_exc = exc

        b, backend, group_callback, frozen_group, _ = self._setup_group_chord_error_test(exc=exc)

        # Call the method
        _ = b._handle_group_chord_error(group_callback, backend, exc)

        # Verify correct exception was used
        backend.fail_from_current_stack.assert_called_with("task-id-1", exc=expected_exc)
        backend.mark_as_failure.assert_called_with("group-id", expected_exc)
        frozen_group.revoke.assert_called_once()

    def test_handle_group_chord_error_multiple_tasks(self):
        """Test _handle_group_chord_error handles multiple tasks in group."""
        task_ids = ["task-id-1", "task-id-2"]
        b, backend, group_callback, frozen_group, exc = self._setup_group_chord_error_test(task_ids=task_ids)

        # Call the method
        b._handle_group_chord_error(group_callback, backend, exc)

        # Verify group revocation and all tasks handled
        frozen_group.revoke.assert_called_once()
        assert backend.fail_from_current_stack.call_count == 2
        backend.fail_from_current_stack.assert_any_call("task-id-1", exc=exc)
        backend.fail_from_current_stack.assert_any_call("task-id-2", exc=exc)

    def test_handle_group_chord_error_with_errbacks(self):
        """Test _handle_group_chord_error calls error callbacks for each task."""
        errbacks = ["errback1", "errback2"]
        b, backend, group_callback, frozen_group, exc = self._setup_group_chord_error_test(errbacks=errbacks)

        # Call the method
        b._handle_group_chord_error(group_callback, backend, exc)

        # Verify error callbacks were called
        backend._call_task_errbacks.assert_called_once()
        call_args = backend._call_task_errbacks.call_args
        fake_request = call_args[0][0]

        # Verify fake request was created correctly
        assert fake_request.id == "task-id-1"
        assert fake_request.errbacks == errbacks
        assert fake_request.task == "test.task"

    def test_handle_group_chord_error_cleanup_exception_handling(self):
        """Test _handle_group_chord_error handles cleanup exceptions gracefully."""
        b = self.b
        backend = Mock()

        exc = ValueError("test error")

        # Mock group callback that raises exception during freeze
        group_callback = Mock(spec=group)
        group_callback.freeze.side_effect = RuntimeError("freeze failed")

        # Mock fallback behavior
        backend.fail_from_current_stack = Mock(return_value="fallback_result")

        # Should not raise exception, but return fallback result
        result = b._handle_group_chord_error(group_callback, backend, exc)

        # Verify fallback was called - the method returns an ExceptionInfo when cleanup fails
        # and falls back to single task handling
        assert result is not None  # Method returns ExceptionInfo from fail_from_current_stack

    def test_handle_group_chord__exceptions_paths(self, caplog):
        """Test _handle_group_chord handles exceptions in various paths."""
        backend = Mock()

        # Mock group callback
        group_callback = Mock(spec=group)
        group_callback.options = {"link_error": []}

        # Mock frozen group with multiple results
        mock_result1 = Mock()
        mock_result1.id = "task-id-1"
        mock_result2 = Mock()
        mock_result2.id = "task-id-2"

        frozen_group = Mock(spec=GroupResult)
        frozen_group.results = [mock_result1, mock_result2]
        frozen_group.revoke = Mock()

        group_callback.freeze.return_value = frozen_group

        # Test exception during fail_from_current_stack
        backend._call_task_errbacks.side_effect = RuntimeError("fail on _call_task_errbacks")

        backend.fail_from_current_stack.side_effect = RuntimeError("fail on fail_from_current_stack")

        _ = self.b._handle_group_chord_error(group_callback, backend, ValueError("test error"))

        assert "Failed to handle chord error for task" in caplog.text


class test_KeyValueStoreBackend:

    def setup_method(self):
        self.b = KVBackend(app=self.app)

    def test_on_chord_part_return(self):
        assert not self.b.implements_incr
        self.b.on_chord_part_return(None, None, None)

    def test_get_store_delete_result(self):
        tid = uuid()
        self.b.mark_as_done(tid, 'Hello world')
        assert self.b.get_result(tid) == 'Hello world'
        assert self.b.get_state(tid) == states.SUCCESS
        self.b.forget(tid)
        assert self.b.get_state(tid) == states.PENDING

    @pytest.mark.parametrize('serializer',
                             ['json', 'pickle', 'yaml', 'msgpack'])
    def test_store_result_parent_id(self, serializer):
        self.app.conf.accept_content = ('json', serializer)
        self.b = KVBackend(app=self.app, serializer=serializer)
        tid = uuid()
        pid = uuid()
        state = 'SUCCESS'
        result = 10
        request = Context(parent_id=pid)
        self.b.store_result(
            tid, state=state, result=result, request=request,
        )
        stored_meta = self.b.decode(self.b.get(self.b.get_key_for_task(tid)))
        assert stored_meta['parent_id'] == request.parent_id

    def test_store_result_group_id(self):
        tid = uuid()
        state = 'SUCCESS'
        result = 10
        request = Context(group='gid', children=[])
        self.b.store_result(
            tid, state=state, result=result, request=request,
        )
        stored_meta = self.b.decode(self.b.get(self.b.get_key_for_task(tid)))
        assert stored_meta['group_id'] == request.group

    def test_store_result_race_second_write_should_ignore_if_previous_success(self):
        tid = uuid()
        state = 'SUCCESS'
        result = 10
        request = Context(group='gid', children=[])
        self.b.store_result(
            tid, state=state, result=result, request=request,
        )
        self.b.store_result(
            tid, state=states.FAILURE, result=result, request=request,
        )
        stored_meta = self.b.decode(self.b.get(self.b.get_key_for_task(tid)))
        assert stored_meta['status'] == states.SUCCESS

    def test_get_key_for_task_none_task_id(self):
        with pytest.raises(ValueError):
            self.b.get_key_for_task(None)

    def test_get_key_for_group_none_group_id(self):
        with pytest.raises(ValueError):
            self.b.get_key_for_task(None)

    def test_get_key_for_chord_none_group_id(self):
        with pytest.raises(ValueError):
            self.b.get_key_for_group(None)

    def test_strip_prefix(self):
        x = self.b.get_key_for_task('x1b34')
        assert self.b._strip_prefix(x) == 'x1b34'
        assert self.b._strip_prefix('x1b34') == 'x1b34'

    def test_global_keyprefix(self):
        global_keyprefix = "test_global_keyprefix"
        app = copy.deepcopy(self.app)
        app.conf.get('result_backend_transport_options', {}).update({"global_keyprefix": global_keyprefix})
        b = KVBackend(app=app)
        tid = uuid()
        assert bytes_to_str(b.get_key_for_task(tid)) == f"{global_keyprefix}_celery-task-meta-{tid}"
        assert bytes_to_str(b.get_key_for_group(tid)) == f"{global_keyprefix}_celery-taskset-meta-{tid}"
        assert bytes_to_str(b.get_key_for_chord(tid)) == f"{global_keyprefix}_chord-unlock-{tid}"

        global_keyprefix = "test_global_keyprefix_"
        app = copy.deepcopy(self.app)
        app.conf.get('result_backend_transport_options', {}).update({"global_keyprefix": global_keyprefix})
        b = KVBackend(app=app)
        tid = uuid()
        assert bytes_to_str(b.get_key_for_task(tid)) == f"{global_keyprefix}celery-task-meta-{tid}"
        assert bytes_to_str(b.get_key_for_group(tid)) == f"{global_keyprefix}celery-taskset-meta-{tid}"
        assert bytes_to_str(b.get_key_for_chord(tid)) == f"{global_keyprefix}chord-unlock-{tid}"

        global_keyprefix = "test_global_keyprefix:"
        app = copy.deepcopy(self.app)
        app.conf.get('result_backend_transport_options', {}).update({"global_keyprefix": global_keyprefix})
        b = KVBackend(app=app)
        tid = uuid()
        assert bytes_to_str(b.get_key_for_task(tid)) == f"{global_keyprefix}celery-task-meta-{tid}"
        assert bytes_to_str(b.get_key_for_group(tid)) == f"{global_keyprefix}celery-taskset-meta-{tid}"
        assert bytes_to_str(b.get_key_for_chord(tid)) == f"{global_keyprefix}chord-unlock-{tid}"

    def test_global_keyprefix_missing(self):
        tid = uuid()
        assert bytes_to_str(self.b.get_key_for_task(tid)) == f"celery-task-meta-{tid}"
        assert bytes_to_str(self.b.get_key_for_group(tid)) == f"celery-taskset-meta-{tid}"
        assert bytes_to_str(self.b.get_key_for_chord(tid)) == f"chord-unlock-{tid}"

    def test_get_many(self):
        for is_dict in True, False:
            self.b.mget_returns_dict = is_dict
            ids = {uuid(): i for i in range(10)}
            for id, i in ids.items():
                self.b.mark_as_done(id, i)
            it = self.b.get_many(list(ids), interval=0.01)
            for i, (got_id, got_state) in enumerate(it):
                assert got_state['result'] == ids[got_id]
            assert i == 9
            assert list(self.b.get_many(list(ids), interval=0.01))

            self.b._cache.clear()
            callback = Mock(name='callback')
            it = self.b.get_many(
                list(ids),
                on_message=callback,
                interval=0.05
            )
            for i, (got_id, got_state) in enumerate(it):
                assert got_state['result'] == ids[got_id]
            assert i == 9
            assert list(
                self.b.get_many(list(ids), interval=0.01)
            )
            callback.assert_has_calls([
                call(ANY) for id in ids
            ])

    def test_get_many_times_out(self):
        tasks = [uuid() for _ in range(4)]
        self.b._cache[tasks[1]] = {'status': 'PENDING'}
        with pytest.raises(self.b.TimeoutError):
            list(self.b.get_many(tasks, timeout=0.01, interval=0.01))

    def test_get_many_passes_ready_states(self):
        tasks_length = 10
        ready_states = frozenset({states.SUCCESS})

        self.b._cache.clear()
        ids = {uuid(): i for i in range(tasks_length)}
        for id, i in ids.items():
            if i % 2 == 0:
                self.b.mark_as_done(id, i)
            else:
                self.b.mark_as_failure(id, Exception())

        it = self.b.get_many(list(ids), interval=0.01, max_iterations=1, READY_STATES=ready_states)
        it_list = list(it)

        assert all([got_state['status'] in ready_states for (got_id, got_state) in it_list])
        assert len(it_list) == tasks_length / 2

    def test_chord_part_return_no_gid(self):
        self.b.implements_incr = True
        task = Mock()
        state = 'SUCCESS'
        result = 10
        task.request.group = None
        self.b.get_key_for_chord = Mock()
        self.b.get_key_for_chord.side_effect = AssertionError(
            'should not get here',
        )
        assert self.b.on_chord_part_return(
            task.request, state, result) is None

    @patch('celery.backends.base.GroupResult')
    @patch('celery.backends.base.maybe_signature')
    def test_chord_part_return_restore_raises(self, maybe_signature,
                                              GroupResult):
        self.b.implements_incr = True
        GroupResult.restore.side_effect = KeyError()
        self.b.chord_error_from_stack = Mock()
        callback = Mock(name='callback')
        request = Mock(name='request')
        request.group = 'gid'
        maybe_signature.return_value = callback
        self.b.on_chord_part_return(request, states.SUCCESS, 10)
        self.b.chord_error_from_stack.assert_called_with(
            callback, ANY,
        )

    @patch('celery.backends.base.GroupResult')
    @patch('celery.backends.base.maybe_signature')
    def test_chord_part_return_restore_empty(self, maybe_signature,
                                             GroupResult):
        self.b.implements_incr = True
        GroupResult.restore.return_value = None
        self.b.chord_error_from_stack = Mock()
        callback = Mock(name='callback')
        request = Mock(name='request')
        request.group = 'gid'
        maybe_signature.return_value = callback
        self.b.on_chord_part_return(request, states.SUCCESS, 10)
        self.b.chord_error_from_stack.assert_called_with(
            callback, ANY,
        )

    def test_filter_ready(self):
        self.b.decode_result = Mock()
        self.b.decode_result.side_effect = pass1
        assert len(list(self.b._filter_ready([
            (1, {'status': states.RETRY}),
            (2, {'status': states.FAILURE}),
            (3, {'status': states.SUCCESS}),
        ]))) == 2

    @contextmanager
    def _chord_part_context(self, b):

        @self.app.task(shared=False)
        def callback(result):
            pass

        b.implements_incr = True
        b.client = Mock()
        with patch('celery.backends.base.GroupResult') as GR:
            deps = GR.restore.return_value = Mock(name='DEPS')
            deps.__len__ = Mock()
            deps.__len__.return_value = 10
            b.incr = Mock()
            b.incr.return_value = 10
            b.expire = Mock()
            task = Mock()
            task.request.group = 'grid'
            cb = task.request.chord = callback.s()
            task.request.chord.freeze()
            callback.backend = b
            callback.backend.fail_from_current_stack = Mock()
            yield task, deps, cb

    def test_chord_part_return_timeout(self):
        with self._chord_part_context(self.b) as (task, deps, _):
            try:
                self.app.conf.result_chord_join_timeout += 1.0
                self.b.on_chord_part_return(task.request, 'SUCCESS', 10)
            finally:
                self.app.conf.result_chord_join_timeout -= 1.0

            self.b.expire.assert_not_called()
            deps.delete.assert_called_with()
            deps.join_native.assert_called_with(propagate=True, timeout=4.0)

    def test_chord_part_return_propagate_set(self):
        with self._chord_part_context(self.b) as (task, deps, _):
            self.b.on_chord_part_return(task.request, 'SUCCESS', 10)
            self.b.expire.assert_not_called()
            deps.delete.assert_called_with()
            deps.join_native.assert_called_with(propagate=True, timeout=3.0)

    def test_chord_part_return_propagate_default(self):
        with self._chord_part_context(self.b) as (task, deps, _):
            self.b.on_chord_part_return(task.request, 'SUCCESS', 10)
            self.b.expire.assert_not_called()
            deps.delete.assert_called_with()
            deps.join_native.assert_called_with(propagate=True, timeout=3.0)

    def test_chord_part_return_join_raises_internal(self):
        with self._chord_part_context(self.b) as (task, deps, callback):
            deps._failed_join_report = lambda: iter([])
            deps.join_native.side_effect = KeyError('foo')
            self.b.on_chord_part_return(task.request, 'SUCCESS', 10)
            self.b.fail_from_current_stack.assert_called()
            args = self.b.fail_from_current_stack.call_args
            exc = args[1]['exc']
            assert isinstance(exc, ChordError)
            assert 'foo' in str(exc)

    def test_chord_part_return_join_raises_task(self):
        b = KVBackend(serializer='pickle', app=self.app)
        with self._chord_part_context(b) as (task, deps, callback):
            deps._failed_join_report = lambda: iter([
                self.app.AsyncResult('culprit'),
            ])
            deps.join_native.side_effect = KeyError('foo')
            b.on_chord_part_return(task.request, 'SUCCESS', 10)
            b.fail_from_current_stack.assert_called()
            args = b.fail_from_current_stack.call_args
            exc = args[1]['exc']
            assert isinstance(exc, ChordError)
            assert 'Dependency culprit raised' in str(exc)

    def test_restore_group_from_json(self):
        b = KVBackend(serializer='json', app=self.app)
        g = self.app.GroupResult(
            'group_id',
            [self.app.AsyncResult('a'), self.app.AsyncResult('b')],
        )
        b._save_group(g.id, g)
        g2 = b._restore_group(g.id)['result']
        assert g2 == g

    def test_restore_group_from_pickle(self):
        b = KVBackend(serializer='pickle', app=self.app)
        g = self.app.GroupResult(
            'group_id',
            [self.app.AsyncResult('a'), self.app.AsyncResult('b')],
        )
        b._save_group(g.id, g)
        g2 = b._restore_group(g.id)['result']
        assert g2 == g

    def test_chord_apply_fallback(self):
        self.b.implements_incr = False
        self.b.fallback_chord_unlock = Mock()
        header_result_args = (
            'group_id',
            [self.app.AsyncResult(x) for x in range(3)],
        )
        self.b.apply_chord(
            header_result_args, 'body', foo=1,
        )
        self.b.fallback_chord_unlock.assert_called_with(
            self.app.GroupResult(*header_result_args), 'body', foo=1,
        )

    def test_get_missing_meta(self):
        assert self.b.get_result('xxx-missing') is None
        assert self.b.get_state('xxx-missing') == states.PENDING

    def test_save_restore_delete_group(self):
        tid = uuid()
        tsr = self.app.GroupResult(
            tid, [self.app.AsyncResult(uuid()) for _ in range(10)],
        )
        self.b.save_group(tid, tsr)
        self.b.restore_group(tid)
        assert self.b.restore_group(tid) == tsr
        self.b.delete_group(tid)
        assert self.b.restore_group(tid) is None

    def test_restore_missing_group(self):
        assert self.b.restore_group('xxx-nonexistant') is None


class test_KeyValueStoreBackend_interface:

    def test_get(self):
        with pytest.raises(NotImplementedError):
            KeyValueStoreBackend(self.app).get('a')

    def test_set(self):
        with pytest.raises(NotImplementedError):
            KeyValueStoreBackend(self.app)._set_with_state('a', 1, states.SUCCESS)

    def test_incr(self):
        with pytest.raises(NotImplementedError):
            KeyValueStoreBackend(self.app).incr('a')

    def test_cleanup(self):
        assert not KeyValueStoreBackend(self.app).cleanup()

    def test_delete(self):
        with pytest.raises(NotImplementedError):
            KeyValueStoreBackend(self.app).delete('a')

    def test_mget(self):
        with pytest.raises(NotImplementedError):
            KeyValueStoreBackend(self.app).mget(['a'])

    def test_forget(self):
        with pytest.raises(NotImplementedError):
            KeyValueStoreBackend(self.app).forget('a')


class test_DisabledBackend:

    def test_store_result(self):
        DisabledBackend(self.app).store_result()

    def test_is_disabled(self):
        with pytest.raises(NotImplementedError):
            DisabledBackend(self.app).get_state('foo')

    def test_as_uri(self):
        assert DisabledBackend(self.app).as_uri() == 'disabled://'

    @pytest.mark.celery(result_backend='disabled')
    def test_chord_raises_error(self):
        with pytest.raises(NotImplementedError):
            chord(self.add.s(i, i) for i in range(10))(self.add.s([2]))

    @pytest.mark.celery(result_backend='disabled')
    def test_chain_with_chord_raises_error(self):
        with pytest.raises(NotImplementedError):
            (self.add.s(2, 2) |
             group(self.add.s(2, 2),
                   self.add.s(5, 6)) | self.add.s()).delay()


class test_as_uri:

    def setup_method(self):
        self.b = BaseBackend(
            app=self.app,
            url='sch://uuuu:pwpw@hostname.dom'
        )

    def test_as_uri_include_password(self):
        assert self.b.as_uri(True) == self.b.url

    def test_as_uri_exclude_password(self):
        assert self.b.as_uri() == 'sch://uuuu:**@hostname.dom/'


class test_backend_retries:

    def test_should_retry_exception(self):
        assert not BaseBackend(app=self.app).exception_safe_to_retry(Exception("test"))

    def test_get_failed_never_retries(self):
        self.app.conf.result_backend_always_retry, prev = False, self.app.conf.result_backend_always_retry

        expected_exc = Exception("failed")
        try:
            b = BaseBackend(app=self.app)
            b.exception_safe_to_retry = lambda exc: True
            b._sleep = Mock()
            b._get_task_meta_for = Mock()
            b._get_task_meta_for.side_effect = [
                expected_exc,
                {'status': states.SUCCESS, 'result': 42}
            ]
            try:
                b.get_task_meta(sentinel.task_id)
                assert False
            except Exception as exc:
                assert b._sleep.call_count == 0
                assert exc == expected_exc
        finally:
            self.app.conf.result_backend_always_retry = prev

    def test_get_with_retries(self):
        self.app.conf.result_backend_always_retry, prev = True, self.app.conf.result_backend_always_retry

        try:
            b = BaseBackend(app=self.app)
            b.exception_safe_to_retry = lambda exc: True
            b._sleep = Mock()
            b._get_task_meta_for = Mock()
            b._get_task_meta_for.side_effect = [
                Exception("failed"),
                {'status': states.SUCCESS, 'result': 42}
            ]
            res = b.get_task_meta(sentinel.task_id)
            assert res == {'status': states.SUCCESS, 'result': 42}
            assert b._sleep.call_count == 1
        finally:
            self.app.conf.result_backend_always_retry = prev

    def test_get_reaching_max_retries(self):
        self.app.conf.result_backend_always_retry, prev = True, self.app.conf.result_backend_always_retry
        self.app.conf.result_backend_max_retries, prev_max_retries = 0, self.app.conf.result_backend_max_retries

        try:
            b = BaseBackend(app=self.app)
            b.exception_safe_to_retry = lambda exc: True
            b._sleep = Mock()
            b._get_task_meta_for = Mock()
            b._get_task_meta_for.side_effect = [
                Exception("failed"),
                {'status': states.SUCCESS, 'result': 42}
            ]
            try:
                b.get_task_meta(sentinel.task_id)
                assert False
            except BackendGetMetaError:
                assert b._sleep.call_count == 0
        finally:
            self.app.conf.result_backend_always_retry = prev
            self.app.conf.result_backend_max_retries = prev_max_retries

    def test_get_unsafe_exception(self):
        self.app.conf.result_backend_always_retry, prev = True, self.app.conf.result_backend_always_retry

        expected_exc = Exception("failed")
        try:
            b = BaseBackend(app=self.app)
            b._sleep = Mock()
            b._get_task_meta_for = Mock()
            b._get_task_meta_for.side_effect = [
                expected_exc,
                {'status': states.SUCCESS, 'result': 42}
            ]
            try:
                b.get_task_meta(sentinel.task_id)
                assert False
            except Exception as exc:
                assert b._sleep.call_count == 0
                assert exc == expected_exc
        finally:
            self.app.conf.result_backend_always_retry = prev

    def test_store_result_never_retries(self):
        self.app.conf.result_backend_always_retry, prev = False, self.app.conf.result_backend_always_retry

        expected_exc = Exception("failed")
        try:
            b = BaseBackend(app=self.app)
            b.exception_safe_to_retry = lambda exc: True
            b._sleep = Mock()
            b._get_task_meta_for = Mock()
            b._get_task_meta_for.return_value = {
                'status': states.RETRY,
                'result': {
                    "exc_type": "Exception",
                    "exc_message": ["failed"],
                    "exc_module": "builtins",
                },
            }
            b._store_result = Mock()
            b._store_result.side_effect = [
                expected_exc,
                42
            ]
            try:
                b.store_result(sentinel.task_id, 42, states.SUCCESS)
            except Exception as exc:
                assert b._sleep.call_count == 0
                assert exc == expected_exc
        finally:
            self.app.conf.result_backend_always_retry = prev

    def test_store_result_with_retries(self):
        self.app.conf.result_backend_always_retry, prev = True, self.app.conf.result_backend_always_retry

        try:
            b = BaseBackend(app=self.app)
            b.exception_safe_to_retry = lambda exc: True
            b._sleep = Mock()
            b._get_task_meta_for = Mock()
            b._get_task_meta_for.return_value = {
                'status': states.RETRY,
                'result': {
                    "exc_type": "Exception",
                    "exc_message": ["failed"],
                    "exc_module": "builtins",
                },
            }
            b._store_result = Mock()
            b._store_result.side_effect = [
                Exception("failed"),
                42
            ]
            res = b.store_result(sentinel.task_id, 42, states.SUCCESS)
            assert res == 42
            assert b._sleep.call_count == 1
        finally:
            self.app.conf.result_backend_always_retry = prev

    def test_store_result_reaching_max_retries(self):
        self.app.conf.result_backend_always_retry, prev = True, self.app.conf.result_backend_always_retry
        self.app.conf.result_backend_max_retries, prev_max_retries = 0, self.app.conf.result_backend_max_retries

        try:
            b = BaseBackend(app=self.app)
            b.exception_safe_to_retry = lambda exc: True
            b._sleep = Mock()
            b._get_task_meta_for = Mock()
            b._get_task_meta_for.return_value = {
                'status': states.RETRY,
                'result': {
                    "exc_type": "Exception",
                    "exc_message": ["failed"],
                    "exc_module": "builtins",
                },
            }
            b._store_result = Mock()
            b._store_result.side_effect = [
                Exception("failed"),
                42
            ]
            try:
                b.store_result(sentinel.task_id, 42, states.SUCCESS)
                assert False
            except BackendStoreError:
                assert b._sleep.call_count == 0
        finally:
            self.app.conf.result_backend_always_retry = prev
            self.app.conf.result_backend_max_retries = prev_max_retries

    def test_result_backend_thread_safe(self):
        # Should identify the backend as thread safe
        self.app.conf.result_backend_thread_safe = True
        b = BaseBackend(app=self.app)
        assert b.thread_safe is True

    def test_result_backend_not_thread_safe(self):
        # Should identify the backend as not being thread safe
        self.app.conf.result_backend_thread_safe = False
        b = BaseBackend(app=self.app)
        assert b.thread_safe is False