test_canvas.py 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785
  1. from __future__ import absolute_import, unicode_literals
  2. import json
  3. import pytest
  4. from case import MagicMock, Mock
  5. from celery._state import _task_stack
  6. from celery.canvas import (Signature, _chain, _maybe_group, chain, chord,
  7. chunks, group, maybe_signature, maybe_unroll_group,
  8. signature, xmap, xstarmap)
  9. from celery.result import AsyncResult, EagerResult, GroupResult
  10. SIG = Signature({
  11. 'task': 'TASK',
  12. 'args': ('A1',),
  13. 'kwargs': {'K1': 'V1'},
  14. 'options': {'task_id': 'TASK_ID'},
  15. 'subtask_type': ''},
  16. )
  17. class test_maybe_unroll_group:
  18. def test_when_no_len_and_no_length_hint(self):
  19. g = MagicMock(name='group')
  20. g.tasks.__len__.side_effect = TypeError()
  21. g.tasks.__length_hint__ = Mock()
  22. g.tasks.__length_hint__.return_value = 0
  23. assert maybe_unroll_group(g) is g
  24. g.tasks.__length_hint__.side_effect = AttributeError()
  25. assert maybe_unroll_group(g) is g
  26. class CanvasCase:
  27. def setup(self):
  28. @self.app.task(shared=False)
  29. def add(x, y):
  30. return x + y
  31. self.add = add
  32. @self.app.task(shared=False)
  33. def mul(x, y):
  34. return x * y
  35. self.mul = mul
  36. @self.app.task(shared=False)
  37. def div(x, y):
  38. return x / y
  39. self.div = div
  40. class test_Signature(CanvasCase):
  41. def test_getitem_property_class(self):
  42. assert Signature.task
  43. assert Signature.args
  44. assert Signature.kwargs
  45. assert Signature.options
  46. assert Signature.subtask_type
  47. def test_getitem_property(self):
  48. assert SIG.task == 'TASK'
  49. assert SIG.args == ('A1',)
  50. assert SIG.kwargs == {'K1': 'V1'}
  51. assert SIG.options == {'task_id': 'TASK_ID'}
  52. assert SIG.subtask_type == ''
  53. def test_call(self):
  54. x = Signature('foo', (1, 2), {'arg1': 33}, app=self.app)
  55. x.type = Mock(name='type')
  56. x(3, 4, arg2=66)
  57. x.type.assert_called_with(3, 4, 1, 2, arg1=33, arg2=66)
  58. def test_link_on_scalar(self):
  59. x = Signature('TASK', link=Signature('B'))
  60. assert x.options['link']
  61. x.link(Signature('C'))
  62. assert isinstance(x.options['link'], list)
  63. assert Signature('B') in x.options['link']
  64. assert Signature('C') in x.options['link']
  65. def test_json(self):
  66. x = Signature('TASK', link=Signature('B', app=self.app), app=self.app)
  67. assert x.__json__() == dict(x)
  68. @pytest.mark.usefixtures('depends_on_current_app')
  69. def test_reduce(self):
  70. x = Signature('TASK', (2, 4), app=self.app)
  71. fun, args = x.__reduce__()
  72. assert fun(*args) == x
  73. def test_replace(self):
  74. x = Signature('TASK', ('A'), {})
  75. assert x.replace(args=('B',)).args == ('B',)
  76. assert x.replace(kwargs={'FOO': 'BAR'}).kwargs == {
  77. 'FOO': 'BAR',
  78. }
  79. assert x.replace(options={'task_id': '123'}).options == {
  80. 'task_id': '123',
  81. }
  82. def test_set(self):
  83. assert Signature('TASK', x=1).set(task_id='2').options == {
  84. 'x': 1, 'task_id': '2',
  85. }
  86. def test_link(self):
  87. x = signature(SIG)
  88. x.link(SIG)
  89. x.link(SIG)
  90. assert SIG in x.options['link']
  91. assert len(x.options['link']) == 1
  92. def test_link_error(self):
  93. x = signature(SIG)
  94. x.link_error(SIG)
  95. x.link_error(SIG)
  96. assert SIG in x.options['link_error']
  97. assert len(x.options['link_error']) == 1
  98. def test_flatten_links(self):
  99. tasks = [self.add.s(2, 2), self.mul.s(4), self.div.s(2)]
  100. tasks[0].link(tasks[1])
  101. tasks[1].link(tasks[2])
  102. assert tasks[0].flatten_links() == tasks
  103. def test_OR(self):
  104. x = self.add.s(2, 2) | self.mul.s(4)
  105. assert isinstance(x, _chain)
  106. y = self.add.s(4, 4) | self.div.s(2)
  107. z = x | y
  108. assert isinstance(y, _chain)
  109. assert isinstance(z, _chain)
  110. assert len(z.tasks) == 4
  111. with pytest.raises(TypeError):
  112. x | 10
  113. ax = self.add.s(2, 2) | (self.add.s(4) | self.add.s(8))
  114. assert isinstance(ax, _chain)
  115. assert len(ax.tasks), 3 == 'consolidates chain to chain'
  116. def test_INVERT(self):
  117. x = self.add.s(2, 2)
  118. x.apply_async = Mock()
  119. x.apply_async.return_value = Mock()
  120. x.apply_async.return_value.get = Mock()
  121. x.apply_async.return_value.get.return_value = 4
  122. assert ~x == 4
  123. x.apply_async.assert_called()
  124. def test_merge_immutable(self):
  125. x = self.add.si(2, 2, foo=1)
  126. args, kwargs, options = x._merge((4,), {'bar': 2}, {'task_id': 3})
  127. assert args == (2, 2)
  128. assert kwargs == {'foo': 1}
  129. assert options == {'task_id': 3}
  130. def test_set_immutable(self):
  131. x = self.add.s(2, 2)
  132. assert not x.immutable
  133. x.set(immutable=True)
  134. assert x.immutable
  135. x.set(immutable=False)
  136. assert not x.immutable
  137. def test_election(self):
  138. x = self.add.s(2, 2)
  139. x.freeze('foo')
  140. x.type.app.control = Mock()
  141. r = x.election()
  142. x.type.app.control.election.assert_called()
  143. assert r.id == 'foo'
  144. def test_AsyncResult_when_not_registered(self):
  145. s = signature('xxx.not.registered', app=self.app)
  146. assert s.AsyncResult
  147. def test_apply_async_when_not_registered(self):
  148. s = signature('xxx.not.registered', app=self.app)
  149. assert s._apply_async
  150. def test_keeping_link_error_on_chaining(self):
  151. x = self.add.s(2, 2) | self.mul.s(4)
  152. assert isinstance(x, _chain)
  153. x.link_error(SIG)
  154. assert SIG in x.options['link_error']
  155. t = signature(SIG)
  156. z = x | t
  157. assert isinstance(z, _chain)
  158. assert t in z.tasks
  159. assert not z.options.get('link_error')
  160. assert SIG in z.tasks[0].options['link_error']
  161. assert not z.tasks[2].options.get('link_error')
  162. assert SIG in x.options['link_error']
  163. assert t not in x.tasks
  164. assert not x.tasks[0].options.get('link_error')
  165. z = t | x
  166. assert isinstance(z, _chain)
  167. assert t in z.tasks
  168. assert not z.options.get('link_error')
  169. assert SIG in z.tasks[1].options['link_error']
  170. assert not z.tasks[0].options.get('link_error')
  171. assert SIG in x.options['link_error']
  172. assert t not in x.tasks
  173. assert not x.tasks[0].options.get('link_error')
  174. y = self.add.s(4, 4) | self.div.s(2)
  175. assert isinstance(y, _chain)
  176. z = x | y
  177. assert isinstance(z, _chain)
  178. assert not z.options.get('link_error')
  179. assert SIG in z.tasks[0].options['link_error']
  180. assert not z.tasks[2].options.get('link_error')
  181. assert SIG in x.options['link_error']
  182. assert not x.tasks[0].options.get('link_error')
  183. z = y | x
  184. assert isinstance(z, _chain)
  185. assert not z.options.get('link_error')
  186. assert SIG in z.tasks[3].options['link_error']
  187. assert not z.tasks[1].options.get('link_error')
  188. assert SIG in x.options['link_error']
  189. assert not x.tasks[0].options.get('link_error')
  190. class test_xmap_xstarmap(CanvasCase):
  191. def test_apply(self):
  192. for type, attr in [(xmap, 'map'), (xstarmap, 'starmap')]:
  193. args = [(i, i) for i in range(10)]
  194. s = getattr(self.add, attr)(args)
  195. s.type = Mock()
  196. s.apply_async(foo=1)
  197. s.type.apply_async.assert_called_with(
  198. (), {'task': self.add.s(), 'it': args}, foo=1,
  199. route_name=self.add.name,
  200. )
  201. assert type.from_dict(dict(s)) == s
  202. assert repr(s)
  203. class test_chunks(CanvasCase):
  204. def test_chunks(self):
  205. x = self.add.chunks(range(100), 10)
  206. assert dict(chunks.from_dict(dict(x), app=self.app)) == dict(x)
  207. assert x.group()
  208. assert len(x.group().tasks) == 10
  209. x.group = Mock()
  210. gr = x.group.return_value = Mock()
  211. x.apply_async()
  212. gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
  213. gr.apply_async.reset_mock()
  214. x()
  215. gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
  216. self.app.conf.task_always_eager = True
  217. chunks.apply_chunks(app=self.app, **x['kwargs'])
  218. class test_chain(CanvasCase):
  219. def test_clone_preserves_state(self):
  220. x = chain(self.add.s(i, i) for i in range(10))
  221. assert x.clone().tasks == x.tasks
  222. assert x.clone().kwargs == x.kwargs
  223. assert x.clone().args == x.args
  224. def test_repr(self):
  225. x = self.add.s(2, 2) | self.add.s(2)
  226. assert repr(x) == '%s(2, 2) | add(2)' % (self.add.name,)
  227. def test_apply_async(self):
  228. c = self.add.s(2, 2) | self.add.s(4) | self.add.s(8)
  229. result = c.apply_async()
  230. assert result.parent
  231. assert result.parent.parent
  232. assert result.parent.parent.parent is None
  233. def test_splices_chains(self):
  234. c = chain(
  235. self.add.s(5, 5),
  236. chain(self.add.s(6), self.add.s(7), self.add.s(8), app=self.app),
  237. app=self.app,
  238. )
  239. c.freeze()
  240. tasks, _ = c._frozen
  241. assert len(tasks) == 4
  242. def test_from_dict_no_tasks(self):
  243. assert chain.from_dict(dict(chain(app=self.app)), app=self.app)
  244. def test_from_dict_full_subtasks(self):
  245. c = chain(self.add.si(1, 2), self.add.si(3, 4), self.add.si(5, 6))
  246. serialized = json.loads(json.dumps(c))
  247. deserialized = chain.from_dict(serialized)
  248. for task in deserialized.tasks:
  249. assert isinstance(task, Signature)
  250. @pytest.mark.usefixtures('depends_on_current_app')
  251. def test_app_falls_back_to_default(self):
  252. from celery._state import current_app
  253. assert chain().app is current_app
  254. def test_handles_dicts(self):
  255. c = chain(
  256. self.add.s(5, 5), dict(self.add.s(8)), app=self.app,
  257. )
  258. c.freeze()
  259. tasks, _ = c._frozen
  260. for task in tasks:
  261. assert isinstance(task, Signature)
  262. assert task.app is self.app
  263. def test_group_to_chord(self):
  264. c = (
  265. self.add.s(5) |
  266. group([self.add.s(i, i) for i in range(5)], app=self.app) |
  267. self.add.s(10) |
  268. self.add.s(20) |
  269. self.add.s(30)
  270. )
  271. c._use_link = True
  272. tasks, results = c.prepare_steps((), {}, c.tasks)
  273. assert tasks[-1].args[0] == 5
  274. assert isinstance(tasks[-2], chord)
  275. assert len(tasks[-2].tasks) == 5
  276. body = tasks[-2].body
  277. assert len(body.tasks) == 3
  278. assert body.tasks[0].args[0] == 10
  279. assert body.tasks[1].args[0] == 20
  280. assert body.tasks[2].args[0] == 30
  281. c2 = self.add.s(2, 2) | group(self.add.s(i, i) for i in range(10))
  282. c2._use_link = True
  283. tasks2, _ = c2.prepare_steps((), {}, c2.tasks)
  284. assert isinstance(tasks2[0], group)
  285. def test_group_to_chord__protocol_2__or(self):
  286. c = (
  287. group([self.add.s(i, i) for i in range(5)], app=self.app) |
  288. self.add.s(10) |
  289. self.add.s(20) |
  290. self.add.s(30)
  291. )
  292. assert isinstance(c, chord)
  293. def test_group_to_chord__protocol_2(self):
  294. c = chain(
  295. group([self.add.s(i, i) for i in range(5)], app=self.app),
  296. self.add.s(10),
  297. self.add.s(20),
  298. self.add.s(30)
  299. )
  300. assert isinstance(c, chord)
  301. assert isinstance(c.body, _chain)
  302. assert len(c.body.tasks) == 3
  303. c2 = self.add.s(2, 2) | group(self.add.s(i, i) for i in range(10))
  304. c2._use_link = False
  305. tasks2, _ = c2.prepare_steps((), {}, c2.tasks)
  306. assert isinstance(tasks2[0], group)
  307. def test_apply_options(self):
  308. class static(Signature):
  309. def clone(self, *args, **kwargs):
  310. return self
  311. def s(*args, **kwargs):
  312. return static(self.add, args, kwargs, type=self.add, app=self.app)
  313. c = s(2, 2) | s(4) | s(8)
  314. r1 = c.apply_async(task_id='some_id')
  315. assert r1.id == 'some_id'
  316. c.apply_async(group_id='some_group_id')
  317. assert c.tasks[-1].options['group_id'] == 'some_group_id'
  318. c.apply_async(chord='some_chord_id')
  319. assert c.tasks[-1].options['chord'] == 'some_chord_id'
  320. c.apply_async(link=[s(32)])
  321. assert c.tasks[-1].options['link'] == [s(32)]
  322. c.apply_async(link_error=[s('error')])
  323. for task in c.tasks:
  324. assert task.options['link_error'] == [s('error')]
  325. def test_reverse(self):
  326. x = self.add.s(2, 2) | self.add.s(2)
  327. assert isinstance(signature(x), _chain)
  328. assert isinstance(signature(dict(x)), _chain)
  329. def test_always_eager(self):
  330. self.app.conf.task_always_eager = True
  331. assert ~(self.add.s(4, 4) | self.add.s(8)) == 16
  332. def test_chain_always_eager(self):
  333. self.app.conf.task_always_eager = True
  334. from celery import _state
  335. from celery import result
  336. fixture_task_join_will_block = _state.task_join_will_block
  337. try:
  338. _state.task_join_will_block = _state.orig_task_join_will_block
  339. result.task_join_will_block = _state.orig_task_join_will_block
  340. @self.app.task(shared=False)
  341. def chain_add():
  342. return (self.add.s(4, 4) | self.add.s(8)).apply_async()
  343. r = chain_add.apply_async(throw=True).get()
  344. assert r.get() == 16
  345. finally:
  346. _state.task_join_will_block = fixture_task_join_will_block
  347. result.task_join_will_block = fixture_task_join_will_block
  348. def test_apply(self):
  349. x = chain(self.add.s(4, 4), self.add.s(8), self.add.s(10))
  350. res = x.apply()
  351. assert isinstance(res, EagerResult)
  352. assert res.get() == 26
  353. assert res.parent.get() == 16
  354. assert res.parent.parent.get() == 8
  355. assert res.parent.parent.parent is None
  356. def test_kwargs_apply(self):
  357. x = chain(self.add.s(), self.add.s(8), self.add.s(10))
  358. res = x.apply(kwargs={'x': 1, 'y': 1}).get()
  359. assert res == 20
  360. def test_single_expresion(self):
  361. x = chain(self.add.s(1, 2)).apply()
  362. assert x.get() == 3
  363. assert x.parent is None
  364. def test_empty_chain_returns_none(self):
  365. assert chain(app=self.app)() is None
  366. assert chain(app=self.app).apply_async() is None
  367. def test_call_no_tasks(self):
  368. x = chain()
  369. assert not x()
  370. def test_call_with_tasks(self):
  371. x = self.add.s(2, 2) | self.add.s(4)
  372. x.apply_async = Mock()
  373. x(2, 2, foo=1)
  374. x.apply_async.assert_called_with((2, 2), {'foo': 1})
  375. def test_from_dict_no_args__with_args(self):
  376. x = dict(self.add.s(2, 2) | self.add.s(4))
  377. x['args'] = None
  378. assert isinstance(chain.from_dict(x), _chain)
  379. x['args'] = (2,)
  380. assert isinstance(chain.from_dict(x), _chain)
  381. def test_accepts_generator_argument(self):
  382. x = chain(self.add.s(i) for i in range(10))
  383. assert x.tasks[0].type, self.add
  384. assert x.type
  385. def test_chord_sets_result_parent(self):
  386. g = (self.add.s(0, 0) |
  387. group(self.add.s(i, i) for i in range(1, 10)) |
  388. self.add.s(2, 2) |
  389. self.add.s(4, 4))
  390. res = g.freeze()
  391. assert isinstance(res, AsyncResult)
  392. assert not isinstance(res, GroupResult)
  393. assert isinstance(res.parent, AsyncResult)
  394. assert not isinstance(res.parent, GroupResult)
  395. assert isinstance(res.parent.parent, GroupResult)
  396. assert isinstance(res.parent.parent.parent, AsyncResult)
  397. assert not isinstance(res.parent.parent.parent, GroupResult)
  398. assert res.parent.parent.parent.parent is None
  399. seen = set()
  400. node = res
  401. while node:
  402. assert node.id not in seen
  403. seen.add(node.id)
  404. node = node.parent
  405. def test_append_to_empty_chain(self):
  406. x = chain()
  407. x |= self.add.s(1, 1)
  408. x |= self.add.s(1)
  409. x.freeze()
  410. tasks, _ = x._frozen
  411. assert len(tasks) == 2
  412. assert x.apply().get() == 3
  413. class test_group(CanvasCase):
  414. def test_repr(self):
  415. x = group([self.add.s(2, 2), self.add.s(4, 4)])
  416. assert repr(x)
  417. def test_reverse(self):
  418. x = group([self.add.s(2, 2), self.add.s(4, 4)])
  419. assert isinstance(signature(x), group)
  420. assert isinstance(signature(dict(x)), group)
  421. def test_cannot_link_on_group(self):
  422. x = group([self.add.s(2, 2), self.add.s(4, 4)])
  423. with pytest.raises(TypeError):
  424. x.apply_async(link=self.add.s(2, 2))
  425. def test_cannot_link_error_on_group(self):
  426. x = group([self.add.s(2, 2), self.add.s(4, 4)])
  427. with pytest.raises(TypeError):
  428. x.apply_async(link_error=self.add.s(2, 2))
  429. def test_group_with_group_argument(self):
  430. g1 = group(self.add.s(2, 2), self.add.s(4, 4), app=self.app)
  431. g2 = group(g1, app=self.app)
  432. assert g2.tasks is g1.tasks
  433. def test_maybe_group_sig(self):
  434. assert _maybe_group(self.add.s(2, 2), self.app) == [self.add.s(2, 2)]
  435. def test_apply(self):
  436. x = group([self.add.s(4, 4), self.add.s(8, 8)])
  437. res = x.apply()
  438. assert res.get(), [8 == 16]
  439. def test_apply_async(self):
  440. x = group([self.add.s(4, 4), self.add.s(8, 8)])
  441. x.apply_async()
  442. def test_prepare_with_dict(self):
  443. x = group([self.add.s(4, 4), dict(self.add.s(8, 8))], app=self.app)
  444. x.apply_async()
  445. def test_group_in_group(self):
  446. g1 = group(self.add.s(2, 2), self.add.s(4, 4), app=self.app)
  447. g2 = group(self.add.s(8, 8), g1, self.add.s(16, 16), app=self.app)
  448. g2.apply_async()
  449. def test_set_immutable(self):
  450. g1 = group(Mock(name='t1'), Mock(name='t2'), app=self.app)
  451. g1.set_immutable(True)
  452. for task in g1.tasks:
  453. task.set_immutable.assert_called_with(True)
  454. def test_link(self):
  455. g1 = group(Mock(name='t1'), Mock(name='t2'), app=self.app)
  456. sig = Mock(name='sig')
  457. g1.link(sig)
  458. g1.tasks[0].link.assert_called_with(sig.clone().set(immutable=True))
  459. def test_link_error(self):
  460. g1 = group(Mock(name='t1'), Mock(name='t2'), app=self.app)
  461. sig = Mock(name='sig')
  462. g1.link_error(sig)
  463. g1.tasks[0].link_error.assert_called_with(
  464. sig.clone().set(immutable=True),
  465. )
  466. def test_apply_empty(self):
  467. x = group(app=self.app)
  468. x.apply()
  469. res = x.apply_async()
  470. assert res
  471. assert not res.results
  472. def test_apply_async_with_parent(self):
  473. _task_stack.push(self.add)
  474. try:
  475. self.add.push_request(called_directly=False)
  476. try:
  477. assert not self.add.request.children
  478. x = group([self.add.s(4, 4), self.add.s(8, 8)])
  479. res = x()
  480. assert self.add.request.children
  481. assert res in self.add.request.children
  482. assert len(self.add.request.children) == 1
  483. finally:
  484. self.add.pop_request()
  485. finally:
  486. _task_stack.pop()
  487. def test_from_dict(self):
  488. x = group([self.add.s(2, 2), self.add.s(4, 4)])
  489. x['args'] = (2, 2)
  490. assert group.from_dict(dict(x))
  491. x['args'] = None
  492. assert group.from_dict(dict(x))
  493. def test_call_empty_group(self):
  494. x = group(app=self.app)
  495. assert not len(x())
  496. x.delay()
  497. x.apply_async()
  498. x()
  499. def test_skew(self):
  500. g = group([self.add.s(i, i) for i in range(10)])
  501. g.skew(start=1, stop=10, step=1)
  502. for i, task in enumerate(g.tasks):
  503. assert task.options['countdown'] == i + 1
  504. def test_iter(self):
  505. g = group([self.add.s(i, i) for i in range(10)])
  506. assert list(iter(g)) == list(g.keys())
  507. def test_single_task(self):
  508. g = group([self.add.s(1, 1)])
  509. assert isinstance(g, group)
  510. assert len(g.tasks) == 1
  511. g = group(self.add.s(1, 1))
  512. assert isinstance(g, group)
  513. assert len(g.tasks) == 1
  514. @staticmethod
  515. def helper_test_get_delay(result):
  516. import time
  517. t0 = time.time()
  518. while not result.ready():
  519. time.sleep(0.01)
  520. if time.time() - t0 > 1:
  521. return None
  522. return result.get()
  523. def test_kwargs_direct(self):
  524. res = [self.add(x=1, y=1), self.add(x=1, y=1)]
  525. assert res == [2, 2]
  526. def test_kwargs_apply(self):
  527. x = group([self.add.s(), self.add.s()])
  528. res = x.apply(kwargs={'x': 1, 'y': 1}).get()
  529. assert res == [2, 2]
  530. def test_kwargs_apply_async(self):
  531. self.app.conf.task_always_eager = True
  532. x = group([self.add.s(), self.add.s()])
  533. res = self.helper_test_get_delay(
  534. x.apply_async(kwargs={'x': 1, 'y': 1})
  535. )
  536. assert res == [2, 2]
  537. def test_kwargs_delay(self):
  538. self.app.conf.task_always_eager = True
  539. x = group([self.add.s(), self.add.s()])
  540. res = self.helper_test_get_delay(x.delay(x=1, y=1))
  541. assert res == [2, 2]
  542. def test_kwargs_delay_partial(self):
  543. self.app.conf.task_always_eager = True
  544. x = group([self.add.s(1), self.add.s(x=1)])
  545. res = self.helper_test_get_delay(x.delay(y=1))
  546. assert res == [2, 2]
  547. class test_chord(CanvasCase):
  548. def test_reverse(self):
  549. x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
  550. assert isinstance(signature(x), chord)
  551. assert isinstance(signature(dict(x)), chord)
  552. def test_clone_clones_body(self):
  553. x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
  554. y = x.clone()
  555. assert x.kwargs['body'] is not y.kwargs['body']
  556. y.kwargs.pop('body')
  557. z = y.clone()
  558. assert z.kwargs.get('body') is None
  559. def test_argument_is_group(self):
  560. x = chord(group(self.add.s(2, 2), self.add.s(4, 4), app=self.app))
  561. assert x.tasks
  562. def test_app_when_app(self):
  563. app = Mock(name='app')
  564. x = chord([self.add.s(4, 4)], app=app)
  565. assert x.app is app
  566. def test_app_when_app_in_task(self):
  567. t1 = Mock(name='t1')
  568. t2 = Mock(name='t2')
  569. x = chord([t1, self.add.s(4, 4)])
  570. assert x.app is x.tasks[0].app
  571. t1.app = None
  572. x = chord([t1], body=t2)
  573. assert x.app is t2._app
  574. def test_app_when_header_is_empty(self):
  575. x = chord([], self.add.s(4, 4))
  576. assert x.app is self.add.app
  577. @pytest.mark.usefixtures('depends_on_current_app')
  578. def test_app_fallback_to_current(self):
  579. from celery._state import current_app
  580. t1 = Mock(name='t1')
  581. t1.app = t1._app = None
  582. x = chord([t1], body=t1)
  583. assert x.app is current_app
  584. def test_set_immutable(self):
  585. x = chord([Mock(name='t1'), Mock(name='t2')], app=self.app)
  586. x.set_immutable(True)
  587. def test_links_to_body(self):
  588. x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
  589. x.link(self.div.s(2))
  590. assert not x.options.get('link')
  591. assert x.kwargs['body'].options['link']
  592. x.link_error(self.div.s(2))
  593. assert not x.options.get('link_error')
  594. assert x.kwargs['body'].options['link_error']
  595. assert x.tasks
  596. assert x.body
  597. def test_repr(self):
  598. x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
  599. assert repr(x)
  600. x.kwargs['body'] = None
  601. assert 'without body' in repr(x)
  602. def test_freeze_tasks_is_not_group(self):
  603. x = chord([self.add.s(2, 2)], body=self.add.s(), app=self.app)
  604. x.freeze()
  605. x.tasks = [self.add.s(2, 2)]
  606. x.freeze()
  607. def test_chain_always_eager(self):
  608. self.app.conf.task_always_eager = True
  609. from celery import _state
  610. from celery import result
  611. fixture_task_join_will_block = _state.task_join_will_block
  612. try:
  613. _state.task_join_will_block = _state.orig_task_join_will_block
  614. result.task_join_will_block = _state.orig_task_join_will_block
  615. @self.app.task(shared=False)
  616. def finalize(*args):
  617. pass
  618. @self.app.task(shared=False)
  619. def chord_add():
  620. return chord([self.add.s(4, 4)], finalize.s()).apply_async()
  621. chord_add.apply_async(throw=True).get()
  622. finally:
  623. _state.task_join_will_block = fixture_task_join_will_block
  624. result.task_join_will_block = fixture_task_join_will_block
  625. class test_maybe_signature(CanvasCase):
  626. def test_is_None(self):
  627. assert maybe_signature(None, app=self.app) is None
  628. def test_is_dict(self):
  629. assert isinstance(maybe_signature(dict(self.add.s()), app=self.app),
  630. Signature)
  631. def test_when_sig(self):
  632. s = self.add.s()
  633. assert maybe_signature(s, app=self.app) is s