import socket import tempfile from datetime import datetime, timedelta from unittest.mock import ANY, MagicMock, Mock, patch, sentinel import pytest from kombu import Queue from kombu.exceptions import EncodeError from celery import Task, chain, group, uuid from celery.app.task import _reprtask from celery.canvas import StampingVisitor, signature from celery.contrib.testing.mocks import ContextMock from celery.exceptions import Ignore, ImproperlyConfigured, Retry from celery.result import AsyncResult, EagerResult from celery.utils.serialization import UnpickleableExceptionWrapper try: from urllib.error import HTTPError except ImportError: from urllib2 import HTTPError def return_True(*args, **kwargs): # Task run functions can't be closures/lambdas, as they're pickled. return True class MockApplyTask(Task): abstract = True applied = 0 def run(self, x, y): return x * y def apply_async(self, *args, **kwargs): self.applied += 1 class TaskWithPriority(Task): priority = 10 class TaskWithRetry(Task): autoretry_for = (TypeError,) retry_kwargs = {'max_retries': 5} retry_backoff = True retry_backoff_max = 700 retry_jitter = False class TaskWithRetryButForTypeError(Task): autoretry_for = (Exception,) dont_autoretry_for = (TypeError,) retry_kwargs = {'max_retries': 5} retry_backoff = True retry_backoff_max = 700 retry_jitter = False class TasksCase: def setup_method(self): self.mytask = self.app.task(shared=False)(return_True) @self.app.task(bind=True, count=0, shared=False) def increment_counter(self, increment_by=1): self.count += increment_by or 1 return self.count self.increment_counter = increment_counter @self.app.task(shared=False) def raising(): raise KeyError('foo') self.raising = raising @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task(self, arg1, arg2, kwarg=1, max_retries=None, care=True): self.iterations += 1 rmax = self.max_retries if max_retries is None else max_retries assert repr(self.request) retries = self.request.retries if care and retries >= rmax: return arg1 else: raise self.retry(countdown=0, max_retries=rmax) self.retry_task = retry_task @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task_noargs(self, **kwargs): self.iterations += 1 if self.request.retries >= 3: return 42 else: raise self.retry(countdown=0) self.retry_task_noargs = retry_task_noargs @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task_return_without_throw(self, **kwargs): self.iterations += 1 try: if self.request.retries >= 3: return 42 else: raise Exception("random code exception") except Exception as exc: return self.retry(exc=exc, throw=False) self.retry_task_return_without_throw = retry_task_return_without_throw @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task_return_with_throw(self, **kwargs): self.iterations += 1 try: if self.request.retries >= 3: return 42 else: raise Exception("random code exception") except Exception as exc: return self.retry(exc=exc, throw=True) self.retry_task_return_with_throw = retry_task_return_with_throw @self.app.task(bind=True, max_retries=3, iterations=0, shared=False, autoretry_for=(Exception,)) def retry_task_auto_retry_with_single_new_arg(self, ret=None, **kwargs): if ret is None: return self.retry(exc=Exception("I have filled now"), args=["test"], kwargs=kwargs) else: return ret self.retry_task_auto_retry_with_single_new_arg = retry_task_auto_retry_with_single_new_arg @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task_auto_retry_with_new_args(self, ret=None, place_holder=None, **kwargs): if ret is None: return self.retry(args=[place_holder, place_holder], kwargs=kwargs) else: return ret self.retry_task_auto_retry_with_new_args = retry_task_auto_retry_with_new_args @self.app.task(bind=True, max_retries=3, iterations=0, shared=False, autoretry_for=(Exception,)) def retry_task_auto_retry_exception_with_new_args(self, ret=None, place_holder=None, **kwargs): if ret is None: return self.retry(exc=Exception("I have filled"), args=[place_holder, place_holder], kwargs=kwargs) else: return ret self.retry_task_auto_retry_exception_with_new_args = retry_task_auto_retry_exception_with_new_args @self.app.task(bind=True, max_retries=10, iterations=0, shared=False, autoretry_for=(Exception,)) def retry_task_max_retries_override(self, **kwargs): # Test for #6436 self.iterations += 1 if self.iterations == 3: # I wanna force fail here cause i have enough self.retry(exc=MyCustomException, max_retries=0) self.retry(exc=MyCustomException) self.retry_task_max_retries_override = retry_task_max_retries_override @self.app.task(bind=True, max_retries=0, iterations=0, shared=False, autoretry_for=(Exception,)) def retry_task_explicit_exception(self, **kwargs): # Test for #6436 self.iterations += 1 raise MyCustomException() self.retry_task_explicit_exception = retry_task_explicit_exception @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task_raise_without_throw(self, **kwargs): self.iterations += 1 try: if self.request.retries >= 3: return 42 else: raise Exception("random code exception") except Exception as exc: raise self.retry(exc=exc, throw=False) self.retry_task_raise_without_throw = retry_task_raise_without_throw @self.app.task(bind=True, max_retries=3, iterations=0, base=MockApplyTask, shared=False) def retry_task_mockapply(self, arg1, arg2, kwarg=1): self.iterations += 1 retries = self.request.retries if retries >= 3: return arg1 raise self.retry(countdown=0) self.retry_task_mockapply = retry_task_mockapply @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task_customexc(self, arg1, arg2, kwarg=1, **kwargs): self.iterations += 1 retries = self.request.retries if retries >= 3: return arg1 + kwarg else: try: raise MyCustomException('Elaine Marie Benes') except MyCustomException as exc: kwargs.update(kwarg=kwarg) raise self.retry(countdown=0, exc=exc) self.retry_task_customexc = retry_task_customexc @self.app.task(bind=True, max_retries=3, iterations=0, shared=False) def retry_task_unpickleable_exc(self, foo, bar): self.iterations += 1 raise self.retry(countdown=0, exc=UnpickleableException(foo, bar)) self.retry_task_unpickleable_exc = retry_task_unpickleable_exc @self.app.task(bind=True, autoretry_for=(ZeroDivisionError,), shared=False) def autoretry_task_no_kwargs(self, a, b): self.iterations += 1 return a / b self.autoretry_task_no_kwargs = autoretry_task_no_kwargs @self.app.task(bind=True, autoretry_for=(ZeroDivisionError,), retry_kwargs={'max_retries': 5}, shared=False) def autoretry_task(self, a, b): self.iterations += 1 return a / b self.autoretry_task = autoretry_task @self.app.task(bind=True, autoretry_for=(ArithmeticError,), dont_autoretry_for=(ZeroDivisionError,), retry_kwargs={'max_retries': 5}, shared=False) def autoretry_arith_task(self, a, b): self.iterations += 1 return a / b self.autoretry_arith_task = autoretry_arith_task @self.app.task(bind=True, base=TaskWithRetry, shared=False) def autoretry_for_from_base_task(self, a, b): self.iterations += 1 return a + b self.autoretry_for_from_base_task = autoretry_for_from_base_task @self.app.task(bind=True, base=TaskWithRetry, autoretry_for=(ZeroDivisionError,), shared=False) def override_autoretry_for_from_base_task(self, a, b): self.iterations += 1 return a / b self.override_autoretry_for = override_autoretry_for_from_base_task @self.app.task(bind=True, base=TaskWithRetry, shared=False) def retry_kwargs_from_base_task(self, a, b): self.iterations += 1 return a + b self.retry_kwargs_from_base_task = retry_kwargs_from_base_task @self.app.task(bind=True, base=TaskWithRetry, retry_kwargs={'max_retries': 2}, shared=False) def override_retry_kwargs_from_base_task(self, a, b): self.iterations += 1 return a + b self.override_retry_kwargs = override_retry_kwargs_from_base_task @self.app.task(bind=True, base=TaskWithRetry, shared=False) def retry_backoff_from_base_task(self, a, b): self.iterations += 1 return a + b self.retry_backoff_from_base_task = retry_backoff_from_base_task @self.app.task(bind=True, base=TaskWithRetry, retry_backoff=False, shared=False) def override_retry_backoff_from_base_task(self, a, b): self.iterations += 1 return a + b self.override_retry_backoff = override_retry_backoff_from_base_task @self.app.task(bind=True, base=TaskWithRetry, shared=False) def retry_backoff_max_from_base_task(self, a, b): self.iterations += 1 return a + b self.retry_backoff_max_from_base_task = retry_backoff_max_from_base_task @self.app.task(bind=True, base=TaskWithRetry, retry_backoff_max=16, shared=False) def override_retry_backoff_max_from_base_task(self, a, b): self.iterations += 1 return a + b self.override_backoff_max = override_retry_backoff_max_from_base_task @self.app.task(bind=True, base=TaskWithRetry, shared=False) def retry_backoff_jitter_from_base_task(self, a, b): self.iterations += 1 return a + b self.retry_backoff_jitter_from_base = retry_backoff_jitter_from_base_task @self.app.task(bind=True, base=TaskWithRetry, retry_jitter=True, shared=False) def override_backoff_jitter_from_base_task(self, a, b): self.iterations += 1 return a + b self.override_backoff_jitter = override_backoff_jitter_from_base_task @self.app.task(bind=True) def task_check_request_context(self): assert self.request.hostname == socket.gethostname() self.task_check_request_context = task_check_request_context @self.app.task(ignore_result=True) def task_with_ignored_result(): pass self.task_with_ignored_result = task_with_ignored_result @self.app.task(bind=True) def task_called_by_other_task(self): pass @self.app.task(bind=True) def task_which_calls_other_task(self): # Couldn't find a better way to mimic an apply_async() # request with set priority self.request.delivery_info['priority'] = 5 task_called_by_other_task.delay() self.task_which_calls_other_task = task_which_calls_other_task @self.app.task(bind=True) def task_replacing_another_task(self): return "replaced" self.task_replacing_another_task = task_replacing_another_task @self.app.task(bind=True) def task_replaced_by_other_task(self): return self.replace(task_replacing_another_task.si()) @self.app.task(bind=True, autoretry_for=(Exception,)) def task_replaced_by_other_task_with_autoretry(self): return self.replace(task_replacing_another_task.si()) self.task_replaced_by_other_task = task_replaced_by_other_task self.task_replaced_by_other_task_with_autoretry = task_replaced_by_other_task_with_autoretry # Remove all messages from memory-transport from kombu.transport.memory import Channel Channel.queues.clear() class MyCustomException(Exception): """Random custom exception.""" class UnpickleableException(Exception): """Exception that doesn't survive a pickling roundtrip (dump + load).""" def __init__(self, foo, bar): super().__init__(foo) self.bar = bar class test_task_retries(TasksCase): def test_retry(self): self.retry_task.max_retries = 3 self.retry_task.iterations = 0 self.retry_task.apply([0xFF, 0xFFFF]) assert self.retry_task.iterations == 4 self.retry_task.max_retries = 3 self.retry_task.iterations = 0 self.retry_task.apply([0xFF, 0xFFFF], {'max_retries': 10}) assert self.retry_task.iterations == 11 def test_retry_priority(self): priority = 7 # Technically, task.priority doesn't need to be set here # since push_request() doesn't populate the delivery_info # with it. However, setting task.priority here also doesn't # cause any problems. self.retry_task.priority = priority self.retry_task.push_request() self.retry_task.request.delivery_info = { 'priority': priority } sig = self.retry_task.signature_from_request() assert sig.options['priority'] == priority def test_retry_no_args(self): self.retry_task_noargs.max_retries = 3 self.retry_task_noargs.iterations = 0 self.retry_task_noargs.apply(propagate=True).get() assert self.retry_task_noargs.iterations == 4 def test_signature_from_request__passes_headers(self): self.retry_task.push_request() self.retry_task.request.headers = {'custom': 10.1} sig = self.retry_task.signature_from_request() assert sig.options['headers']['custom'] == 10.1 def test_signature_from_request__delivery_info(self): self.retry_task.push_request() self.retry_task.request.delivery_info = { 'exchange': 'testex', 'routing_key': 'testrk', } sig = self.retry_task.signature_from_request() assert sig.options['exchange'] == 'testex' assert sig.options['routing_key'] == 'testrk' def test_signature_from_request__shadow_name(self): self.retry_task.push_request() self.retry_task.request.shadow = 'test' sig = self.retry_task.signature_from_request() assert sig.options['shadow'] == 'test' def test_retry_kwargs_can_be_empty(self): self.retry_task_mockapply.push_request() try: with pytest.raises(Retry): import sys try: sys.exc_clear() except AttributeError: pass self.retry_task_mockapply.retry(args=[4, 4], kwargs=None) finally: self.retry_task_mockapply.pop_request() def test_retry_without_throw_eager(self): assert self.retry_task_return_without_throw.apply().get() == 42 def test_raise_without_throw_eager(self): assert self.retry_task_raise_without_throw.apply().get() == 42 def test_return_with_throw_eager(self): assert self.retry_task_return_with_throw.apply().get() == 42 def test_eager_retry_with_single_new_params(self): assert self.retry_task_auto_retry_with_single_new_arg.apply().get() == "test" def test_eager_retry_with_new_params(self): assert self.retry_task_auto_retry_with_new_args.si(place_holder="test").apply().get() == "test" def test_eager_retry_with_autoretry_for_exception(self): assert self.retry_task_auto_retry_exception_with_new_args.si(place_holder="test").apply().get() == "test" def test_retry_task_max_retries_override(self): self.retry_task_max_retries_override.max_retries = 10 self.retry_task_max_retries_override.iterations = 0 result = self.retry_task_max_retries_override.apply() with pytest.raises(MyCustomException): result.get() assert self.retry_task_max_retries_override.iterations == 3 def test_retry_task_explicit_exception(self): self.retry_task_explicit_exception.max_retries = 0 self.retry_task_explicit_exception.iterations = 0 result = self.retry_task_explicit_exception.apply() with pytest.raises(MyCustomException): result.get() assert self.retry_task_explicit_exception.iterations == 1 def test_retry_eager_should_return_value(self): self.retry_task.max_retries = 3 self.retry_task.iterations = 0 assert self.retry_task.apply([0xFF, 0xFFFF]).get() == 0xFF assert self.retry_task.iterations == 4 def test_retry_not_eager(self): self.retry_task_mockapply.push_request() try: self.retry_task_mockapply.request.called_directly = False exc = Exception('baz') try: self.retry_task_mockapply.retry( args=[4, 4], kwargs={'task_retries': 0}, exc=exc, throw=False, ) assert self.retry_task_mockapply.applied finally: self.retry_task_mockapply.applied = 0 try: with pytest.raises(Retry): self.retry_task_mockapply.retry( args=[4, 4], kwargs={'task_retries': 0}, exc=exc, throw=True) assert self.retry_task_mockapply.applied finally: self.retry_task_mockapply.applied = 0 finally: self.retry_task_mockapply.pop_request() def test_retry_with_kwargs(self): self.retry_task_customexc.max_retries = 3 self.retry_task_customexc.iterations = 0 self.retry_task_customexc.apply([0xFF, 0xFFFF], {'kwarg': 0xF}) assert self.retry_task_customexc.iterations == 4 def test_retry_with_custom_exception(self): self.retry_task_customexc.max_retries = 2 self.retry_task_customexc.iterations = 0 result = self.retry_task_customexc.apply( [0xFF, 0xFFFF], {'kwarg': 0xF}, ) with pytest.raises(MyCustomException): result.get() assert self.retry_task_customexc.iterations == 3 def test_retry_with_unpickleable_exception(self): self.retry_task_unpickleable_exc.max_retries = 2 self.retry_task_unpickleable_exc.iterations = 0 result = self.retry_task_unpickleable_exc.apply( ["foo", "bar"] ) with pytest.raises(UnpickleableExceptionWrapper) as exc_info: result.get() assert self.retry_task_unpickleable_exc.iterations == 3 exc_wrapper = exc_info.value assert exc_wrapper.exc_cls_name == "UnpickleableException" assert exc_wrapper.exc_args == ("foo", ) def test_max_retries_exceeded(self): self.retry_task.max_retries = 2 self.retry_task.iterations = 0 result = self.retry_task.apply([0xFF, 0xFFFF], {'care': False}) with pytest.raises(self.retry_task.MaxRetriesExceededError): result.get() assert self.retry_task.iterations == 3 self.retry_task.max_retries = 1 self.retry_task.iterations = 0 result = self.retry_task.apply([0xFF, 0xFFFF], {'care': False}) with pytest.raises(self.retry_task.MaxRetriesExceededError): result.get() assert self.retry_task.iterations == 2 def test_max_retries_exceeded_task_args(self): self.retry_task.max_retries = 2 self.retry_task.iterations = 0 args = (0xFF, 0xFFFF) kwargs = {'care': False} result = self.retry_task.apply(args, kwargs) with pytest.raises(self.retry_task.MaxRetriesExceededError) as e: result.get() assert e.value.task_args == args assert e.value.task_kwargs == kwargs def test_autoretry_no_kwargs(self): self.autoretry_task_no_kwargs.max_retries = 3 self.autoretry_task_no_kwargs.iterations = 0 self.autoretry_task_no_kwargs.apply((1, 0)) assert self.autoretry_task_no_kwargs.iterations == 4 def test_autoretry(self): self.autoretry_task.max_retries = 3 self.autoretry_task.iterations = 0 self.autoretry_task.apply((1, 0)) assert self.autoretry_task.iterations == 6 def test_autoretry_arith(self): self.autoretry_arith_task.max_retries = 3 self.autoretry_arith_task.iterations = 0 self.autoretry_arith_task.apply((1, 0)) assert self.autoretry_arith_task.iterations == 1 @pytest.mark.parametrize( 'retry_backoff, expected_countdowns', [ (False, [None, None, None, None]), (0, [None, None, None, None]), (0.0, [None, None, None, None]), (True, [1, 2, 4, 8]), (-1, [1, 2, 4, 8]), (0.1, [1, 2, 4, 8]), (1, [1, 2, 4, 8]), (1.9, [1, 2, 4, 8]), (2, [2, 4, 8, 16]), ], ) def test_autoretry_backoff(self, retry_backoff, expected_countdowns): @self.app.task(bind=True, shared=False, autoretry_for=(ZeroDivisionError,), retry_backoff=retry_backoff, retry_jitter=False, max_retries=3) def task(self_, x, y): self_.iterations += 1 return x / y task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply((1, 0)) assert task.iterations == 4 retry_call_countdowns = [ call_[1].get('countdown') for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == expected_countdowns @pytest.mark.parametrize( 'retry_backoff, expected_countdowns', [ (False, [None, None, None, None]), (0, [None, None, None, None]), (0.0, [None, None, None, None]), (True, [0, 1, 3, 7]), (-1, [0, 1, 3, 7]), (0.1, [0, 1, 3, 7]), (1, [0, 1, 3, 7]), (1.9, [0, 1, 3, 7]), (2, [1, 3, 7, 15]), ], ) @patch('random.randrange', side_effect=lambda i: i - 2) def test_autoretry_backoff_jitter(self, randrange, retry_backoff, expected_countdowns): @self.app.task(bind=True, shared=False, autoretry_for=(HTTPError,), retry_backoff=retry_backoff, retry_jitter=True, max_retries=3) def task(self_, url): self_.iterations += 1 if "error" in url: fp = tempfile.TemporaryFile() raise HTTPError(url, '500', 'Error', '', fp) task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply(("http://httpbin.org/error",)) assert task.iterations == 4 retry_call_countdowns = [ call_[1].get('countdown') for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == expected_countdowns def test_autoretry_for_from_base(self): self.autoretry_for_from_base_task.iterations = 0 self.autoretry_for_from_base_task.apply((1, "a")) assert self.autoretry_for_from_base_task.iterations == 6 def test_override_autoretry_for_from_base(self): self.override_autoretry_for.iterations = 0 self.override_autoretry_for.apply((1, 0)) assert self.override_autoretry_for.iterations == 6 def test_retry_kwargs_from_base(self): self.retry_kwargs_from_base_task.iterations = 0 self.retry_kwargs_from_base_task.apply((1, "a")) assert self.retry_kwargs_from_base_task.iterations == 6 def test_override_retry_kwargs_from_base(self): self.override_retry_kwargs.iterations = 0 self.override_retry_kwargs.apply((1, "a")) assert self.override_retry_kwargs.iterations == 3 def test_retry_backoff_from_base(self): task = self.retry_backoff_from_base_task task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply((1, "a")) assert task.iterations == 6 retry_call_countdowns = [ call_[1]['countdown'] for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == [1, 2, 4, 8, 16, 32] @patch('celery.app.autoretry.get_exponential_backoff_interval') def test_override_retry_backoff_from_base(self, backoff): self.override_retry_backoff.iterations = 0 self.override_retry_backoff.apply((1, "a")) assert self.override_retry_backoff.iterations == 6 assert backoff.call_count == 0 def test_retry_backoff_max_from_base(self): task = self.retry_backoff_max_from_base_task task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply((1, "a")) assert task.iterations == 6 retry_call_countdowns = [ call_[1]['countdown'] for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == [1, 2, 4, 8, 16, 32] def test_override_retry_backoff_max_from_base(self): task = self.override_backoff_max task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply((1, "a")) assert task.iterations == 6 retry_call_countdowns = [ call_[1]['countdown'] for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == [1, 2, 4, 8, 16, 16] def test_retry_backoff_jitter_from_base(self): task = self.retry_backoff_jitter_from_base task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply((1, "a")) assert task.iterations == 6 retry_call_countdowns = [ call_[1]['countdown'] for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == [1, 2, 4, 8, 16, 32] @patch('random.randrange', side_effect=lambda i: i - 2) def test_override_backoff_jitter_from_base(self, randrange): task = self.override_backoff_jitter task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply((1, "a")) assert task.iterations == 6 retry_call_countdowns = [ call_[1]['countdown'] for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == [0, 1, 3, 7, 15, 31] def test_retry_wrong_eta_when_not_enable_utc(self): """Issue #3753""" self.app.conf.enable_utc = False self.app.conf.timezone = 'US/Eastern' self.autoretry_task.iterations = 0 self.autoretry_task.default_retry_delay = 2 self.autoretry_task.apply((1, 0)) assert self.autoretry_task.iterations == 6 @pytest.mark.parametrize( 'backoff_value, expected_countdowns', [ (False, [None, None, None]), (0, [None, None, None]), (0.0, [None, None, None]), (True, [1, 2, 4]), (-1, [1, 2, 4]), (0.1, [1, 2, 4]), (1, [1, 2, 4]), (1.9, [1, 2, 4]), (2, [2, 4, 8]), ], ) def test_autoretry_class_based_task(self, backoff_value, expected_countdowns): class ClassBasedAutoRetryTask(Task): name = 'ClassBasedAutoRetryTask' autoretry_for = (ZeroDivisionError,) retry_kwargs = {'max_retries': 2} retry_backoff = backoff_value retry_backoff_max = 700 retry_jitter = False iterations = 0 _app = self.app def run(self, x, y): self.iterations += 1 return x / y task = ClassBasedAutoRetryTask() self.app.tasks.register(task) task.iterations = 0 with patch.object(task, 'retry', wraps=task.retry) as fake_retry: task.apply((1, 0)) assert task.iterations == 3 retry_call_countdowns = [ call_[1].get('countdown') for call_ in fake_retry.call_args_list ] assert retry_call_countdowns == expected_countdowns class test_canvas_utils(TasksCase): def test_si(self): assert self.retry_task.si() assert self.retry_task.si().immutable def test_chunks(self): assert self.retry_task.chunks(range(100), 10) def test_map(self): assert self.retry_task.map(range(100)) def test_starmap(self): assert self.retry_task.starmap(range(100)) def test_on_success(self): self.retry_task.on_success(1, 1, (), {}) class test_tasks(TasksCase): def now(self): return self.app.now() def test_typing(self): @self.app.task() def add(x, y, kw=1): pass with pytest.raises(TypeError): add.delay(1) with pytest.raises(TypeError): add.delay(1, kw=2) with pytest.raises(TypeError): add.delay(1, 2, foobar=3) add.delay(2, 2) def test_shadow_name(self): def shadow_name(task, args, kwargs, options): return 'fooxyz' @self.app.task(shadow_name=shadow_name) def shadowed(): pass old_send_task = self.app.send_task self.app.send_task = Mock() shadowed.delay() self.app.send_task.assert_called_once_with(ANY, ANY, ANY, compression=ANY, delivery_mode=ANY, exchange=ANY, expires=ANY, immediate=ANY, link=ANY, link_error=ANY, mandatory=ANY, priority=ANY, producer=ANY, queue=ANY, result_cls=ANY, routing_key=ANY, serializer=ANY, soft_time_limit=ANY, task_id=ANY, task_type=ANY, time_limit=ANY, shadow='fooxyz', ignore_result=False) self.app.send_task = old_send_task def test_inherit_parent_priority_child_task(self): self.app.conf.task_inherit_parent_priority = True self.app.producer_or_acquire = Mock() self.app.producer_or_acquire.attach_mock( ContextMock(serializer='json'), 'return_value') self.app.amqp.send_task_message = Mock(name="send_task_message") self.task_which_calls_other_task.apply(args=[]) self.app.amqp.send_task_message.assert_called_with( ANY, 't.unit.tasks.test_tasks.task_called_by_other_task', ANY, priority=5, queue=ANY, serializer=ANY) def test_typing__disabled(self): @self.app.task(typing=False) def add(x, y, kw=1): pass add.delay(1) add.delay(1, kw=2) add.delay(1, 2, foobar=3) def test_typing__disabled_by_app(self): with self.Celery(set_as_current=False, strict_typing=False) as app: @app.task() def add(x, y, kw=1): pass assert not add.typing add.delay(1) add.delay(1, kw=2) add.delay(1, 2, foobar=3) @pytest.mark.usefixtures('depends_on_current_app') def test_unpickle_task(self): import pickle @self.app.task(shared=True) def xxx(): pass assert pickle.loads(pickle.dumps(xxx)) is xxx.app.tasks[xxx.name] @patch('celery.app.task.current_app') @pytest.mark.usefixtures('depends_on_current_app') def test_bind__no_app(self, current_app): class XTask(Task): _app = None XTask._app = None XTask.__bound__ = False XTask.bind = Mock(name='bind') assert XTask.app is current_app XTask.bind.assert_called_with(current_app) def test_reprtask__no_fmt(self): assert _reprtask(self.mytask) def test_AsyncResult(self): task_id = uuid() result = self.retry_task.AsyncResult(task_id) assert result.backend == self.retry_task.backend assert result.id == task_id def assert_next_task_data_equal(self, consumer, presult, task_name, test_eta=False, test_expires=False, properties=None, headers=None, **kwargs): next_task = consumer.queues[0].get(accept=['pickle', 'json']) task_properties = next_task.properties task_headers = next_task.headers task_body = next_task.decode() task_args, task_kwargs, embed = task_body assert task_headers['id'] == presult.id assert task_headers['task'] == task_name if test_eta: assert isinstance(task_headers.get('eta'), str) to_datetime = datetime.fromisoformat(task_headers.get('eta')) assert isinstance(to_datetime, datetime) if test_expires: assert isinstance(task_headers.get('expires'), str) to_datetime = datetime.fromisoformat(task_headers.get('expires')) assert isinstance(to_datetime, datetime) properties = properties or {} for arg_name, arg_value in properties.items(): assert task_properties.get(arg_name) == arg_value headers = headers or {} for arg_name, arg_value in headers.items(): assert task_headers.get(arg_name) == arg_value for arg_name, arg_value in kwargs.items(): assert task_kwargs.get(arg_name) == arg_value def test_incomplete_task_cls(self): class IncompleteTask(Task): app = self.app name = 'c.unittest.t.itask' with pytest.raises(NotImplementedError): IncompleteTask().run() def test_task_kwargs_must_be_dictionary(self): with pytest.raises(TypeError): self.increment_counter.apply_async([], 'str') def test_task_args_must_be_list(self): with pytest.raises(TypeError): self.increment_counter.apply_async('s', {}) def test_regular_task(self): assert isinstance(self.mytask, Task) assert self.mytask.run() assert callable(self.mytask) assert self.mytask(), 'Task class runs run() when called' with self.app.connection_or_acquire() as conn: consumer = self.app.amqp.TaskConsumer(conn) with pytest.raises(NotImplementedError): consumer.receive('foo', 'foo') consumer.purge() assert consumer.queues[0].get() is None self.app.amqp.TaskConsumer(conn, queues=[Queue('foo')]) # Without arguments. presult = self.mytask.delay() self.assert_next_task_data_equal( consumer, presult, self.mytask.name) # With arguments. presult2 = self.mytask.apply_async( kwargs={'name': 'George Costanza'}, ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, name='George Costanza', ) # send_task sresult = self.app.send_task(self.mytask.name, kwargs={'name': 'Elaine M. Benes'}) self.assert_next_task_data_equal( consumer, sresult, self.mytask.name, name='Elaine M. Benes', ) # With ETA, absolute expires. presult2 = self.mytask.apply_async( kwargs={'name': 'George Costanza'}, eta=self.now() + timedelta(days=1), expires=self.now() + timedelta(days=2), ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, name='George Costanza', test_eta=True, test_expires=True, ) # With ETA, absolute expires without timezone. presult2 = self.mytask.apply_async( kwargs={'name': 'George Constanza'}, eta=self.now() + timedelta(days=1), expires=(self.now() + timedelta(hours=2)).replace(tzinfo=None), ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, name='George Constanza', test_eta=True, test_expires=True, ) # With ETA, absolute expires in the past. presult2 = self.mytask.apply_async( kwargs={'name': 'George Costanza'}, eta=self.now() + timedelta(days=1), expires=self.now() - timedelta(days=2), ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, name='George Costanza', test_eta=True, test_expires=True, ) # With ETA, relative expires. presult2 = self.mytask.apply_async( kwargs={'name': 'George Costanza'}, eta=self.now() + timedelta(days=1), expires=2 * 24 * 60 * 60, ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, name='George Costanza', test_eta=True, test_expires=True, ) # With countdown. presult2 = self.mytask.apply_async( kwargs={'name': 'George Costanza'}, countdown=10, expires=12, ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, name='George Costanza', test_eta=True, test_expires=True, ) # With ETA, absolute expires in the past in ISO format. presult2 = self.mytask.apply_async( kwargs={'name': 'George Costanza'}, eta=self.now() + timedelta(days=1), expires=self.now() - timedelta(days=2), ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, name='George Costanza', test_eta=True, test_expires=True, ) # Default argsrepr/kwargsrepr behavior presult2 = self.mytask.apply_async( args=('spam',), kwargs={'name': 'Jerry Seinfeld'} ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, headers={'argsrepr': "('spam',)", 'kwargsrepr': "{'name': 'Jerry Seinfeld'}"}, ) # With argsrepr/kwargsrepr presult2 = self.mytask.apply_async( args=('secret',), argsrepr="'***'", kwargs={'password': 'foo'}, kwargsrepr="{'password': '***'}", ) self.assert_next_task_data_equal( consumer, presult2, self.mytask.name, headers={'argsrepr': "'***'", 'kwargsrepr': "{'password': '***'}"}, ) # Discarding all tasks. consumer.purge() self.mytask.apply_async() assert consumer.purge() == 1 assert consumer.queues[0].get() is None assert not presult.successful() self.mytask.backend.mark_as_done(presult.id, result=None) assert presult.successful() def test_send_event(self): mytask = self.mytask._get_current_object() mytask.app.events = Mock(name='events') mytask.app.events.attach_mock(ContextMock(), 'default_dispatcher') mytask.request.id = 'fb' mytask.send_event('task-foo', id=3122) mytask.app.events.default_dispatcher().send.assert_called_with( 'task-foo', uuid='fb', id=3122, retry=True, retry_policy=self.app.conf.task_publish_retry_policy) @pytest.mark.usefixtures('depends_on_current_app') def test_on_replace(self): class CustomStampingVisitor(StampingVisitor): def on_signature(self, sig, **headers) -> dict: return {'header': 'value'} class MyTask(Task): def on_replace(self, sig): sig.stamp(CustomStampingVisitor()) return super().on_replace(sig) mytask = self.app.task(shared=False, base=MyTask)(return_True) sig1 = signature('sig1') with pytest.raises(Ignore): mytask.replace(sig1) assert sig1.options['header'] == 'value' def test_replace(self): sig1 = MagicMock(name='sig1') sig1.options = {} self.mytask.request.id = sentinel.request_id with pytest.raises(Ignore): self.mytask.replace(sig1) sig1.freeze.assert_called_once_with(self.mytask.request.id) sig1.set.assert_called_once_with(replaced_task_nesting=1, chord=ANY, group_id=ANY, group_index=ANY, root_id=ANY) def test_replace_with_chord(self): sig1 = Mock(name='sig1') sig1.options = {'chord': None} with pytest.raises(ImproperlyConfigured): self.mytask.replace(sig1) def test_replace_callback(self): c = group([self.mytask.s()], app=self.app) c.freeze = Mock(name='freeze') c.delay = Mock(name='delay') self.mytask.request.id = 'id' self.mytask.request.group = 'group' self.mytask.request.root_id = 'root_id' self.mytask.request.callbacks = callbacks = 'callbacks' self.mytask.request.errbacks = errbacks = 'errbacks' # Replacement groups get uplifted to chords so that we can accumulate # the results and link call/errbacks - patch the appropriate `chord` # methods so we can validate this behaviour with patch( "celery.canvas.chord.link" ) as mock_chord_link, patch( "celery.canvas.chord.link_error" ) as mock_chord_link_error: with pytest.raises(Ignore): self.mytask.replace(c) # Confirm that the call/errbacks on the original signature are linked # to the replacement signature as expected mock_chord_link.assert_called_once_with(callbacks) mock_chord_link_error.assert_called_once_with(errbacks) def test_replace_group(self): c = group([self.mytask.s()], app=self.app) c.freeze = Mock(name='freeze') c.delay = Mock(name='delay') self.mytask.request.id = 'id' self.mytask.request.group = 'group' self.mytask.request.root_id = 'root_id', with pytest.raises(Ignore): self.mytask.replace(c) def test_replace_chain(self): c = chain([self.mytask.si(), self.mytask.si()], app=self.app) c.freeze = Mock(name='freeze') c.delay = Mock(name='delay') self.mytask.request.id = 'id' self.mytask.request.chain = c with pytest.raises(Ignore): self.mytask.replace(c) def test_replace_run(self): with pytest.raises(Ignore): self.task_replaced_by_other_task.run() def test_replace_run_with_autoretry(self): with pytest.raises(Ignore): self.task_replaced_by_other_task_with_autoretry.run() def test_replace_delay(self): res = self.task_replaced_by_other_task.delay() assert isinstance(res, AsyncResult) def test_replace_apply(self): res = self.task_replaced_by_other_task.apply() assert isinstance(res, EagerResult) assert res.get() == "replaced" def test_add_trail__no_trail(self): mytask = self.increment_counter._get_current_object() mytask.trail = False mytask.add_trail('foo') def test_repr_v2_compat(self): self.mytask.__v2_compat__ = True assert 'v2 compatible' in repr(self.mytask) def test_context_get(self): self.mytask.push_request() try: request = self.mytask.request request.foo = 32 assert request.get('foo') == 32 assert request.get('bar', 36) == 36 request.clear() finally: self.mytask.pop_request() def test_annotate(self): with patch('celery.app.task.resolve_all_annotations') as anno: anno.return_value = [{'FOO': 'BAR'}] @self.app.task(shared=False) def task(): pass task.annotate() assert task.FOO == 'BAR' def test_after_return(self): self.mytask.push_request() try: self.mytask.request.chord = self.mytask.s() self.mytask.after_return('SUCCESS', 1.0, 'foobar', (), {}, None) self.mytask.request.clear() finally: self.mytask.pop_request() def test_update_state(self): @self.app.task(shared=False) def yyy(): pass yyy.push_request() try: tid = uuid() # update_state should accept arbitrary kwargs, which are passed to # the backend store_result method yyy.update_state(tid, 'FROBULATING', {'fooz': 'baaz'}, arbitrary_kwarg=None) assert yyy.AsyncResult(tid).status == 'FROBULATING' assert yyy.AsyncResult(tid).result == {'fooz': 'baaz'} yyy.request.id = tid yyy.update_state(state='FROBUZATING', meta={'fooz': 'baaz'}) assert yyy.AsyncResult(tid).status == 'FROBUZATING' assert yyy.AsyncResult(tid).result == {'fooz': 'baaz'} finally: yyy.pop_request() def test_update_state_passes_request_to_backend(self): backend = Mock() @self.app.task(shared=False, backend=backend) def ttt(): pass ttt.push_request() tid = uuid() ttt.update_state(tid, 'SHRIMMING', {'foo': 'bar'}) backend.store_result.assert_called_once_with( tid, {'foo': 'bar'}, 'SHRIMMING', request=ttt.request ) def test_repr(self): @self.app.task(shared=False) def task_test_repr(): pass assert 'task_test_repr' in repr(task_test_repr) def test_has___name__(self): @self.app.task(shared=False) def yyy2(): pass assert yyy2.__name__ def test_default_priority(self): @self.app.task(shared=False) def yyy3(): pass @self.app.task(shared=False, priority=66) def yyy4(): pass @self.app.task(shared=False, bind=True, base=TaskWithPriority) def yyy5(self): pass self.app.conf.task_default_priority = 42 old_send_task = self.app.send_task self.app.send_task = Mock() yyy3.delay() self.app.send_task.assert_called_once_with(ANY, ANY, ANY, compression=ANY, delivery_mode=ANY, exchange=ANY, expires=ANY, immediate=ANY, link=ANY, link_error=ANY, mandatory=ANY, priority=42, producer=ANY, queue=ANY, result_cls=ANY, routing_key=ANY, serializer=ANY, soft_time_limit=ANY, task_id=ANY, task_type=ANY, time_limit=ANY, shadow=None, ignore_result=False) self.app.send_task = Mock() yyy4.delay() self.app.send_task.assert_called_once_with(ANY, ANY, ANY, compression=ANY, delivery_mode=ANY, exchange=ANY, expires=ANY, immediate=ANY, link=ANY, link_error=ANY, mandatory=ANY, priority=66, producer=ANY, queue=ANY, result_cls=ANY, routing_key=ANY, serializer=ANY, soft_time_limit=ANY, task_id=ANY, task_type=ANY, time_limit=ANY, shadow=None, ignore_result=False) self.app.send_task = Mock() yyy5.delay() self.app.send_task.assert_called_once_with(ANY, ANY, ANY, compression=ANY, delivery_mode=ANY, exchange=ANY, expires=ANY, immediate=ANY, link=ANY, link_error=ANY, mandatory=ANY, priority=10, producer=ANY, queue=ANY, result_cls=ANY, routing_key=ANY, serializer=ANY, soft_time_limit=ANY, task_id=ANY, task_type=ANY, time_limit=ANY, shadow=None, ignore_result=False) self.app.send_task = old_send_task def test_soft_time_limit_failure(self): @self.app.task(soft_time_limit=5, time_limit=3) def yyy(): pass try: yyy_result = yyy.apply_async() yyy_result.get(timeout=5) assert yyy_result.state == 'FAILURE' except ValueError as e: assert str(e) == 'soft_time_limit must be less than or equal to time_limit' class test_apply_task(TasksCase): def test_apply_throw(self): with pytest.raises(KeyError): self.raising.apply(throw=True) def test_apply_with_task_eager_propagates(self): self.app.conf.task_eager_propagates = True with pytest.raises(KeyError): self.raising.apply() def test_apply_request_context_is_ok(self): self.app.conf.task_eager_propagates = True self.task_check_request_context.apply() def test_apply(self): self.increment_counter.count = 0 e = self.increment_counter.apply() assert isinstance(e, EagerResult) assert e.get() == 1 e = self.increment_counter.apply(args=[1]) assert e.get() == 2 e = self.increment_counter.apply(kwargs={'increment_by': 4}) assert e.get() == 6 assert e.successful() assert e.ready() assert e.name == 't.unit.tasks.test_tasks.increment_counter' assert repr(e).startswith('