test_result.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826
  1. from __future__ import absolute_import
  2. from contextlib import contextmanager
  3. from celery import states
  4. from celery.backends.base import SyncBackendMixin
  5. from celery.exceptions import (
  6. ImproperlyConfigured, IncompleteStream, TimeoutError,
  7. )
  8. from celery.five import range
  9. from celery.result import (
  10. AsyncResult,
  11. EagerResult,
  12. ResultSet,
  13. result_from_tuple,
  14. assert_will_not_block,
  15. )
  16. from celery.utils import uuid
  17. from celery.utils.serialization import pickle
  18. from celery.tests.case import (
  19. AppCase, Mock, call, depends_on_current_app, patch,
  20. )
  21. def mock_task(name, state, result):
  22. return dict(id=uuid(), name=name, state=state, result=result)
  23. def save_result(app, task):
  24. traceback = 'Some traceback'
  25. if task['state'] == states.SUCCESS:
  26. app.backend.mark_as_done(task['id'], task['result'])
  27. elif task['state'] == states.RETRY:
  28. app.backend.mark_as_retry(
  29. task['id'], task['result'], traceback=traceback,
  30. )
  31. else:
  32. app.backend.mark_as_failure(
  33. task['id'], task['result'], traceback=traceback,
  34. )
  35. def make_mock_group(app, size=10):
  36. tasks = [mock_task('ts%d' % i, states.SUCCESS, i) for i in range(size)]
  37. [save_result(app, task) for task in tasks]
  38. return [app.AsyncResult(task['id']) for task in tasks]
  39. class test_AsyncResult(AppCase):
  40. def setup(self):
  41. self.app.conf.result_cache_max = 100
  42. self.app.conf.result_serializer = 'pickle'
  43. self.task1 = mock_task('task1', states.SUCCESS, 'the')
  44. self.task2 = mock_task('task2', states.SUCCESS, 'quick')
  45. self.task3 = mock_task('task3', states.FAILURE, KeyError('brown'))
  46. self.task4 = mock_task('task3', states.RETRY, KeyError('red'))
  47. for task in (self.task1, self.task2, self.task3, self.task4):
  48. save_result(self.app, task)
  49. @self.app.task(shared=False)
  50. def mytask():
  51. pass
  52. self.mytask = mytask
  53. @patch('celery.result.task_join_will_block')
  54. def test_assert_will_not_block(self, task_join_will_block):
  55. task_join_will_block.return_value = True
  56. with self.assertRaises(RuntimeError):
  57. assert_will_not_block()
  58. task_join_will_block.return_value = False
  59. assert_will_not_block()
  60. def test_without_id(self):
  61. with self.assertRaises(ValueError):
  62. AsyncResult(None, app=self.app)
  63. def test_compat_properties(self):
  64. x = self.app.AsyncResult('1')
  65. self.assertEqual(x.task_id, x.id)
  66. x.task_id = '2'
  67. self.assertEqual(x.id, '2')
  68. @depends_on_current_app
  69. def test_reduce_direct(self):
  70. x = AsyncResult('1', app=self.app)
  71. fun, args = x.__reduce__()
  72. self.assertEqual(fun(*args), x)
  73. def test_children(self):
  74. x = self.app.AsyncResult('1')
  75. children = [EagerResult(str(i), i, states.SUCCESS) for i in range(3)]
  76. x._cache = {'children': children, 'status': states.SUCCESS}
  77. x.backend = Mock()
  78. self.assertTrue(x.children)
  79. self.assertEqual(len(x.children), 3)
  80. def test_propagates_for_parent(self):
  81. x = self.app.AsyncResult(uuid())
  82. x.backend = Mock(name='backend')
  83. x.backend.get_task_meta.return_value = {}
  84. x.backend.wait_for_pending.return_value = 84
  85. x.parent = EagerResult(uuid(), KeyError('foo'), states.FAILURE)
  86. with self.assertRaises(KeyError):
  87. x.get(propagate=True)
  88. self.assertFalse(x.backend.wait_for_pending.called)
  89. x.parent = EagerResult(uuid(), 42, states.SUCCESS)
  90. self.assertEqual(x.get(propagate=True), 84)
  91. self.assertTrue(x.backend.wait_for_pending.called)
  92. def test_get_children(self):
  93. tid = uuid()
  94. x = self.app.AsyncResult(tid)
  95. child = [self.app.AsyncResult(uuid()).as_tuple()
  96. for i in range(10)]
  97. x._cache = {'children': child}
  98. self.assertTrue(x.children)
  99. self.assertEqual(len(x.children), 10)
  100. x._cache = {'status': states.SUCCESS}
  101. x.backend._cache[tid] = {'result': None}
  102. self.assertIsNone(x.children)
  103. def test_build_graph_get_leaf_collect(self):
  104. x = self.app.AsyncResult('1')
  105. x.backend._cache['1'] = {'status': states.SUCCESS, 'result': None}
  106. c = [EagerResult(str(i), i, states.SUCCESS) for i in range(3)]
  107. x.iterdeps = Mock()
  108. x.iterdeps.return_value = (
  109. (None, x),
  110. (x, c[0]),
  111. (c[0], c[1]),
  112. (c[1], c[2])
  113. )
  114. x.backend.READY_STATES = states.READY_STATES
  115. self.assertTrue(x.graph)
  116. self.assertIs(x.get_leaf(), 2)
  117. it = x.collect()
  118. self.assertListEqual(list(it), [
  119. (x, None),
  120. (c[0], 0),
  121. (c[1], 1),
  122. (c[2], 2),
  123. ])
  124. def test_iterdeps(self):
  125. x = self.app.AsyncResult('1')
  126. c = [EagerResult(str(i), i, states.SUCCESS) for i in range(3)]
  127. x._cache = {'status': states.SUCCESS, 'result': None, 'children': c}
  128. for child in c:
  129. child.backend = Mock()
  130. child.backend.get_children.return_value = []
  131. it = x.iterdeps()
  132. self.assertListEqual(list(it), [
  133. (None, x),
  134. (x, c[0]),
  135. (x, c[1]),
  136. (x, c[2]),
  137. ])
  138. x._cache = None
  139. x.ready = Mock()
  140. x.ready.return_value = False
  141. with self.assertRaises(IncompleteStream):
  142. list(x.iterdeps())
  143. list(x.iterdeps(intermediate=True))
  144. def test_eq_not_implemented(self):
  145. self.assertFalse(self.app.AsyncResult('1') == object())
  146. @depends_on_current_app
  147. def test_reduce(self):
  148. a1 = self.app.AsyncResult('uuid')
  149. restored = pickle.loads(pickle.dumps(a1))
  150. self.assertEqual(restored.id, 'uuid')
  151. a2 = self.app.AsyncResult('uuid')
  152. self.assertEqual(pickle.loads(pickle.dumps(a2)).id, 'uuid')
  153. def test_maybe_set_cache_empty(self):
  154. self.app.AsyncResult('uuid')._maybe_set_cache(None)
  155. def test_set_cache__children(self):
  156. r1 = self.app.AsyncResult('id1')
  157. r2 = self.app.AsyncResult('id2')
  158. r1._set_cache({'children': [r2.as_tuple()]})
  159. self.assertIn(r2, r1.children)
  160. def test_successful(self):
  161. ok_res = self.app.AsyncResult(self.task1['id'])
  162. nok_res = self.app.AsyncResult(self.task3['id'])
  163. nok_res2 = self.app.AsyncResult(self.task4['id'])
  164. self.assertTrue(ok_res.successful())
  165. self.assertFalse(nok_res.successful())
  166. self.assertFalse(nok_res2.successful())
  167. pending_res = self.app.AsyncResult(uuid())
  168. self.assertFalse(pending_res.successful())
  169. def test_str(self):
  170. ok_res = self.app.AsyncResult(self.task1['id'])
  171. ok2_res = self.app.AsyncResult(self.task2['id'])
  172. nok_res = self.app.AsyncResult(self.task3['id'])
  173. self.assertEqual(str(ok_res), self.task1['id'])
  174. self.assertEqual(str(ok2_res), self.task2['id'])
  175. self.assertEqual(str(nok_res), self.task3['id'])
  176. pending_id = uuid()
  177. pending_res = self.app.AsyncResult(pending_id)
  178. self.assertEqual(str(pending_res), pending_id)
  179. def test_repr(self):
  180. ok_res = self.app.AsyncResult(self.task1['id'])
  181. ok2_res = self.app.AsyncResult(self.task2['id'])
  182. nok_res = self.app.AsyncResult(self.task3['id'])
  183. self.assertEqual(repr(ok_res), '<AsyncResult: %s>' % (
  184. self.task1['id']))
  185. self.assertEqual(repr(ok2_res), '<AsyncResult: %s>' % (
  186. self.task2['id']))
  187. self.assertEqual(repr(nok_res), '<AsyncResult: %s>' % (
  188. self.task3['id']))
  189. pending_id = uuid()
  190. pending_res = self.app.AsyncResult(pending_id)
  191. self.assertEqual(repr(pending_res), '<AsyncResult: %s>' % (
  192. pending_id))
  193. def test_hash(self):
  194. self.assertEqual(hash(self.app.AsyncResult('x0w991')),
  195. hash(self.app.AsyncResult('x0w991')))
  196. self.assertNotEqual(hash(self.app.AsyncResult('x0w991')),
  197. hash(self.app.AsyncResult('x1w991')))
  198. def test_get_traceback(self):
  199. ok_res = self.app.AsyncResult(self.task1['id'])
  200. nok_res = self.app.AsyncResult(self.task3['id'])
  201. nok_res2 = self.app.AsyncResult(self.task4['id'])
  202. self.assertFalse(ok_res.traceback)
  203. self.assertTrue(nok_res.traceback)
  204. self.assertTrue(nok_res2.traceback)
  205. pending_res = self.app.AsyncResult(uuid())
  206. self.assertFalse(pending_res.traceback)
  207. def test_get__backend_gives_None(self):
  208. res = self.app.AsyncResult(self.task1['id'])
  209. res.backend.wait_for = Mock(name='wait_for')
  210. res.backend.wait_for.return_value = None
  211. self.assertIsNone(res.get())
  212. def test_get(self):
  213. ok_res = self.app.AsyncResult(self.task1['id'])
  214. ok2_res = self.app.AsyncResult(self.task2['id'])
  215. nok_res = self.app.AsyncResult(self.task3['id'])
  216. nok2_res = self.app.AsyncResult(self.task4['id'])
  217. callback = Mock(name='callback')
  218. self.assertEqual(ok_res.get(callback=callback), 'the')
  219. callback.assert_called_with(ok_res.id, 'the')
  220. self.assertEqual(ok2_res.get(), 'quick')
  221. with self.assertRaises(KeyError):
  222. nok_res.get()
  223. self.assertTrue(nok_res.get(propagate=False))
  224. self.assertIsInstance(nok2_res.result, KeyError)
  225. self.assertEqual(ok_res.info, 'the')
  226. def test_eq_ne(self):
  227. r1 = self.app.AsyncResult(self.task1['id'])
  228. r2 = self.app.AsyncResult(self.task1['id'])
  229. r3 = self.app.AsyncResult(self.task2['id'])
  230. self.assertEqual(r1, r2)
  231. self.assertNotEqual(r1, r3)
  232. self.assertEqual(r1, r2.id)
  233. self.assertNotEqual(r1, r3.id)
  234. @depends_on_current_app
  235. def test_reduce_restore(self):
  236. r1 = self.app.AsyncResult(self.task1['id'])
  237. fun, args = r1.__reduce__()
  238. self.assertEqual(fun(*args), r1)
  239. def test_get_timeout(self):
  240. res = self.app.AsyncResult(self.task4['id']) # has RETRY state
  241. with self.assertRaises(TimeoutError):
  242. res.get(timeout=0.001)
  243. pending_res = self.app.AsyncResult(uuid())
  244. with patch('celery.result.time') as _time:
  245. with self.assertRaises(TimeoutError):
  246. pending_res.get(timeout=0.001, interval=0.001)
  247. _time.sleep.assert_called_with(0.001)
  248. def test_get_timeout_longer(self):
  249. res = self.app.AsyncResult(self.task4['id']) # has RETRY state
  250. with patch('celery.result.time') as _time:
  251. with self.assertRaises(TimeoutError):
  252. res.get(timeout=1, interval=1)
  253. _time.sleep.assert_called_with(1)
  254. def test_ready(self):
  255. oks = (self.app.AsyncResult(self.task1['id']),
  256. self.app.AsyncResult(self.task2['id']),
  257. self.app.AsyncResult(self.task3['id']))
  258. self.assertTrue(all(result.ready() for result in oks))
  259. self.assertFalse(self.app.AsyncResult(self.task4['id']).ready())
  260. self.assertFalse(self.app.AsyncResult(uuid()).ready())
  261. class test_ResultSet(AppCase):
  262. def test_resultset_repr(self):
  263. self.assertTrue(repr(self.app.ResultSet(
  264. [self.app.AsyncResult(t) for t in ['1', '2', '3']])))
  265. def test_eq_other(self):
  266. self.assertFalse(self.app.ResultSet(
  267. [self.app.AsyncResult(t) for t in [1, 3, 3]]) == 1)
  268. rs1 = self.app.ResultSet([self.app.AsyncResult(1)])
  269. rs2 = self.app.ResultSet([self.app.AsyncResult(1)])
  270. self.assertTrue(rs1 == rs2)
  271. def test_get(self):
  272. x = self.app.ResultSet([self.app.AsyncResult(t) for t in [1, 2, 3]])
  273. b = x.results[0].backend = Mock()
  274. b.supports_native_join = False
  275. x.join_native = Mock()
  276. x.join = Mock()
  277. x.get()
  278. self.assertTrue(x.join.called)
  279. b.supports_native_join = True
  280. x.get()
  281. self.assertTrue(x.join_native.called)
  282. def test_eq_ne(self):
  283. g1 = self.app.ResultSet([
  284. self.app.AsyncResult('id1'),
  285. self.app.AsyncResult('id2'),
  286. ])
  287. g2 = self.app.ResultSet([
  288. self.app.AsyncResult('id1'),
  289. self.app.AsyncResult('id2'),
  290. ])
  291. g3 = self.app.ResultSet([
  292. self.app.AsyncResult('id3'),
  293. self.app.AsyncResult('id1'),
  294. ])
  295. self.assertEqual(g1, g2)
  296. self.assertNotEqual(g1, g3)
  297. self.assertNotEqual(g1, object())
  298. def test_takes_app_from_first_task(self):
  299. x = ResultSet([self.app.AsyncResult('id1')])
  300. self.assertIs(x.app, x.results[0].app)
  301. x.app = self.app
  302. self.assertIs(x.app, self.app)
  303. def test_get_empty(self):
  304. x = self.app.ResultSet([])
  305. self.assertIsNone(x.supports_native_join)
  306. x.join = Mock(name='join')
  307. x.get()
  308. self.assertTrue(x.join.called)
  309. def test_add(self):
  310. x = self.app.ResultSet([self.app.AsyncResult(1)])
  311. x.add(self.app.AsyncResult(2))
  312. self.assertEqual(len(x), 2)
  313. x.add(self.app.AsyncResult(2))
  314. self.assertEqual(len(x), 2)
  315. @contextmanager
  316. def dummy_copy(self):
  317. with patch('celery.result.copy') as copy:
  318. def passt(arg):
  319. return arg
  320. copy.side_effect = passt
  321. yield
  322. def test_iterate_respects_subpolling_interval(self):
  323. r1 = self.app.AsyncResult(uuid())
  324. r2 = self.app.AsyncResult(uuid())
  325. backend = r1.backend = r2.backend = Mock()
  326. backend.subpolling_interval = 10
  327. ready = r1.ready = r2.ready = Mock()
  328. def se(*args, **kwargs):
  329. ready.side_effect = KeyError()
  330. return False
  331. ready.return_value = False
  332. ready.side_effect = se
  333. x = self.app.ResultSet([r1, r2])
  334. with self.dummy_copy():
  335. with patch('celery.result.time') as _time:
  336. with self.assertPendingDeprecation():
  337. with self.assertRaises(KeyError):
  338. list(x.iterate())
  339. _time.sleep.assert_called_with(10)
  340. backend.subpolling_interval = 0
  341. with patch('celery.result.time') as _time:
  342. with self.assertPendingDeprecation():
  343. with self.assertRaises(KeyError):
  344. ready.return_value = False
  345. ready.side_effect = se
  346. list(x.iterate())
  347. self.assertFalse(_time.sleep.called)
  348. def test_times_out(self):
  349. r1 = self.app.AsyncResult(uuid)
  350. r1.ready = Mock()
  351. r1.ready.return_value = False
  352. x = self.app.ResultSet([r1])
  353. with self.dummy_copy():
  354. with patch('celery.result.time'):
  355. with self.assertPendingDeprecation():
  356. with self.assertRaises(TimeoutError):
  357. list(x.iterate(timeout=1))
  358. def test_add_discard(self):
  359. x = self.app.ResultSet([])
  360. x.add(self.app.AsyncResult('1'))
  361. self.assertIn(self.app.AsyncResult('1'), x.results)
  362. x.discard(self.app.AsyncResult('1'))
  363. x.discard(self.app.AsyncResult('1'))
  364. x.discard('1')
  365. self.assertNotIn(self.app.AsyncResult('1'), x.results)
  366. x.update([self.app.AsyncResult('2')])
  367. def test_clear(self):
  368. x = self.app.ResultSet([])
  369. r = x.results
  370. x.clear()
  371. self.assertIs(x.results, r)
  372. class MockAsyncResultFailure(AsyncResult):
  373. @property
  374. def result(self):
  375. return KeyError('baz')
  376. @property
  377. def state(self):
  378. return states.FAILURE
  379. def get(self, propagate=True, **kwargs):
  380. if propagate:
  381. raise self.result
  382. return self.result
  383. class MockAsyncResultSuccess(AsyncResult):
  384. forgotten = False
  385. def forget(self):
  386. self.forgotten = True
  387. @property
  388. def result(self):
  389. return 42
  390. @property
  391. def state(self):
  392. return states.SUCCESS
  393. def get(self, **kwargs):
  394. return self.result
  395. class SimpleBackend(SyncBackendMixin):
  396. ids = []
  397. def __init__(self, ids=[]):
  398. self.ids = ids
  399. def get_many(self, *args, **kwargs):
  400. return ((id, {'result': i, 'status': states.SUCCESS})
  401. for i, id in enumerate(self.ids))
  402. class test_GroupResult(AppCase):
  403. def setup(self):
  404. self.size = 10
  405. self.ts = self.app.GroupResult(
  406. uuid(), make_mock_group(self.app, self.size),
  407. )
  408. @depends_on_current_app
  409. def test_is_pickleable(self):
  410. ts = self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  411. self.assertEqual(pickle.loads(pickle.dumps(ts)), ts)
  412. ts2 = self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  413. self.assertEqual(pickle.loads(pickle.dumps(ts2)), ts2)
  414. @depends_on_current_app
  415. def test_reduce(self):
  416. ts = self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  417. fun, args = ts.__reduce__()
  418. ts2 = fun(*args)
  419. self.assertEqual(ts2.id, ts.id)
  420. self.assertEqual(ts, ts2)
  421. def test_eq_ne(self):
  422. ts = self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  423. ts2 = self.app.GroupResult(ts.id, ts.results)
  424. ts3 = self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  425. ts4 = self.app.GroupResult(ts.id, [self.app.AsyncResult(uuid())])
  426. self.assertEqual(ts, ts2)
  427. self.assertNotEqual(ts, ts3)
  428. self.assertNotEqual(ts, ts4)
  429. self.assertNotEqual(ts, object())
  430. def test_len(self):
  431. self.assertEqual(len(self.ts), self.size)
  432. def test_eq_other(self):
  433. self.assertFalse(self.ts == 1)
  434. @depends_on_current_app
  435. def test_pickleable(self):
  436. self.assertTrue(pickle.loads(pickle.dumps(self.ts)))
  437. def test_iterate_raises(self):
  438. ar = MockAsyncResultFailure(uuid(), app=self.app)
  439. ts = self.app.GroupResult(uuid(), [ar])
  440. with self.assertPendingDeprecation():
  441. it = ts.iterate()
  442. with self.assertRaises(KeyError):
  443. next(it)
  444. def test_forget(self):
  445. subs = [MockAsyncResultSuccess(uuid(), app=self.app),
  446. MockAsyncResultSuccess(uuid(), app=self.app)]
  447. ts = self.app.GroupResult(uuid(), subs)
  448. ts.forget()
  449. for sub in subs:
  450. self.assertTrue(sub.forgotten)
  451. def test_getitem(self):
  452. subs = [MockAsyncResultSuccess(uuid(), app=self.app),
  453. MockAsyncResultSuccess(uuid(), app=self.app)]
  454. ts = self.app.GroupResult(uuid(), subs)
  455. self.assertIs(ts[0], subs[0])
  456. def test_save_restore(self):
  457. subs = [MockAsyncResultSuccess(uuid(), app=self.app),
  458. MockAsyncResultSuccess(uuid(), app=self.app)]
  459. ts = self.app.GroupResult(uuid(), subs)
  460. ts.save()
  461. with self.assertRaises(AttributeError):
  462. ts.save(backend=object())
  463. self.assertEqual(self.app.GroupResult.restore(ts.id).results,
  464. ts.results)
  465. ts.delete()
  466. self.assertIsNone(self.app.GroupResult.restore(ts.id))
  467. with self.assertRaises(AttributeError):
  468. self.app.GroupResult.restore(ts.id, backend=object())
  469. def test_join_native(self):
  470. backend = SimpleBackend()
  471. results = [self.app.AsyncResult(uuid(), backend=backend)
  472. for i in range(10)]
  473. ts = self.app.GroupResult(uuid(), results)
  474. ts.app.backend = backend
  475. backend.ids = [result.id for result in results]
  476. res = ts.join_native()
  477. self.assertEqual(res, list(range(10)))
  478. callback = Mock(name='callback')
  479. self.assertFalse(ts.join_native(callback=callback))
  480. callback.assert_has_calls([
  481. call(r.id, i) for i, r in enumerate(ts.results)
  482. ])
  483. def test_join_native_raises(self):
  484. ts = self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  485. ts.iter_native = Mock()
  486. ts.iter_native.return_value = iter([
  487. (uuid(), {'status': states.FAILURE, 'result': KeyError()})
  488. ])
  489. with self.assertRaises(KeyError):
  490. ts.join_native(propagate=True)
  491. def test_failed_join_report(self):
  492. res = Mock()
  493. ts = self.app.GroupResult(uuid(), [res])
  494. res.state = states.FAILURE
  495. res.backend.is_cached.return_value = True
  496. self.assertIs(next(ts._failed_join_report()), res)
  497. res.backend.is_cached.return_value = False
  498. with self.assertRaises(StopIteration):
  499. next(ts._failed_join_report())
  500. def test_repr(self):
  501. self.assertTrue(repr(
  502. self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  503. ))
  504. def test_children_is_results(self):
  505. ts = self.app.GroupResult(uuid(), [self.app.AsyncResult(uuid())])
  506. self.assertIs(ts.children, ts.results)
  507. def test_iter_native(self):
  508. backend = SimpleBackend()
  509. results = [self.app.AsyncResult(uuid(), backend=backend)
  510. for i in range(10)]
  511. ts = self.app.GroupResult(uuid(), results)
  512. ts.app.backend = backend
  513. backend.ids = [result.id for result in results]
  514. self.assertEqual(len(list(ts.iter_native())), 10)
  515. def test_iterate_yields(self):
  516. ar = MockAsyncResultSuccess(uuid(), app=self.app)
  517. ar2 = MockAsyncResultSuccess(uuid(), app=self.app)
  518. ts = self.app.GroupResult(uuid(), [ar, ar2])
  519. with self.assertPendingDeprecation():
  520. it = ts.iterate()
  521. self.assertEqual(next(it), 42)
  522. self.assertEqual(next(it), 42)
  523. def test_iterate_eager(self):
  524. ar1 = EagerResult(uuid(), 42, states.SUCCESS)
  525. ar2 = EagerResult(uuid(), 42, states.SUCCESS)
  526. ts = self.app.GroupResult(uuid(), [ar1, ar2])
  527. with self.assertPendingDeprecation():
  528. it = ts.iterate()
  529. self.assertEqual(next(it), 42)
  530. self.assertEqual(next(it), 42)
  531. def test_join_timeout(self):
  532. ar = MockAsyncResultSuccess(uuid(), app=self.app)
  533. ar2 = MockAsyncResultSuccess(uuid(), app=self.app)
  534. ar3 = self.app.AsyncResult(uuid())
  535. ts = self.app.GroupResult(uuid(), [ar, ar2, ar3])
  536. with self.assertRaises(TimeoutError):
  537. ts.join(timeout=0.0000001)
  538. ar4 = self.app.AsyncResult(uuid())
  539. ar4.get = Mock()
  540. ts2 = self.app.GroupResult(uuid(), [ar4])
  541. self.assertTrue(ts2.join(timeout=0.1))
  542. callback = Mock(name='callback')
  543. self.assertFalse(ts2.join(timeout=0.1, callback=callback))
  544. callback.assert_called_with(ar4.id, ar4.get())
  545. def test_iter_native_when_empty_group(self):
  546. ts = self.app.GroupResult(uuid(), [])
  547. self.assertListEqual(list(ts.iter_native()), [])
  548. def test_iterate_simple(self):
  549. with self.assertPendingDeprecation():
  550. it = self.ts.iterate()
  551. results = sorted(list(it))
  552. self.assertListEqual(results, list(range(self.size)))
  553. def test___iter__(self):
  554. self.assertListEqual(list(iter(self.ts)), self.ts.results)
  555. def test_join(self):
  556. joined = self.ts.join()
  557. self.assertListEqual(joined, list(range(self.size)))
  558. def test_successful(self):
  559. self.assertTrue(self.ts.successful())
  560. def test_failed(self):
  561. self.assertFalse(self.ts.failed())
  562. def test_maybe_throw(self):
  563. self.ts.results = [Mock(name='r1')]
  564. self.ts.maybe_throw()
  565. self.ts.results[0].maybe_throw.assert_called_with(
  566. callback=None, propagate=True,
  567. )
  568. def test_join__on_message(self):
  569. with self.assertRaises(ImproperlyConfigured):
  570. self.ts.join(on_message=Mock())
  571. def test_waiting(self):
  572. self.assertFalse(self.ts.waiting())
  573. def test_ready(self):
  574. self.assertTrue(self.ts.ready())
  575. def test_completed_count(self):
  576. self.assertEqual(self.ts.completed_count(), len(self.ts))
  577. class test_pending_AsyncResult(AppCase):
  578. def setup(self):
  579. self.task = self.app.AsyncResult(uuid())
  580. def test_result(self):
  581. self.assertIsNone(self.task.result)
  582. class test_failed_AsyncResult(test_GroupResult):
  583. def setup(self):
  584. self.app.conf.result_serializer = 'pickle'
  585. self.size = 11
  586. results = make_mock_group(self.app, 10)
  587. failed = mock_task('ts11', states.FAILURE, KeyError('Baz'))
  588. save_result(self.app, failed)
  589. failed_res = self.app.AsyncResult(failed['id'])
  590. self.ts = self.app.GroupResult(uuid(), results + [failed_res])
  591. def test_completed_count(self):
  592. self.assertEqual(self.ts.completed_count(), len(self.ts) - 1)
  593. def test_iterate_simple(self):
  594. with self.assertPendingDeprecation():
  595. it = self.ts.iterate()
  596. def consume():
  597. return list(it)
  598. with self.assertRaises(KeyError):
  599. consume()
  600. def test_join(self):
  601. with self.assertRaises(KeyError):
  602. self.ts.join()
  603. def test_successful(self):
  604. self.assertFalse(self.ts.successful())
  605. def test_failed(self):
  606. self.assertTrue(self.ts.failed())
  607. class test_pending_Group(AppCase):
  608. def setup(self):
  609. self.ts = self.app.GroupResult(
  610. uuid(), [self.app.AsyncResult(uuid()),
  611. self.app.AsyncResult(uuid())])
  612. def test_completed_count(self):
  613. self.assertEqual(self.ts.completed_count(), 0)
  614. def test_ready(self):
  615. self.assertFalse(self.ts.ready())
  616. def test_waiting(self):
  617. self.assertTrue(self.ts.waiting())
  618. def x_join(self):
  619. with self.assertRaises(TimeoutError):
  620. self.ts.join(timeout=0.001)
  621. def x_join_longer(self):
  622. with self.assertRaises(TimeoutError):
  623. self.ts.join(timeout=1)
  624. class test_EagerResult(AppCase):
  625. def setup(self):
  626. @self.app.task(shared=False)
  627. def raising(x, y):
  628. raise KeyError(x, y)
  629. self.raising = raising
  630. def test_wait_raises(self):
  631. res = self.raising.apply(args=[3, 3])
  632. with self.assertRaises(KeyError):
  633. res.wait()
  634. self.assertTrue(res.wait(propagate=False))
  635. def test_wait(self):
  636. res = EagerResult('x', 'x', states.RETRY)
  637. res.wait()
  638. self.assertEqual(res.state, states.RETRY)
  639. self.assertEqual(res.status, states.RETRY)
  640. def test_forget(self):
  641. res = EagerResult('x', 'x', states.RETRY)
  642. res.forget()
  643. def test_revoke(self):
  644. res = self.raising.apply(args=[3, 3])
  645. self.assertFalse(res.revoke())
  646. class test_tuples(AppCase):
  647. def test_AsyncResult(self):
  648. x = self.app.AsyncResult(uuid())
  649. self.assertEqual(x, result_from_tuple(x.as_tuple(), self.app))
  650. self.assertEqual(x, result_from_tuple(x, self.app))
  651. def test_with_parent(self):
  652. x = self.app.AsyncResult(uuid())
  653. x.parent = self.app.AsyncResult(uuid())
  654. y = result_from_tuple(x.as_tuple(), self.app)
  655. self.assertEqual(y, x)
  656. self.assertEqual(y.parent, x.parent)
  657. self.assertIsInstance(y.parent, AsyncResult)
  658. def test_compat(self):
  659. uid = uuid()
  660. x = result_from_tuple([uid, []], app=self.app)
  661. self.assertEqual(x.id, uid)
  662. def test_GroupResult(self):
  663. x = self.app.GroupResult(
  664. uuid(), [self.app.AsyncResult(uuid()) for _ in range(10)],
  665. )
  666. self.assertEqual(x, result_from_tuple(x.as_tuple(), self.app))
  667. self.assertEqual(x, result_from_tuple(x, self.app))