import json import math from collections.abc import Iterable from unittest.mock import ANY, MagicMock, Mock, call, patch, sentinel import pytest import pytest_subtests # noqa from celery._state import _task_stack from celery.canvas import (Signature, _chain, _maybe_group, _merge_dictionaries, chain, chord, chunks, group, maybe_signature, maybe_unroll_group, signature, xmap, xstarmap) from celery.result import AsyncResult, EagerResult, GroupResult SIG = Signature({ 'task': 'TASK', 'args': ('A1',), 'kwargs': {'K1': 'V1'}, 'options': {'task_id': 'TASK_ID'}, 'subtask_type': ''}, ) def return_True(*args, **kwargs): # Task run functions can't be closures/lambdas, as they're pickled. return True class test_maybe_unroll_group: def test_when_no_len_and_no_length_hint(self): g = MagicMock(name='group') g.tasks.__len__.side_effect = TypeError() g.tasks.__length_hint__ = Mock() g.tasks.__length_hint__.return_value = 0 assert maybe_unroll_group(g) is g g.tasks.__length_hint__.side_effect = AttributeError() assert maybe_unroll_group(g) is g class CanvasCase: def setup_method(self): @self.app.task(shared=False) def add(x, y): return x + y self.add = add @self.app.task(shared=False) def mul(x, y): return x * y self.mul = mul @self.app.task(shared=False) def div(x, y): return x / y self.div = div @self.app.task(shared=False) def xsum(numbers): return sum(sum(num) if isinstance(num, Iterable) else num for num in numbers) self.xsum = xsum @self.app.task(shared=False, bind=True) def replaced(self, x, y): return self.replace(add.si(x, y)) self.replaced = replaced @self.app.task(shared=False, bind=True) def replaced_group(self, x, y): return self.replace(group(add.si(x, y), mul.si(x, y))) self.replaced_group = replaced_group @self.app.task(shared=False, bind=True) def replace_with_group(self, x, y): return self.replace(group(add.si(x, y), mul.si(x, y))) self.replace_with_group = replace_with_group @self.app.task(shared=False, bind=True) def replace_with_chain(self, x, y): return self.replace(group(add.si(x, y) | mul.s(y), add.si(x, y))) self.replace_with_chain = replace_with_chain @self.app.task(shared=False) def xprod(numbers): return math.prod(numbers) self.xprod = xprod @Signature.register_type() class chord_subclass(chord): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.subtask_type = "chord_subclass" @Signature.register_type() class group_subclass(group): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.subtask_type = "group_subclass" @Signature.register_type() class chain_subclass(chain): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.subtask_type = "chain_subclass" @Signature.register_type() class chunks_subclass(chunks): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.subtask_type = "chunks_subclass" class test_Signature(CanvasCase): def test_getitem_property_class(self): assert Signature.task assert Signature.args assert Signature.kwargs assert Signature.options assert Signature.subtask_type def test_getitem_property(self): assert SIG.task == 'TASK' assert SIG.args == ('A1',) assert SIG.kwargs == {'K1': 'V1'} assert SIG.options == {'task_id': 'TASK_ID'} assert SIG.subtask_type == '' def test_call(self): x = Signature('foo', (1, 2), {'arg1': 33}, app=self.app) x.type = Mock(name='type') x(3, 4, arg2=66) x.type.assert_called_with(3, 4, 1, 2, arg1=33, arg2=66) def test_link_on_scalar(self): x = Signature('TASK', link=Signature('B')) assert x.options['link'] x.link(Signature('C')) assert isinstance(x.options['link'], list) assert Signature('B') in x.options['link'] assert Signature('C') in x.options['link'] def test_json(self): x = Signature('TASK', link=Signature('B', app=self.app), app=self.app) assert x.__json__() == dict(x) @pytest.mark.usefixtures('depends_on_current_app') def test_reduce(self): x = Signature('TASK', (2, 4), app=self.app) fun, args = x.__reduce__() assert fun(*args) == x def test_replace(self): x = Signature('TASK', ('A',), {}) assert x.replace(args=('B',)).args == ('B',) assert x.replace(kwargs={'FOO': 'BAR'}).kwargs == { 'FOO': 'BAR', } assert x.replace(options={'task_id': '123'}).options == { 'task_id': '123', } def test_set(self): assert Signature('TASK', x=1).set(task_id='2').options == { 'x': 1, 'task_id': '2', } def test_link(self): x = signature(SIG) x.link(SIG) x.link(SIG) assert SIG in x.options['link'] assert len(x.options['link']) == 1 def test_link_error(self): x = signature(SIG) x.link_error(SIG) x.link_error(SIG) assert SIG in x.options['link_error'] assert len(x.options['link_error']) == 1 def test_flatten_links(self): tasks = [self.add.s(2, 2), self.mul.s(4), self.div.s(2)] tasks[0].link(tasks[1]) tasks[1].link(tasks[2]) assert tasks[0].flatten_links() == tasks def test_OR(self, subtests): x = self.add.s(2, 2) | self.mul.s(4) assert isinstance(x, _chain) y = self.add.s(4, 4) | self.div.s(2) z = x | y assert isinstance(y, _chain) assert isinstance(z, _chain) assert len(z.tasks) == 4 with pytest.raises(TypeError): x | 10 ax = self.add.s(2, 2) | (self.add.s(4) | self.add.s(8)) assert isinstance(ax, _chain) assert len(ax.tasks), 3 == 'consolidates chain to chain' with subtests.test('Test chaining with a non-signature object'): with pytest.raises(TypeError): assert signature('foo') | None def test_INVERT(self): x = self.add.s(2, 2) x.apply_async = Mock() x.apply_async.return_value = Mock() x.apply_async.return_value.get = Mock() x.apply_async.return_value.get.return_value = 4 assert ~x == 4 x.apply_async.assert_called() def test_merge_immutable(self): x = self.add.si(2, 2, foo=1) args, kwargs, options = x._merge((4,), {'bar': 2}, {'task_id': 3}) assert args == (2, 2) assert kwargs == {'foo': 1} assert options == {'task_id': 3} def test_merge_options__none(self): sig = self.add.si() _, _, new_options = sig._merge() assert new_options is sig.options _, _, new_options = sig._merge(options=None) assert new_options is sig.options @pytest.mark.parametrize("immutable_sig", (True, False)) def test_merge_options__group_id(self, immutable_sig): # This is to avoid testing the behaviour in `test_set_immutable()` if immutable_sig: sig = self.add.si() else: sig = self.add.s() # If the signature has no group ID, it can be set assert not sig.options _, _, new_options = sig._merge(options={"group_id": sentinel.gid}) assert new_options == {"group_id": sentinel.gid} # But if one is already set, the new one is silently ignored sig.set(group_id=sentinel.old_gid) _, _, new_options = sig._merge(options={"group_id": sentinel.new_gid}) assert new_options == {"group_id": sentinel.old_gid} def test_set_immutable(self): x = self.add.s(2, 2) assert not x.immutable x.set(immutable=True) assert x.immutable x.set(immutable=False) assert not x.immutable def test_election(self): x = self.add.s(2, 2) x.freeze('foo') x.type.app.control = Mock() r = x.election() x.type.app.control.election.assert_called() assert r.id == 'foo' def test_AsyncResult_when_not_registered(self): s = signature('xxx.not.registered', app=self.app) assert s.AsyncResult def test_apply_async_when_not_registered(self): s = signature('xxx.not.registered', app=self.app) assert s._apply_async def test_keeping_link_error_on_chaining(self): x = self.add.s(2, 2) | self.mul.s(4) assert isinstance(x, _chain) x.link_error(SIG) assert SIG in x.options['link_error'] t = signature(SIG) z = x | t assert isinstance(z, _chain) assert t in z.tasks assert not z.options.get('link_error') assert SIG in z.tasks[0].options['link_error'] assert not z.tasks[2].options.get('link_error') assert SIG in x.options['link_error'] assert t not in x.tasks assert not x.tasks[0].options.get('link_error') z = t | x assert isinstance(z, _chain) assert t in z.tasks assert not z.options.get('link_error') assert SIG in z.tasks[1].options['link_error'] assert not z.tasks[0].options.get('link_error') assert SIG in x.options['link_error'] assert t not in x.tasks assert not x.tasks[0].options.get('link_error') y = self.add.s(4, 4) | self.div.s(2) assert isinstance(y, _chain) z = x | y assert isinstance(z, _chain) assert not z.options.get('link_error') assert SIG in z.tasks[0].options['link_error'] assert not z.tasks[2].options.get('link_error') assert SIG in x.options['link_error'] assert not x.tasks[0].options.get('link_error') z = y | x assert isinstance(z, _chain) assert not z.options.get('link_error') assert SIG in z.tasks[3].options['link_error'] assert not z.tasks[1].options.get('link_error') assert SIG in x.options['link_error'] assert not x.tasks[0].options.get('link_error') def test_signature_on_error_adds_error_callback(self): sig = signature('sig').on_error(signature('on_error')) assert sig.options['link_error'] == [signature('on_error')] @pytest.mark.parametrize('_id, group_id, chord, root_id, parent_id, group_index', [ ('_id', 'group_id', 'chord', 'root_id', 'parent_id', 1), ]) def test_freezing_args_set_in_options(self, _id, group_id, chord, root_id, parent_id, group_index): sig = self.add.s(1, 1) sig.freeze( _id=_id, group_id=group_id, chord=chord, root_id=root_id, parent_id=parent_id, group_index=group_index, ) options = sig.options assert options['task_id'] == _id assert options['group_id'] == group_id assert options['chord'] == chord assert options['root_id'] == root_id assert options['parent_id'] == parent_id assert options['group_index'] == group_index class test_xmap_xstarmap(CanvasCase): def test_apply(self): for type, attr in [(xmap, 'map'), (xstarmap, 'starmap')]: args = [(i, i) for i in range(10)] s = getattr(self.add, attr)(args) s.type = Mock() s.apply_async(foo=1) s.type.apply_async.assert_called_with( (), {'task': self.add.s(), 'it': args}, foo=1, route_name=self.add.name, ) assert type.from_dict(dict(s)) == s assert repr(s) class test_chunks(CanvasCase): def test_chunks_preserves_state(self): x = self.add.chunks(range(100), 10) d = dict(x) d['subtask_type'] = "chunks_subclass" isinstance(chunks_subclass.from_dict(d), chunks_subclass) isinstance(chunks_subclass.from_dict(d).clone(), chunks_subclass) def test_chunks(self): x = self.add.chunks(range(100), 10) assert dict(chunks.from_dict(dict(x), app=self.app)) == dict(x) assert x.group() assert len(x.group().tasks) == 10 x.group = Mock() gr = x.group.return_value = Mock() x.apply_async() gr.apply_async.assert_called_with((), {}, route_name=self.add.name) gr.apply_async.reset_mock() x() gr.apply_async.assert_called_with((), {}, route_name=self.add.name) self.app.conf.task_always_eager = True chunks.apply_chunks(app=self.app, **x['kwargs']) class test_chain(CanvasCase): def test_chain_of_chain_with_a_single_task(self): s = self.add.s(1, 1) assert chain([chain(s)]).tasks == list(chain(s).tasks) @pytest.mark.parametrize("chain_type", (_chain, chain_subclass)) def test_clone_preserves_state(self, chain_type): x = chain_type(self.add.s(i, i) for i in range(10)) assert x.clone().tasks == x.tasks assert x.clone().kwargs == x.kwargs assert x.clone().args == x.args assert isinstance(x.clone(), chain_type) def test_repr(self): x = self.add.s(2, 2) | self.add.s(2) assert repr(x) == f'{self.add.name}(2, 2) | add(2)' def test_apply_async(self): c = self.add.s(2, 2) | self.add.s(4) | self.add.s(8) result = c.apply_async() assert result.parent assert result.parent.parent assert result.parent.parent.parent is None @pytest.mark.parametrize("chain_type", (_chain, chain_subclass)) def test_splices_chains(self, chain_type): c = chain_type( self.add.s(5, 5), chain_type(self.add.s(6), self.add.s(7), self.add.s(8), app=self.app), app=self.app, ) c.freeze() tasks, _ = c._frozen assert len(tasks) == 4 assert isinstance(c, chain_type) @pytest.mark.parametrize("chain_type", [_chain, chain_subclass]) def test_from_dict_no_tasks(self, chain_type): assert chain_type.from_dict(dict(chain_type(app=self.app)), app=self.app) assert isinstance(chain_type.from_dict(dict(chain_type(app=self.app)), app=self.app), chain_type) @pytest.mark.parametrize("chain_type", [_chain, chain_subclass]) def test_from_dict_full_subtasks(self, chain_type): c = chain_type(self.add.si(1, 2), self.add.si(3, 4), self.add.si(5, 6)) serialized = json.loads(json.dumps(c)) deserialized = chain_type.from_dict(serialized) assert all(isinstance(task, Signature) for task in deserialized.tasks) assert isinstance(deserialized, chain_type) @pytest.mark.usefixtures('depends_on_current_app') def test_app_falls_back_to_default(self): from celery._state import current_app assert chain().app is current_app def test_handles_dicts(self): c = chain( self.add.s(5, 5), dict(self.add.s(8)), app=self.app, ) c.freeze() tasks, _ = c._frozen assert all(isinstance(task, Signature) for task in tasks) assert all(task.app is self.app for task in tasks) def test_groups_in_chain_to_chord(self): g1 = group([self.add.s(2, 2), self.add.s(4, 4)]) g2 = group([self.add.s(3, 3), self.add.s(5, 5)]) c = g1 | g2 assert isinstance(c, chord) def test_prepare_steps_set_last_task_id_to_chain(self): last_task = self.add.s(2).set(task_id='42') c = self.add.s(4) | last_task assert c.id is None tasks, _ = c.prepare_steps((), {}, c.tasks, last_task_id=last_task.id) assert c.id == last_task.id def test_group_to_chord(self): c = ( self.add.s(5) | group([self.add.s(i, i) for i in range(5)], app=self.app) | self.add.s(10) | self.add.s(20) | self.add.s(30) ) c._use_link = True tasks, results = c.prepare_steps((), {}, c.tasks) assert tasks[-1].args[0] == 5 assert isinstance(tasks[-2], chord) assert len(tasks[-2].tasks) == 5 body = tasks[-2].body assert len(body.tasks) == 3 assert body.tasks[0].args[0] == 10 assert body.tasks[1].args[0] == 20 assert body.tasks[2].args[0] == 30 c2 = self.add.s(2, 2) | group(self.add.s(i, i) for i in range(10)) c2._use_link = True tasks2, _ = c2.prepare_steps((), {}, c2.tasks) assert isinstance(tasks2[0], group) def test_group_to_chord__protocol_2__or(self): c = ( group([self.add.s(i, i) for i in range(5)], app=self.app) | self.add.s(10) | self.add.s(20) | self.add.s(30) ) assert isinstance(c, chord) def test_group_to_chord__protocol_2(self): c = chain( group([self.add.s(i, i) for i in range(5)], app=self.app), self.add.s(10), self.add.s(20), self.add.s(30) ) assert isinstance(c, chord) assert isinstance(c.body, _chain) assert len(c.body.tasks) == 3 c2 = self.add.s(2, 2) | group(self.add.s(i, i) for i in range(10)) c2._use_link = False tasks2, _ = c2.prepare_steps((), {}, c2.tasks) assert isinstance(tasks2[0], group) def test_chord_to_chain(self): c = ( chord([self.add.s('x0', 'y0'), self.add.s('x1', 'y1')], self.add.s(['foo'])) | chain(self.add.s(['y']), self.add.s(['z'])) ) assert isinstance(c, _chain) assert c.apply().get() == ['x0y0', 'x1y1', 'foo', 'y', 'z'] def test_chord_to_group(self): c = ( chord([self.add.s('x0', 'y0'), self.add.s('x1', 'y1')], self.add.s(['foo'])) | group([self.add.s(['y']), self.add.s(['z'])]) ) assert isinstance(c, _chain) assert c.apply().get() == [ ['x0y0', 'x1y1', 'foo', 'y'], ['x0y0', 'x1y1', 'foo', 'z'] ] def test_chain_of_chord__or__group_of_single_task(self): c = chord([signature('header')], signature('body')) c = chain(c) g = group(signature('t')) new_chain = c | g # g should be chained with the body of c[0] assert isinstance(new_chain, _chain) assert isinstance(new_chain.tasks[0].body, _chain) def test_chain_of_chord_upgrade_on_chaining(self): c = chord([signature('header')], group(signature('body'))) c = chain(c) t = signature('t') new_chain = c | t # t should be chained with the body of c[0] and create a new chord assert isinstance(new_chain, _chain) assert isinstance(new_chain.tasks[0].body, chord) @pytest.mark.parametrize( "group_last_task", [False, True], ) def test_chain_of_chord_upgrade_on_chaining__protocol_2( self, group_last_task): c = chain( group([self.add.s(i, i) for i in range(5)], app=self.app), group([self.add.s(i, i) for i in range(10, 15)], app=self.app), group([self.add.s(i, i) for i in range(20, 25)], app=self.app), self.add.s(30) if not group_last_task else group(self.add.s(30), app=self.app)) assert isinstance(c, _chain) assert len( c.tasks ) == 1, "Consecutive chords should be further upgraded to a single chord." assert isinstance(c.tasks[0], chord) def test_chain_of_chord_upgrade_on_chaining__protocol_3(self): c = chain( chain([self.add.s(i, i) for i in range(5)]), group([self.add.s(i, i) for i in range(10, 15)], app=self.app), chord([signature('header')], signature('body'), app=self.app), group([self.add.s(i, i) for i in range(20, 25)], app=self.app)) assert isinstance(c, _chain) assert isinstance( c.tasks[-1], chord ), "Chord followed by a group should be upgraded to a single chord with chained body." assert len(c.tasks) == 6 def test_apply_options(self): class static(Signature): def clone(self, *args, **kwargs): return self def s(*args, **kwargs): return static(self.add, args, kwargs, type=self.add, app=self.app) c = s(2, 2) | s(4) | s(8) r1 = c.apply_async(task_id='some_id') assert r1.id == 'some_id' c.apply_async(group_id='some_group_id') assert c.tasks[-1].options['group_id'] == 'some_group_id' c.apply_async(chord='some_chord_id') assert c.tasks[-1].options['chord'] == 'some_chord_id' c.apply_async(link=[s(32)]) assert c.tasks[-1].options['link'] == [s(32)] c.apply_async(link_error=[s('error')]) for task in c.tasks: assert task.options['link_error'] == [s('error')] def test_apply_options_none(self): class static(Signature): def clone(self, *args, **kwargs): return self def _apply_async(self, *args, **kwargs): self.args = args self.kwargs = kwargs c = static(self.add, (2, 2), type=self.add, app=self.app, priority=5) c.apply_async(priority=4) assert c.kwargs['priority'] == 4 c.apply_async(priority=None) assert c.kwargs['priority'] == 5 def test_reverse(self): x = self.add.s(2, 2) | self.add.s(2) assert isinstance(signature(x), _chain) assert isinstance(signature(dict(x)), _chain) def test_always_eager(self): self.app.conf.task_always_eager = True assert ~(self.add.s(4, 4) | self.add.s(8)) == 16 def test_chain_always_eager(self): self.app.conf.task_always_eager = True from celery import _state, result fixture_task_join_will_block = _state.task_join_will_block try: _state.task_join_will_block = _state.orig_task_join_will_block result.task_join_will_block = _state.orig_task_join_will_block @self.app.task(shared=False) def chain_add(): return (self.add.s(4, 4) | self.add.s(8)).apply_async() r = chain_add.apply_async(throw=True).get() assert r.get() == 16 finally: _state.task_join_will_block = fixture_task_join_will_block result.task_join_will_block = fixture_task_join_will_block def test_apply(self): x = chain(self.add.s(4, 4), self.add.s(8), self.add.s(10)) res = x.apply() assert isinstance(res, EagerResult) assert res.get() == 26 assert res.parent.get() == 16 assert res.parent.parent.get() == 8 assert res.parent.parent.parent is None def test_kwargs_apply(self): x = chain(self.add.s(), self.add.s(8), self.add.s(10)) res = x.apply(kwargs={'x': 1, 'y': 1}).get() assert res == 20 def test_single_expresion(self): x = chain(self.add.s(1, 2)).apply() assert x.get() == 3 assert x.parent is None def test_empty_chain_returns_none(self): assert chain(app=self.app)() is None assert chain(app=self.app).apply_async() is None def test_call_no_tasks(self): x = chain() assert not x() def test_call_with_tasks(self): x = self.add.s(2, 2) | self.add.s(4) x.apply_async = Mock() x(2, 2, foo=1) x.apply_async.assert_called_with((2, 2), {'foo': 1}) def test_from_dict_no_args__with_args(self): x = dict(self.add.s(2, 2) | self.add.s(4)) x['args'] = None assert isinstance(chain.from_dict(x), _chain) x['args'] = (2,) assert isinstance(chain.from_dict(x), _chain) def test_accepts_generator_argument(self): x = chain(self.add.s(i) for i in range(10)) assert x.tasks[0].type, self.add assert x.type def test_chord_sets_result_parent(self): g = (self.add.s(0, 0) | group(self.add.s(i, i) for i in range(1, 10)) | self.add.s(2, 2) | self.add.s(4, 4)) res = g.freeze() assert isinstance(res, AsyncResult) assert not isinstance(res, GroupResult) assert isinstance(res.parent, AsyncResult) assert not isinstance(res.parent, GroupResult) assert isinstance(res.parent.parent, GroupResult) assert isinstance(res.parent.parent.parent, AsyncResult) assert not isinstance(res.parent.parent.parent, GroupResult) assert res.parent.parent.parent.parent is None seen = set() node = res while node: assert node.id not in seen seen.add(node.id) node = node.parent def test_append_to_empty_chain(self): x = chain() x |= self.add.s(1, 1) x |= self.add.s(1) x.freeze() tasks, _ = x._frozen assert len(tasks) == 2 assert x.apply().get() == 3 @pytest.mark.usefixtures('depends_on_current_app') def test_chain_single_child_result(self): child_sig = self.add.si(1, 1) chain_sig = chain(child_sig) assert chain_sig.tasks[0] is child_sig with patch.object( # We want to get back the result of actually applying the task child_sig, "apply_async", ) as mock_apply, patch.object( # The child signature may be clone by `chain.prepare_steps()` child_sig, "clone", return_value=child_sig, ): res = chain_sig() # `_prepare_chain_from_options()` sets this `chain` kwarg with the # subsequent tasks which would be run - nothing in this case mock_apply.assert_called_once_with(chain=[]) assert res is mock_apply.return_value @pytest.mark.usefixtures('depends_on_current_app') def test_chain_single_child_group_result(self): child_sig = self.add.si(1, 1) # The group will `clone()` the child during instantiation so mock it with patch.object(child_sig, "clone", return_value=child_sig): group_sig = group(child_sig) # Now we can construct the chain signature which is actually under test chain_sig = chain(group_sig) assert chain_sig.tasks[0].tasks[0] is child_sig with patch.object( # We want to get back the result of actually applying the task child_sig, "apply_async", ) as mock_apply, patch.object( # The child signature may be clone by `chain.prepare_steps()` child_sig, "clone", return_value=child_sig, ): res = chain_sig() # `_prepare_chain_from_options()` sets this `chain` kwarg with the # subsequent tasks which would be run - nothing in this case mock_apply.assert_called_once_with(chain=[]) assert res is mock_apply.return_value def test_chain_flattening_keep_links_of_inner_chain(self): def link_chain(sig): sig.link(signature('link_b')) sig.link_error(signature('link_ab')) return sig inner_chain = link_chain(chain(signature('a'), signature('b'))) assert inner_chain.options['link'][0] == signature('link_b') assert inner_chain.options['link_error'][0] == signature('link_ab') assert inner_chain.tasks[0] == signature('a') assert inner_chain.tasks[0].options == {} assert inner_chain.tasks[1] == signature('b') assert inner_chain.tasks[1].options == {} flat_chain = chain(inner_chain, signature('c')) assert flat_chain.options == {} assert flat_chain.tasks[0].name == 'a' assert 'link' not in flat_chain.tasks[0].options assert signature(flat_chain.tasks[0].options['link_error'][0]) == signature('link_ab') assert flat_chain.tasks[1].name == 'b' assert 'link' in flat_chain.tasks[1].options, "b is missing the link from inner_chain.options['link'][0]" assert signature(flat_chain.tasks[1].options['link'][0]) == signature('link_b') assert signature(flat_chain.tasks[1].options['link_error'][0]) == signature('link_ab') def test_group_in_center_of_chain(self): t1 = chain(self.add.si(1, 1), group(self.add.si(1, 1), self.add.si(1, 1)), self.add.si(1, 1) | self.add.si(1, 1)) t2 = chord([self.add.si(1, 1), self.add.si(1, 1)], t1) t2.freeze() # should not raise def test_upgrade_to_chord_on_chain(self): group1 = group(self.add.si(10, 10), self.add.si(10, 10)) group2 = group(self.xsum.s(), self.xsum.s()) chord1 = group1 | group2 chain1 = (self.xsum.si([5]) | self.add.s(1)) final_task = chain(chord1, chain1) assert len(final_task.tasks) == 1 and isinstance(final_task.tasks[0], chord) assert isinstance(final_task.tasks[0].body, chord) assert final_task.tasks[0].body.body == chain1 class test_group(CanvasCase): def test_repr(self): x = group([self.add.s(2, 2), self.add.s(4, 4)]) assert repr(x) def test_repr_empty_group(self): x = group([]) assert repr(x) == 'group()' def test_reverse(self): x = group([self.add.s(2, 2), self.add.s(4, 4)]) assert isinstance(signature(x), group) assert isinstance(signature(dict(x)), group) def test_reverse_with_subclass(self): x = group_subclass([self.add.s(2, 2), self.add.s(4, 4)]) assert isinstance(signature(x), group_subclass) assert isinstance(signature(dict(x)), group_subclass) def test_cannot_link_on_group(self): x = group([self.add.s(2, 2), self.add.s(4, 4)]) with pytest.raises(TypeError): x.apply_async(link=self.add.s(2, 2)) def test_cannot_link_error_on_group(self): x = group([self.add.s(2, 2), self.add.s(4, 4)]) with pytest.raises(TypeError): x.apply_async(link_error=self.add.s(2, 2)) def test_group_with_group_argument(self): g1 = group(self.add.s(2, 2), self.add.s(4, 4), app=self.app) g2 = group(g1, app=self.app) assert g2.tasks is g1.tasks def test_maybe_group_sig(self): assert _maybe_group(self.add.s(2, 2), self.app) == [self.add.s(2, 2)] def test_apply(self): x = group([self.add.s(4, 4), self.add.s(8, 8)]) res = x.apply() assert res.get(), [8 == 16] def test_apply_async(self): x = group([self.add.s(4, 4), self.add.s(8, 8)]) x.apply_async() def test_prepare_with_dict(self): x = group([self.add.s(4, 4), dict(self.add.s(8, 8))], app=self.app) x.apply_async() def test_group_in_group(self): g1 = group(self.add.s(2, 2), self.add.s(4, 4), app=self.app) g2 = group(self.add.s(8, 8), g1, self.add.s(16, 16), app=self.app) g2.apply_async() def test_set_immutable(self): g1 = group(Mock(name='t1'), Mock(name='t2'), app=self.app) g1.set_immutable(True) for task in g1.tasks: task.set_immutable.assert_called_with(True) def test_link(self): g1 = group(Mock(name='t1'), Mock(name='t2'), app=self.app) sig = Mock(name='sig') g1.link(sig) # Only the first child signature of a group will be given the callback # and it is cloned and made immutable to avoid passing results to it, # since that first task can't pass along its siblings' return values g1.tasks[0].link.assert_called_with(sig.clone().set(immutable=True)) def test_link_error(self): g1 = group(Mock(name='t1'), Mock(name='t2'), app=self.app) sig = Mock(name='sig') g1.link_error(sig) # We expect that all group children will be given the errback to ensure # it gets called for child_sig in g1.tasks: child_sig.link_error.assert_called_with(sig.clone(immutable=True)) def test_link_error_with_dict_sig(self): g1 = group(Mock(name='t1'), Mock(name='t2'), app=self.app) errback = signature('tcb') errback_dict = dict(errback) g1.link_error(errback_dict) # We expect that all group children will be given the errback to ensure # it gets called for child_sig in g1.tasks: child_sig.link_error.assert_called_with(errback.clone(immutable=True)) def test_apply_empty(self): x = group(app=self.app) x.apply() res = x.apply_async() assert res assert not res.results def test_apply_async_with_parent(self): _task_stack.push(self.add) try: self.add.push_request(called_directly=False) try: assert not self.add.request.children x = group([self.add.s(4, 4), self.add.s(8, 8)]) res = x() assert self.add.request.children assert res in self.add.request.children assert len(self.add.request.children) == 1 finally: self.add.pop_request() finally: _task_stack.pop() @pytest.mark.parametrize("group_type", (group, group_subclass)) def test_from_dict(self, group_type): x = group_type([self.add.s(2, 2), self.add.s(4, 4)]) x['args'] = (2, 2) value = group_type.from_dict(dict(x)) assert value and isinstance(value, group_type) x['args'] = None value = group_type.from_dict(dict(x)) assert value and isinstance(value, group_type) @pytest.mark.parametrize("group_type", (group, group_subclass)) def test_from_dict_deep_deserialize(self, group_type): original_group = group_type([self.add.s(1, 2)] * 42) serialized_group = json.loads(json.dumps(original_group)) deserialized_group = group_type.from_dict(serialized_group) assert isinstance(deserialized_group, group_type) assert all( isinstance(child_task, Signature) for child_task in deserialized_group.tasks ) @pytest.mark.parametrize("group_type", (group, group_subclass)) def test_from_dict_deeper_deserialize(self, group_type): inner_group = group_type([self.add.s(1, 2)] * 42) outer_group = group_type([inner_group] * 42) serialized_group = json.loads(json.dumps(outer_group)) deserialized_group = group_type.from_dict(serialized_group) assert isinstance(deserialized_group, group_type) assert all( isinstance(child_task, group_type) for child_task in deserialized_group.tasks ) assert all( isinstance(grandchild_task, Signature) for child_task in deserialized_group.tasks for grandchild_task in child_task.tasks ) def test_call_empty_group(self): x = group(app=self.app) assert not len(x()) x.delay() x.apply_async() x() def test_skew(self): g = group([self.add.s(i, i) for i in range(10)]) g.skew(start=1, stop=10, step=1) for i, task in enumerate(g.tasks): assert task.options['countdown'] == i + 1 def test_iter(self): g = group([self.add.s(i, i) for i in range(10)]) assert list(iter(g)) == list(g.keys()) def test_single_task(self): g = group([self.add.s(1, 1)]) assert isinstance(g, group) assert len(g.tasks) == 1 g = group(self.add.s(1, 1)) assert isinstance(g, group) assert len(g.tasks) == 1 @staticmethod def helper_test_get_delay(result): import time t0 = time.time() while not result.ready(): time.sleep(0.01) if time.time() - t0 > 1: return None return result.get() def test_kwargs_direct(self): res = [self.add(x=1, y=1), self.add(x=1, y=1)] assert res == [2, 2] def test_kwargs_apply(self): x = group([self.add.s(), self.add.s()]) res = x.apply(kwargs={'x': 1, 'y': 1}).get() assert res == [2, 2] def test_kwargs_apply_async(self): self.app.conf.task_always_eager = True x = group([self.add.s(), self.add.s()]) res = self.helper_test_get_delay( x.apply_async(kwargs={'x': 1, 'y': 1}) ) assert res == [2, 2] def test_kwargs_delay(self): self.app.conf.task_always_eager = True x = group([self.add.s(), self.add.s()]) res = self.helper_test_get_delay(x.delay(x=1, y=1)) assert res == [2, 2] def test_kwargs_delay_partial(self): self.app.conf.task_always_eager = True x = group([self.add.s(1), self.add.s(x=1)]) res = self.helper_test_get_delay(x.delay(y=1)) assert res == [2, 2] def test_apply_from_generator(self): child_count = 42 child_sig = self.add.si(0, 0) child_sigs_gen = (child_sig for _ in range(child_count)) group_sig = group(child_sigs_gen) with patch("celery.canvas.Signature.apply_async") as mock_apply_async: res_obj = group_sig.apply_async() assert mock_apply_async.call_count == child_count assert len(res_obj.children) == child_count # This needs the current app for some reason not worth digging into @pytest.mark.usefixtures('depends_on_current_app') def test_apply_from_generator_empty(self): empty_gen = (False for _ in range(0)) group_sig = group(empty_gen) with patch("celery.canvas.Signature.apply_async") as mock_apply_async: res_obj = group_sig.apply_async() assert mock_apply_async.call_count == 0 assert len(res_obj.children) == 0 # In the following tests, getting the group ID is a pain so we just use # `ANY` to wildcard it when we're checking on calls made to our mocks def test_apply_contains_chord(self): gchild_count = 42 gchild_sig = self.add.si(0, 0) gchild_sigs = (gchild_sig,) * gchild_count child_chord = chord(gchild_sigs, gchild_sig) group_sig = group((child_chord,)) with patch.object( self.app.backend, "set_chord_size", ) as mock_set_chord_size, patch( "celery.canvas.Signature.apply_async", ) as mock_apply_async: res_obj = group_sig.apply_async() # We only see applies for the header grandchildren because the tasks # are never actually run due to our mocking of `apply_async()` assert mock_apply_async.call_count == gchild_count assert len(res_obj.children) == len(group_sig.tasks) # We must have set the chord size for the group of tasks which makes up # the header of the `child_chord`, just before we apply the last task. mock_set_chord_size.assert_called_once_with(ANY, gchild_count) def test_apply_contains_chords_containing_chain(self): ggchild_count = 42 ggchild_sig = self.add.si(0, 0) gchild_sig = chain((ggchild_sig,) * ggchild_count) child_count = 24 child_chord = chord((gchild_sig,), ggchild_sig) group_sig = group((child_chord,) * child_count) with patch.object( self.app.backend, "set_chord_size", ) as mock_set_chord_size, patch( "celery.canvas.Signature.apply_async", ) as mock_apply_async: res_obj = group_sig.apply_async() # We only see applies for the header grandchildren because the tasks # are never actually run due to our mocking of `apply_async()` assert mock_apply_async.call_count == child_count assert len(res_obj.children) == child_count # We must have set the chord sizes based on the number of tail tasks of # the encapsulated chains - in this case 1 for each child chord mock_set_chord_size.assert_has_calls((call(ANY, 1),) * child_count) @pytest.mark.xfail(reason="Invalid canvas setup with bad exception") def test_apply_contains_chords_containing_empty_chain(self): gchild_sig = chain(tuple()) child_count = 24 child_chord = chord((gchild_sig,), self.add.si(0, 0)) group_sig = group((child_chord,) * child_count) # This is an invalid setup because we can't complete a chord header if # there are no actual tasks which will run in it. However, the current # behaviour of an `IndexError` isn't particularly helpful to a user. group_sig.apply_async() def test_apply_contains_chords_containing_chain_with_empty_tail(self): ggchild_count = 42 ggchild_sig = self.add.si(0, 0) tail_count = 24 gchild_sig = chain( (ggchild_sig,) * ggchild_count + (group((ggchild_sig,) * tail_count), group(tuple()),), ) child_chord = chord((gchild_sig,), ggchild_sig) group_sig = group((child_chord,)) with patch.object( self.app.backend, "set_chord_size", ) as mock_set_chord_size, patch( "celery.canvas.Signature.apply_async", ) as mock_apply_async: res_obj = group_sig.apply_async() # We only see applies for the header grandchildren because the tasks # are never actually run due to our mocking of `apply_async()` assert mock_apply_async.call_count == 1 assert len(res_obj.children) == 1 # We must have set the chord sizes based on the size of the last # non-empty task in the encapsulated chains - in this case `tail_count` # for the group preceding the empty one in each grandchild chain mock_set_chord_size.assert_called_once_with(ANY, tail_count) def test_apply_contains_chords_containing_group(self): ggchild_count = 42 ggchild_sig = self.add.si(0, 0) gchild_sig = group((ggchild_sig,) * ggchild_count) child_count = 24 child_chord = chord((gchild_sig,), ggchild_sig) group_sig = group((child_chord,) * child_count) with patch.object( self.app.backend, "set_chord_size", ) as mock_set_chord_size, patch( "celery.canvas.Signature.apply_async", ) as mock_apply_async: res_obj = group_sig.apply_async() # We see applies for all of the header grandchildren because the tasks # are never actually run due to our mocking of `apply_async()` assert mock_apply_async.call_count == child_count * ggchild_count assert len(res_obj.children) == child_count # We must have set the chord sizes based on the number of tail tasks of # the encapsulated groups - in this case `ggchild_count` mock_set_chord_size.assert_has_calls( (call(ANY, ggchild_count),) * child_count, ) @pytest.mark.xfail(reason="Invalid canvas setup but poor behaviour") def test_apply_contains_chords_containing_empty_group(self): gchild_sig = group(tuple()) child_count = 24 child_chord = chord((gchild_sig,), self.add.si(0, 0)) group_sig = group((child_chord,) * child_count) with patch.object( self.app.backend, "set_chord_size", ) as mock_set_chord_size, patch( "celery.canvas.Signature.apply_async", ) as mock_apply_async: res_obj = group_sig.apply_async() # We only see applies for the header grandchildren because the tasks # are never actually run due to our mocking of `apply_async()` assert mock_apply_async.call_count == child_count assert len(res_obj.children) == child_count # This is actually kind of meaningless because, similar to the empty # chain test, this is an invalid setup. However, we should probably # expect that the chords are dealt with in some other way the probably # being left incomplete forever... mock_set_chord_size.assert_has_calls((call(ANY, 0),) * child_count) def test_apply_contains_chords_containing_chord(self): ggchild_count = 42 ggchild_sig = self.add.si(0, 0) gchild_sig = chord((ggchild_sig,) * ggchild_count, ggchild_sig) child_count = 24 child_chord = chord((gchild_sig,), ggchild_sig) group_sig = group((child_chord,) * child_count) with patch.object( self.app.backend, "set_chord_size", ) as mock_set_chord_size, patch( "celery.canvas.Signature.apply_async", ) as mock_apply_async: res_obj = group_sig.apply_async() # We see applies for all of the header great-grandchildren because the # tasks are never actually run due to our mocking of `apply_async()` assert mock_apply_async.call_count == child_count * ggchild_count assert len(res_obj.children) == child_count # We must have set the chord sizes based on the number of tail tasks of # the deeply encapsulated chords' header tasks, as well as for each # child chord. This means we have `child_count` interleaved calls to # set chord sizes of 1 and `ggchild_count`. mock_set_chord_size.assert_has_calls( (call(ANY, 1), call(ANY, ggchild_count),) * child_count, ) def test_apply_contains_chords_containing_empty_chord(self): gchild_sig = chord(tuple(), self.add.si(0, 0)) child_count = 24 child_chord = chord((gchild_sig,), self.add.si(0, 0)) group_sig = group((child_chord,) * child_count) with patch.object( self.app.backend, "set_chord_size", ) as mock_set_chord_size, patch( "celery.canvas.Signature.apply_async", ) as mock_apply_async: res_obj = group_sig.apply_async() # We only see applies for the header grandchildren because the tasks # are never actually run due to our mocking of `apply_async()` assert mock_apply_async.call_count == child_count assert len(res_obj.children) == child_count # We must have set the chord sizes based on the number of tail tasks of # the encapsulated chains - in this case 1 for each child chord mock_set_chord_size.assert_has_calls((call(ANY, 1),) * child_count) def test_group_prepared(self): # Using both partial and dict based signatures sig = group(dict(self.add.s(0)), self.add.s(0)) _, group_id, root_id = sig._freeze_gid({}) tasks = sig._prepared(sig.tasks, [42], group_id, root_id, self.app) for task, result, group_id in tasks: assert isinstance(task, Signature) assert task.args[0] == 42 assert task.args[1] == 0 assert isinstance(result, AsyncResult) assert group_id is not None class test_chord(CanvasCase): def test__get_app_does_not_exhaust_generator(self): def build_generator(): yield self.add.s(1, 1) self.second_item_returned = True yield self.add.s(2, 2) raise pytest.fail("This should never be reached") self.second_item_returned = False c = chord(build_generator(), self.add.s(3)) c.app # The second task gets returned due to lookahead in `regen()` assert self.second_item_returned # Access it again to make sure the generator is not further evaluated c.app @pytest.mark.parametrize("chord_type", [chord, chord_subclass]) def test_reverse(self, chord_type): x = chord_type([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4)) assert isinstance(signature(x), chord_type) assert isinstance(signature(dict(x)), chord_type) def test_clone_clones_body(self): x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4)) y = x.clone() assert x.kwargs['body'] is not y.kwargs['body'] y.kwargs.pop('body') z = y.clone() assert z.kwargs.get('body') is None def test_argument_is_group(self): x = chord(group(self.add.s(2, 2), self.add.s(4, 4), app=self.app)) assert x.tasks def test_app_when_app(self): app = Mock(name='app') x = chord([self.add.s(4, 4)], app=app) assert x.app is app def test_app_when_app_in_task(self): t1 = Mock(name='t1') t2 = Mock(name='t2') x = chord([t1, self.add.s(4, 4)]) assert x.app is x.tasks[0].app t1.app = None x = chord([t1], body=t2) assert x.app is t2._app def test_app_when_header_is_empty(self): x = chord([], self.add.s(4, 4)) assert x.app is self.add.app @pytest.mark.usefixtures('depends_on_current_app') def test_app_fallback_to_current(self): from celery._state import current_app t1 = Mock(name='t1') t1.app = t1._app = None x = chord([t1], body=t1) assert x.app is current_app def test_chord_size_simple(self): sig = chord(self.add.s()) assert sig.__length_hint__() == 1 def test_chord_size_with_body(self): sig = chord(self.add.s(), self.add.s()) assert sig.__length_hint__() == 1 def test_chord_size_explicit_group_single(self): sig = chord(group(self.add.s())) assert sig.__length_hint__() == 1 def test_chord_size_explicit_group_many(self): sig = chord(group([self.add.s()] * 42)) assert sig.__length_hint__() == 42 def test_chord_size_implicit_group_single(self): sig = chord([self.add.s()]) assert sig.__length_hint__() == 1 def test_chord_size_implicit_group_many(self): sig = chord([self.add.s()] * 42) assert sig.__length_hint__() == 42 def test_chord_size_chain_single(self): sig = chord(chain(self.add.s())) assert sig.__length_hint__() == 1 def test_chord_size_chain_many(self): # Chains get flattened into the encapsulating chord so even though the # chain would only count for 1, the tasks we pulled into the chord's # header and are counted as a bunch of simple signature objects sig = chord(chain([self.add.s()] * 42)) assert sig.__length_hint__() == 42 def test_chord_size_nested_chain_chain_single(self): sig = chord(chain(chain(self.add.s()))) assert sig.__length_hint__() == 1 def test_chord_size_nested_chain_chain_many(self): # The outer chain will be pulled up into the chord but the lower one # remains and will only count as a single final element sig = chord(chain(chain([self.add.s()] * 42))) assert sig.__length_hint__() == 1 def test_chord_size_implicit_chain_single(self): sig = chord([self.add.s()]) assert sig.__length_hint__() == 1 def test_chord_size_implicit_chain_many(self): # This isn't a chain object so the `tasks` attribute can't be lifted # into the chord - this isn't actually valid and would blow up we tried # to run it but it sanity checks our recursion sig = chord([[self.add.s()] * 42]) assert sig.__length_hint__() == 1 def test_chord_size_nested_implicit_chain_chain_single(self): sig = chord([chain(self.add.s())]) assert sig.__length_hint__() == 1 def test_chord_size_nested_implicit_chain_chain_many(self): sig = chord([chain([self.add.s()] * 42)]) assert sig.__length_hint__() == 1 def test_chord_size_nested_chord_body_simple(self): sig = chord(chord(tuple(), self.add.s())) assert sig.__length_hint__() == 1 def test_chord_size_nested_chord_body_implicit_group_single(self): sig = chord(chord(tuple(), [self.add.s()])) assert sig.__length_hint__() == 1 def test_chord_size_nested_chord_body_implicit_group_many(self): sig = chord(chord(tuple(), [self.add.s()] * 42)) assert sig.__length_hint__() == 42 # Nested groups in a chain only affect the chord size if they are the last # element in the chain - in that case each group element is counted def test_chord_size_nested_group_chain_group_head_single(self): x = chord( group( [group(self.add.s()) | self.add.s()] * 42 ), body=self.add.s() ) assert x.__length_hint__() == 42 def test_chord_size_nested_group_chain_group_head_many(self): x = chord( group( [group([self.add.s()] * 4) | self.add.s()] * 2 ), body=self.add.s() ) assert x.__length_hint__() == 2 def test_chord_size_nested_group_chain_group_mid_single(self): x = chord( group( [self.add.s() | group(self.add.s()) | self.add.s()] * 42 ), body=self.add.s() ) assert x.__length_hint__() == 42 def test_chord_size_nested_group_chain_group_mid_many(self): x = chord( group( [self.add.s() | group([self.add.s()] * 4) | self.add.s()] * 2 ), body=self.add.s() ) assert x.__length_hint__() == 2 def test_chord_size_nested_group_chain_group_tail_single(self): x = chord( group( [self.add.s() | group(self.add.s())] * 42 ), body=self.add.s() ) assert x.__length_hint__() == 42 def test_chord_size_nested_group_chain_group_tail_many(self): x = chord( group( [self.add.s() | group([self.add.s()] * 4)] * 2 ), body=self.add.s() ) assert x.__length_hint__() == 4 * 2 def test_chord_size_nested_implicit_group_chain_group_tail_single(self): x = chord( [self.add.s() | group(self.add.s())] * 42, body=self.add.s() ) assert x.__length_hint__() == 42 def test_chord_size_nested_implicit_group_chain_group_tail_many(self): x = chord( [self.add.s() | group([self.add.s()] * 4)] * 2, body=self.add.s() ) assert x.__length_hint__() == 4 * 2 def test_chord_size_deserialized_element_single(self): child_sig = self.add.s() deserialized_child_sig = json.loads(json.dumps(child_sig)) # We have to break in to be sure that a child remains as a `dict` so we # can confirm that the length hint will instantiate a `Signature` # object and then descend as expected chord_sig = chord(tuple()) chord_sig.tasks = [deserialized_child_sig] with patch( "celery.canvas.Signature.from_dict", return_value=child_sig ) as mock_from_dict: assert chord_sig.__length_hint__() == 1 mock_from_dict.assert_called_once_with(deserialized_child_sig) def test_chord_size_deserialized_element_many(self): child_sig = self.add.s() deserialized_child_sig = json.loads(json.dumps(child_sig)) # We have to break in to be sure that a child remains as a `dict` so we # can confirm that the length hint will instantiate a `Signature` # object and then descend as expected chord_sig = chord(tuple()) chord_sig.tasks = [deserialized_child_sig] * 42 with patch( "celery.canvas.Signature.from_dict", return_value=child_sig ) as mock_from_dict: assert chord_sig.__length_hint__() == 42 mock_from_dict.assert_has_calls([call(deserialized_child_sig)] * 42) def test_set_immutable(self): x = chord([Mock(name='t1'), Mock(name='t2')], app=self.app) x.set_immutable(True) def test_links_to_body(self): x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4)) x.link(self.div.s(2)) assert not x.options.get('link') assert x.kwargs['body'].options['link'] x.link_error(self.div.s(2)) assert not x.options.get('link_error') assert x.kwargs['body'].options['link_error'] assert x.tasks assert x.body def test_repr(self): x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4)) assert repr(x) x.kwargs['body'] = None assert 'without body' in repr(x) @pytest.mark.parametrize("group_type", [group, group_subclass]) def test_freeze_tasks_body_is_group(self, subtests, group_type): # Confirm that `group index` values counting up from 0 are set for # elements of a chord's body when the chord is encapsulated in a group body_elem = self.add.s() chord_body = group_type([body_elem] * 42) chord_obj = chord(self.add.s(), body=chord_body) top_group = group_type([chord_obj]) # We expect the body to be the signature we passed in before we freeze with subtests.test(msg="Validate body type and tasks are retained"): assert isinstance(chord_obj.body, group_type) assert all( embedded_body_elem is body_elem for embedded_body_elem in chord_obj.body.tasks ) # We also expect the body to have no initial options - since all of the # embedded body elements are confirmed to be `body_elem` this is valid assert body_elem.options == {} # When we freeze the chord, its body will be cloned and options set top_group.freeze() with subtests.test( msg="Validate body group indices count from 0 after freezing" ): assert isinstance(chord_obj.body, group_type) assert all( embedded_body_elem is not body_elem for embedded_body_elem in chord_obj.body.tasks ) assert all( embedded_body_elem.options["group_index"] == i for i, embedded_body_elem in enumerate(chord_obj.body.tasks) ) def test_freeze_tasks_is_not_group(self): x = chord([self.add.s(2, 2)], body=self.add.s(), app=self.app) x.freeze() x.tasks = [self.add.s(2, 2)] x.freeze() def test_chain_always_eager(self): self.app.conf.task_always_eager = True from celery import _state, result fixture_task_join_will_block = _state.task_join_will_block try: _state.task_join_will_block = _state.orig_task_join_will_block result.task_join_will_block = _state.orig_task_join_will_block @self.app.task(shared=False) def finalize(*args): pass @self.app.task(shared=False) def chord_add(): return chord([self.add.s(4, 4)], finalize.s()).apply_async() chord_add.apply_async(throw=True).get() finally: _state.task_join_will_block = fixture_task_join_will_block result.task_join_will_block = fixture_task_join_will_block @pytest.mark.parametrize("chord_type", [chord, chord_subclass]) def test_from_dict(self, chord_type): header = self.add.s(1, 2) original_chord = chord_type(header=header) rebuilt_chord = chord_type.from_dict(dict(original_chord)) assert isinstance(rebuilt_chord, chord_type) @pytest.mark.parametrize("chord_type", [chord, chord_subclass]) def test_from_dict_with_body(self, chord_type): header = body = self.add.s(1, 2) original_chord = chord_type(header=header, body=body) rebuilt_chord = chord_type.from_dict(dict(original_chord)) assert isinstance(rebuilt_chord, chord_type) def test_from_dict_deep_deserialize(self, subtests): header = body = self.add.s(1, 2) original_chord = chord(header=header, body=body) serialized_chord = json.loads(json.dumps(original_chord)) deserialized_chord = chord.from_dict(serialized_chord) with subtests.test(msg="Verify chord is deserialized"): assert isinstance(deserialized_chord, chord) with subtests.test(msg="Validate chord header tasks is deserialized"): assert all( isinstance(child_task, Signature) for child_task in deserialized_chord.tasks ) with subtests.test(msg="Verify chord body is deserialized"): assert isinstance(deserialized_chord.body, Signature) @pytest.mark.parametrize("group_type", [group, group_subclass]) def test_from_dict_deep_deserialize_group(self, subtests, group_type): header = body = group_type([self.add.s(1, 2)] * 42) original_chord = chord(header=header, body=body) serialized_chord = json.loads(json.dumps(original_chord)) deserialized_chord = chord.from_dict(serialized_chord) with subtests.test(msg="Verify chord is deserialized"): assert isinstance(deserialized_chord, chord) # A header which is a group gets unpacked into the chord's `tasks` with subtests.test( msg="Validate chord header tasks are deserialized and unpacked" ): assert all( isinstance(child_task, Signature) and not isinstance(child_task, group_type) for child_task in deserialized_chord.tasks ) # A body which is a group remains as it we passed in with subtests.test( msg="Validate chord body is deserialized and not unpacked" ): assert isinstance(deserialized_chord.body, group_type) assert all( isinstance(body_child_task, Signature) for body_child_task in deserialized_chord.body.tasks ) @pytest.mark.parametrize("group_type", [group, group_subclass]) def test_from_dict_deeper_deserialize_group(self, subtests, group_type): inner_group = group_type([self.add.s(1, 2)] * 42) header = body = group_type([inner_group] * 42) original_chord = chord(header=header, body=body) serialized_chord = json.loads(json.dumps(original_chord)) deserialized_chord = chord.from_dict(serialized_chord) with subtests.test(msg="Verify chord is deserialized"): assert isinstance(deserialized_chord, chord) # A header which is a group gets unpacked into the chord's `tasks` with subtests.test( msg="Validate chord header tasks are deserialized and unpacked" ): assert all( isinstance(child_task, group_type) for child_task in deserialized_chord.tasks ) assert all( isinstance(grandchild_task, Signature) for child_task in deserialized_chord.tasks for grandchild_task in child_task.tasks ) # A body which is a group remains as it we passed in with subtests.test( msg="Validate chord body is deserialized and not unpacked" ): assert isinstance(deserialized_chord.body, group) assert all( isinstance(body_child_task, group) for body_child_task in deserialized_chord.body.tasks ) assert all( isinstance(body_grandchild_task, Signature) for body_child_task in deserialized_chord.body.tasks for body_grandchild_task in body_child_task.tasks ) def test_from_dict_deep_deserialize_chain(self, subtests): header = body = chain([self.add.s(1, 2)] * 42) original_chord = chord(header=header, body=body) serialized_chord = json.loads(json.dumps(original_chord)) deserialized_chord = chord.from_dict(serialized_chord) with subtests.test(msg="Verify chord is deserialized"): assert isinstance(deserialized_chord, chord) # A header which is a chain gets unpacked into the chord's `tasks` with subtests.test( msg="Validate chord header tasks are deserialized and unpacked" ): assert all( isinstance(child_task, Signature) and not isinstance(child_task, chain) for child_task in deserialized_chord.tasks ) # A body which is a chain gets mutatated into the hidden `_chain` class with subtests.test( msg="Validate chord body is deserialized and not unpacked" ): assert isinstance(deserialized_chord.body, _chain) def test_chord_clone_kwargs(self, subtests): """ Test that chord clone ensures the kwargs are the same """ with subtests.test(msg='Verify chord cloning clones kwargs correctly'): c = chord([signature('g'), signature('h')], signature('i'), kwargs={'U': 6}) c2 = c.clone() assert c2.kwargs == c.kwargs with subtests.test(msg='Cloning the chord with overridden kwargs'): override_kw = {'X': 2} c3 = c.clone(args=(1,), kwargs=override_kw) with subtests.test(msg='Verify the overridden kwargs were cloned correctly'): new_kw = c.kwargs.copy() new_kw.update(override_kw) assert c3.kwargs == new_kw def test_flag_allow_error_cb_on_chord_header(self, subtests): header_mock = [Mock(name='t1'), Mock(name='t2')] header = group(header_mock) body = Mock(name='tbody') errback_sig = Mock(name='errback_sig') chord_sig = chord(header, body, app=self.app) with subtests.test(msg='Verify the errback is not linked'): # header for child_sig in header_mock: child_sig.link_error.assert_not_called() # body body.link_error.assert_not_called() with subtests.test(msg='Verify flag turned off links only the body'): self.app.conf.task_allow_error_cb_on_chord_header = False chord_sig.link_error(errback_sig) # header for child_sig in header_mock: child_sig.link_error.assert_not_called() # body body.link_error.assert_called_once_with(errback_sig) with subtests.test(msg='Verify flag turned on links the header'): self.app.conf.task_allow_error_cb_on_chord_header = True chord_sig.link_error(errback_sig) # header for child_sig in header_mock: child_sig.link_error.assert_called_once_with(errback_sig.clone(immutable=True)) # body body.link_error.assert_has_calls([call(errback_sig), call(errback_sig)]) @pytest.mark.usefixtures('depends_on_current_app') def test_flag_allow_error_cb_on_chord_header_various_header_types(self): """ Test chord link_error with various header types. """ self.app.conf.task_allow_error_cb_on_chord_header = True headers = [ signature('t'), [signature('t'), signature('t')], group(signature('t'), signature('t')) ] for chord_header in headers: c = chord(chord_header, signature('t'), app=self.app) sig = signature('t') errback = c.link_error(sig) assert errback == sig @pytest.mark.usefixtures('depends_on_current_app') def test_flag_allow_error_cb_on_chord_header_with_dict_callback(self): self.app.conf.task_allow_error_cb_on_chord_header = True c = chord(group(signature('th1'), signature('th2')), signature('tbody'), app=self.app) errback_dict = dict(signature('tcb')) errback = c.link_error(errback_dict) assert errback == errback_dict def test_chord__or__group_of_single_task(self): """ Test chaining a chord to a group of a single task. """ c = chord([signature('header')], signature('body')) g = group(signature('t')) stil_chord = c | g # g should be chained with the body of c assert isinstance(stil_chord, chord) assert isinstance(stil_chord.body, _chain) def test_chord_upgrade_on_chaining(self): """ Test that chaining a chord with a group body upgrades to a new chord """ c = chord([signature('header')], group(signature('body'))) t = signature('t') stil_chord = c | t # t should be chained with the body of c and create a new chord assert isinstance(stil_chord, chord) assert isinstance(stil_chord.body, chord) @pytest.mark.parametrize('header', [ [signature('s1'), signature('s2')], group(signature('s1'), signature('s2')) ]) @pytest.mark.usefixtures('depends_on_current_app') def test_link_error_on_chord_header(self, header): """ Test that link_error on a chord also links the header """ self.app.conf.task_allow_error_cb_on_chord_header = True c = chord(header, signature('body'), app=self.app) err = signature('err') errback = c.link_error(err) assert errback == err for header_task in c.tasks: assert header_task.options['link_error'] == [err.clone(immutable=True)] assert c.body.options['link_error'] == [err] class test_maybe_signature(CanvasCase): def test_is_None(self): assert maybe_signature(None, app=self.app) is None def test_is_dict(self): assert isinstance(maybe_signature(dict(self.add.s()), app=self.app), Signature) def test_when_sig(self): s = self.add.s() assert maybe_signature(s, app=self.app) is s class test_merge_dictionaries(CanvasCase): def test_docstring_example(self): d1 = {'dict': {'a': 1}, 'list': [1, 2], 'tuple': (1, 2)} d2 = {'dict': {'b': 2}, 'list': [3, 4], 'set': {'a', 'b'}} _merge_dictionaries(d1, d2) assert d1 == { 'dict': {'a': 1, 'b': 2}, 'list': [1, 2, 3, 4], 'tuple': (1, 2), 'set': {'a', 'b'} } @pytest.mark.parametrize('d1,d2,expected_result', [ ( {'None': None}, {'None': None}, {'None': [None]} ), ( {'None': None}, {'None': [None]}, {'None': [[None]]} ), ( {'None': None}, {'None': 'Not None'}, {'None': ['Not None']} ), ( {'None': None}, {'None': ['Not None']}, {'None': [['Not None']]} ), ( {'None': [None]}, {'None': None}, {'None': [None, None]} ), ( {'None': [None]}, {'None': [None]}, {'None': [None, None]} ), ( {'None': [None]}, {'None': 'Not None'}, {'None': [None, 'Not None']} ), ( {'None': [None]}, {'None': ['Not None']}, {'None': [None, 'Not None']} ), ]) def test_none_values(self, d1, d2, expected_result): _merge_dictionaries(d1, d2) assert d1 == expected_result