test_base.py 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702
  1. from __future__ import absolute_import, unicode_literals
  2. import sys
  3. import types
  4. from contextlib import contextmanager
  5. import pytest
  6. from case import ANY, Mock, call, patch, skip
  7. from celery import chord, group, states, uuid
  8. from celery.backends.base import (BaseBackend, DisabledBackend,
  9. KeyValueStoreBackend, _nulldict)
  10. from celery.exceptions import ChordError, TimeoutError
  11. from celery.five import bytes_if_py2, items, range
  12. from celery.result import result_from_tuple
  13. from celery.utils import serialization
  14. from celery.utils.functional import pass1
  15. from celery.utils.serialization import UnpickleableExceptionWrapper
  16. from celery.utils.serialization import find_pickleable_exception as fnpe
  17. from celery.utils.serialization import get_pickleable_exception as gpe
  18. from celery.utils.serialization import subclass_exception
  19. class wrapobject(object):
  20. def __init__(self, *args, **kwargs):
  21. self.args = args
  22. if sys.version_info[0] == 3 or getattr(sys, 'pypy_version_info', None):
  23. Oldstyle = None
  24. else:
  25. Oldstyle = types.ClassType(bytes_if_py2('Oldstyle'), (), {})
  26. Unpickleable = subclass_exception(
  27. bytes_if_py2('Unpickleable'), KeyError, 'foo.module',
  28. )
  29. Impossible = subclass_exception(
  30. bytes_if_py2('Impossible'), object, 'foo.module',
  31. )
  32. Lookalike = subclass_exception(
  33. bytes_if_py2('Lookalike'), wrapobject, 'foo.module',
  34. )
  35. class test_nulldict:
  36. def test_nulldict(self):
  37. x = _nulldict()
  38. x['foo'] = 1
  39. x.update(foo=1, bar=2)
  40. x.setdefault('foo', 3)
  41. class test_serialization:
  42. def test_create_exception_cls(self):
  43. assert serialization.create_exception_cls('FooError', 'm')
  44. assert serialization.create_exception_cls('FooError', 'm', KeyError)
  45. class test_BaseBackend_interface:
  46. def setup(self):
  47. self.b = BaseBackend(self.app)
  48. @self.app.task(shared=False)
  49. def callback(result):
  50. pass
  51. self.callback = callback
  52. def test__forget(self):
  53. with pytest.raises(NotImplementedError):
  54. self.b._forget('SOMExx-N0Nex1stant-IDxx-')
  55. def test_forget(self):
  56. with pytest.raises(NotImplementedError):
  57. self.b.forget('SOMExx-N0nex1stant-IDxx-')
  58. def test_on_chord_part_return(self):
  59. self.b.on_chord_part_return(None, None, None)
  60. def test_apply_chord(self, unlock='celery.chord_unlock'):
  61. self.app.tasks[unlock] = Mock()
  62. header_result = self.app.GroupResult(
  63. uuid(),
  64. [self.app.AsyncResult(x) for x in range(3)],
  65. )
  66. self.b.apply_chord(header_result, self.callback.s())
  67. assert self.app.tasks[unlock].apply_async.call_count
  68. def test_chord_unlock_queue(self, unlock='celery.chord_unlock'):
  69. self.app.tasks[unlock] = Mock()
  70. header_result = self.app.GroupResult(
  71. uuid(),
  72. [self.app.AsyncResult(x) for x in range(3)],
  73. )
  74. body = self.callback.s()
  75. self.b.apply_chord(header_result, body)
  76. called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
  77. assert called_kwargs['queue'] is None
  78. self.b.apply_chord(header_result, body.set(queue='test_queue'))
  79. called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
  80. assert called_kwargs['queue'] == 'test_queue'
  81. @self.app.task(shared=False, queue='test_queue_two')
  82. def callback_queue(result):
  83. pass
  84. self.b.apply_chord(header_result, callback_queue.s())
  85. called_kwargs = self.app.tasks[unlock].apply_async.call_args[1]
  86. assert called_kwargs['queue'] == 'test_queue_two'
  87. class test_exception_pickle:
  88. @skip.if_python3(reason='does not support old style classes')
  89. @skip.if_pypy()
  90. def test_oldstyle(self):
  91. assert fnpe(Oldstyle())
  92. def test_BaseException(self):
  93. assert fnpe(Exception()) is None
  94. def test_get_pickleable_exception(self):
  95. exc = Exception('foo')
  96. assert gpe(exc) == exc
  97. def test_unpickleable(self):
  98. assert isinstance(fnpe(Unpickleable()), KeyError)
  99. assert fnpe(Impossible()) is None
  100. class test_prepare_exception:
  101. def setup(self):
  102. self.b = BaseBackend(self.app)
  103. def test_unpickleable(self):
  104. self.b.serializer = 'pickle'
  105. x = self.b.prepare_exception(Unpickleable(1, 2, 'foo'))
  106. assert isinstance(x, KeyError)
  107. y = self.b.exception_to_python(x)
  108. assert isinstance(y, KeyError)
  109. def test_json_exception_arguments(self):
  110. self.b.serializer = 'json'
  111. x = self.b.prepare_exception(Exception(object))
  112. assert x == {
  113. 'exc_message': serialization.ensure_serializable(
  114. (object,), self.b.encode),
  115. 'exc_type': Exception.__name__,
  116. 'exc_module': Exception.__module__}
  117. y = self.b.exception_to_python(x)
  118. assert isinstance(y, Exception)
  119. def test_impossible(self):
  120. self.b.serializer = 'pickle'
  121. x = self.b.prepare_exception(Impossible())
  122. assert isinstance(x, UnpickleableExceptionWrapper)
  123. assert str(x)
  124. y = self.b.exception_to_python(x)
  125. assert y.__class__.__name__ == 'Impossible'
  126. if sys.version_info < (2, 5):
  127. assert y.__class__.__module__
  128. else:
  129. assert y.__class__.__module__ == 'foo.module'
  130. def test_regular(self):
  131. self.b.serializer = 'pickle'
  132. x = self.b.prepare_exception(KeyError('baz'))
  133. assert isinstance(x, KeyError)
  134. y = self.b.exception_to_python(x)
  135. assert isinstance(y, KeyError)
  136. def test_unicode_message(self):
  137. message = u'\u03ac'
  138. x = self.b.prepare_exception(Exception(message))
  139. assert x == {'exc_message': (message,),
  140. 'exc_type': Exception.__name__,
  141. 'exc_module': Exception.__module__}
  142. class KVBackend(KeyValueStoreBackend):
  143. mget_returns_dict = False
  144. def __init__(self, app, *args, **kwargs):
  145. self.db = {}
  146. super(KVBackend, self).__init__(app)
  147. def get(self, key):
  148. return self.db.get(key)
  149. def set(self, key, value):
  150. self.db[key] = value
  151. def mget(self, keys):
  152. if self.mget_returns_dict:
  153. return {key: self.get(key) for key in keys}
  154. else:
  155. return [self.get(k) for k in keys]
  156. def delete(self, key):
  157. self.db.pop(key, None)
  158. class DictBackend(BaseBackend):
  159. def __init__(self, *args, **kwargs):
  160. BaseBackend.__init__(self, *args, **kwargs)
  161. self._data = {'can-delete': {'result': 'foo'}}
  162. def _restore_group(self, group_id):
  163. if group_id == 'exists':
  164. return {'result': 'group'}
  165. def _get_task_meta_for(self, task_id):
  166. if task_id == 'task-exists':
  167. return {'result': 'task'}
  168. def _delete_group(self, group_id):
  169. self._data.pop(group_id, None)
  170. class test_BaseBackend_dict:
  171. def setup(self):
  172. self.b = DictBackend(app=self.app)
  173. @self.app.task(shared=False, bind=True)
  174. def bound_errback(self, result):
  175. pass
  176. @self.app.task(shared=False)
  177. def errback(arg1, arg2):
  178. errback.last_result = arg1 + arg2
  179. self.bound_errback = bound_errback
  180. self.errback = errback
  181. def test_delete_group(self):
  182. self.b.delete_group('can-delete')
  183. assert 'can-delete' not in self.b._data
  184. def test_prepare_exception_json(self):
  185. x = DictBackend(self.app, serializer='json')
  186. e = x.prepare_exception(KeyError('foo'))
  187. assert 'exc_type' in e
  188. e = x.exception_to_python(e)
  189. assert e.__class__.__name__ == 'KeyError'
  190. assert str(e).strip('u') == "'foo'"
  191. def test_save_group(self):
  192. b = BaseBackend(self.app)
  193. b._save_group = Mock()
  194. b.save_group('foofoo', 'xxx')
  195. b._save_group.assert_called_with('foofoo', 'xxx')
  196. def test_add_to_chord_interface(self):
  197. b = BaseBackend(self.app)
  198. with pytest.raises(NotImplementedError):
  199. b.add_to_chord('group_id', 'sig')
  200. def test_forget_interface(self):
  201. b = BaseBackend(self.app)
  202. with pytest.raises(NotImplementedError):
  203. b.forget('foo')
  204. def test_restore_group(self):
  205. assert self.b.restore_group('missing') is None
  206. assert self.b.restore_group('missing') is None
  207. assert self.b.restore_group('exists') == 'group'
  208. assert self.b.restore_group('exists') == 'group'
  209. assert self.b.restore_group('exists', cache=False) == 'group'
  210. def test_reload_group_result(self):
  211. self.b._cache = {}
  212. self.b.reload_group_result('exists')
  213. self.b._cache['exists'] = {'result': 'group'}
  214. def test_reload_task_result(self):
  215. self.b._cache = {}
  216. self.b.reload_task_result('task-exists')
  217. self.b._cache['task-exists'] = {'result': 'task'}
  218. def test_fail_from_current_stack(self):
  219. self.b.mark_as_failure = Mock()
  220. try:
  221. raise KeyError('foo')
  222. except KeyError as exc:
  223. self.b.fail_from_current_stack('task_id')
  224. self.b.mark_as_failure.assert_called()
  225. args = self.b.mark_as_failure.call_args[0]
  226. assert args[0] == 'task_id'
  227. assert args[1] is exc
  228. assert args[2]
  229. def test_prepare_value_serializes_group_result(self):
  230. self.b.serializer = 'json'
  231. g = self.app.GroupResult('group_id', [self.app.AsyncResult('foo')])
  232. v = self.b.prepare_value(g)
  233. assert isinstance(v, (list, tuple))
  234. assert result_from_tuple(v, app=self.app) == g
  235. v2 = self.b.prepare_value(g[0])
  236. assert isinstance(v2, (list, tuple))
  237. assert result_from_tuple(v2, app=self.app) == g[0]
  238. self.b.serializer = 'pickle'
  239. assert isinstance(self.b.prepare_value(g), self.app.GroupResult)
  240. def test_is_cached(self):
  241. b = BaseBackend(app=self.app, max_cached_results=1)
  242. b._cache['foo'] = 1
  243. assert b.is_cached('foo')
  244. assert not b.is_cached('false')
  245. def test_mark_as_done__chord(self):
  246. b = BaseBackend(app=self.app)
  247. b._store_result = Mock()
  248. request = Mock(name='request')
  249. b.on_chord_part_return = Mock()
  250. b.mark_as_done('id', 10, request=request)
  251. b.on_chord_part_return.assert_called_with(request, states.SUCCESS, 10)
  252. def test_mark_as_failure__bound_errback(self):
  253. b = BaseBackend(app=self.app)
  254. b._store_result = Mock()
  255. request = Mock(name='request')
  256. request.errbacks = [
  257. self.bound_errback.subtask(args=[1], immutable=True)]
  258. exc = KeyError()
  259. group = self.patching('celery.backends.base.group')
  260. b.mark_as_failure('id', exc, request=request)
  261. group.assert_called_with(request.errbacks, app=self.app)
  262. group.return_value.apply_async.assert_called_with(
  263. (request.id, ), parent_id=request.id, root_id=request.root_id)
  264. def test_mark_as_failure__errback(self):
  265. b = BaseBackend(app=self.app)
  266. b._store_result = Mock()
  267. request = Mock(name='request')
  268. request.errbacks = [self.errback.subtask(args=[2, 3], immutable=True)]
  269. exc = KeyError()
  270. b.mark_as_failure('id', exc, request=request)
  271. assert self.errback.last_result == 5
  272. def test_mark_as_failure__chord(self):
  273. b = BaseBackend(app=self.app)
  274. b._store_result = Mock()
  275. request = Mock(name='request')
  276. request.errbacks = []
  277. b.on_chord_part_return = Mock()
  278. exc = KeyError()
  279. b.mark_as_failure('id', exc, request=request)
  280. b.on_chord_part_return.assert_called_with(request, states.FAILURE, exc)
  281. def test_mark_as_revoked__chord(self):
  282. b = BaseBackend(app=self.app)
  283. b._store_result = Mock()
  284. request = Mock(name='request')
  285. request.errbacks = []
  286. b.on_chord_part_return = Mock()
  287. b.mark_as_revoked('id', 'revoked', request=request)
  288. b.on_chord_part_return.assert_called_with(request, states.REVOKED, ANY)
  289. def test_chord_error_from_stack_raises(self):
  290. b = BaseBackend(app=self.app)
  291. exc = KeyError()
  292. callback = Mock(name='callback')
  293. callback.options = {'link_error': []}
  294. task = self.app.tasks[callback.task] = Mock()
  295. b.fail_from_current_stack = Mock()
  296. group = self.patching('celery.group')
  297. group.side_effect = exc
  298. b.chord_error_from_stack(callback, exc=ValueError())
  299. task.backend.fail_from_current_stack.assert_called_with(
  300. callback.id, exc=exc)
  301. def test_exception_to_python_when_None(self):
  302. b = BaseBackend(app=self.app)
  303. assert b.exception_to_python(None) is None
  304. def test_wait_for__on_interval(self):
  305. self.patching('time.sleep')
  306. b = BaseBackend(app=self.app)
  307. b._get_task_meta_for = Mock()
  308. b._get_task_meta_for.return_value = {'status': states.PENDING}
  309. callback = Mock(name='callback')
  310. with pytest.raises(TimeoutError):
  311. b.wait_for(task_id='1', on_interval=callback, timeout=1)
  312. callback.assert_called_with()
  313. b._get_task_meta_for.return_value = {'status': states.SUCCESS}
  314. b.wait_for(task_id='1', timeout=None)
  315. def test_get_children(self):
  316. b = BaseBackend(app=self.app)
  317. b._get_task_meta_for = Mock()
  318. b._get_task_meta_for.return_value = {}
  319. assert b.get_children('id') is None
  320. b._get_task_meta_for.return_value = {'children': 3}
  321. assert b.get_children('id') == 3
  322. class test_KeyValueStoreBackend:
  323. def setup(self):
  324. self.b = KVBackend(app=self.app)
  325. def test_on_chord_part_return(self):
  326. assert not self.b.implements_incr
  327. self.b.on_chord_part_return(None, None, None)
  328. def test_get_store_delete_result(self):
  329. tid = uuid()
  330. self.b.mark_as_done(tid, 'Hello world')
  331. assert self.b.get_result(tid) == 'Hello world'
  332. assert self.b.get_state(tid) == states.SUCCESS
  333. self.b.forget(tid)
  334. assert self.b.get_state(tid) == states.PENDING
  335. def test_strip_prefix(self):
  336. x = self.b.get_key_for_task('x1b34')
  337. assert self.b._strip_prefix(x) == 'x1b34'
  338. assert self.b._strip_prefix('x1b34') == 'x1b34'
  339. def test_get_many(self):
  340. for is_dict in True, False:
  341. self.b.mget_returns_dict = is_dict
  342. ids = {uuid(): i for i in range(10)}
  343. for id, i in items(ids):
  344. self.b.mark_as_done(id, i)
  345. it = self.b.get_many(list(ids), interval=0.01)
  346. for i, (got_id, got_state) in enumerate(it):
  347. assert got_state['result'] == ids[got_id]
  348. assert i == 9
  349. assert list(self.b.get_many(list(ids), interval=0.01))
  350. self.b._cache.clear()
  351. callback = Mock(name='callback')
  352. it = self.b.get_many(
  353. list(ids),
  354. on_message=callback,
  355. interval=0.05
  356. )
  357. for i, (got_id, got_state) in enumerate(it):
  358. assert got_state['result'] == ids[got_id]
  359. assert i == 9
  360. assert list(
  361. self.b.get_many(list(ids), interval=0.01)
  362. )
  363. callback.assert_has_calls([
  364. call(ANY) for id in ids
  365. ])
  366. def test_get_many_times_out(self):
  367. tasks = [uuid() for _ in range(4)]
  368. self.b._cache[tasks[1]] = {'status': 'PENDING'}
  369. with pytest.raises(self.b.TimeoutError):
  370. list(self.b.get_many(tasks, timeout=0.01, interval=0.01))
  371. def test_chord_part_return_no_gid(self):
  372. self.b.implements_incr = True
  373. task = Mock()
  374. state = 'SUCCESS'
  375. result = 10
  376. task.request.group = None
  377. self.b.get_key_for_chord = Mock()
  378. self.b.get_key_for_chord.side_effect = AssertionError(
  379. 'should not get here',
  380. )
  381. assert self.b.on_chord_part_return(
  382. task.request, state, result) is None
  383. @patch('celery.backends.base.GroupResult')
  384. @patch('celery.backends.base.maybe_signature')
  385. def test_chord_part_return_restore_raises(self, maybe_signature,
  386. GroupResult):
  387. self.b.implements_incr = True
  388. GroupResult.restore.side_effect = KeyError()
  389. self.b.chord_error_from_stack = Mock()
  390. callback = Mock(name='callback')
  391. request = Mock(name='request')
  392. request.group = 'gid'
  393. maybe_signature.return_value = callback
  394. self.b.on_chord_part_return(request, states.SUCCESS, 10)
  395. self.b.chord_error_from_stack.assert_called_with(
  396. callback, ANY,
  397. )
  398. @patch('celery.backends.base.GroupResult')
  399. @patch('celery.backends.base.maybe_signature')
  400. def test_chord_part_return_restore_empty(self, maybe_signature,
  401. GroupResult):
  402. self.b.implements_incr = True
  403. GroupResult.restore.return_value = None
  404. self.b.chord_error_from_stack = Mock()
  405. callback = Mock(name='callback')
  406. request = Mock(name='request')
  407. request.group = 'gid'
  408. maybe_signature.return_value = callback
  409. self.b.on_chord_part_return(request, states.SUCCESS, 10)
  410. self.b.chord_error_from_stack.assert_called_with(
  411. callback, ANY,
  412. )
  413. def test_filter_ready(self):
  414. self.b.decode_result = Mock()
  415. self.b.decode_result.side_effect = pass1
  416. assert len(list(self.b._filter_ready([
  417. (1, {'status': states.RETRY}),
  418. (2, {'status': states.FAILURE}),
  419. (3, {'status': states.SUCCESS}),
  420. ]))) == 2
  421. @contextmanager
  422. def _chord_part_context(self, b):
  423. @self.app.task(shared=False)
  424. def callback(result):
  425. pass
  426. b.implements_incr = True
  427. b.client = Mock()
  428. with patch('celery.backends.base.GroupResult') as GR:
  429. deps = GR.restore.return_value = Mock(name='DEPS')
  430. deps.__len__ = Mock()
  431. deps.__len__.return_value = 10
  432. b.incr = Mock()
  433. b.incr.return_value = 10
  434. b.expire = Mock()
  435. task = Mock()
  436. task.request.group = 'grid'
  437. cb = task.request.chord = callback.s()
  438. task.request.chord.freeze()
  439. callback.backend = b
  440. callback.backend.fail_from_current_stack = Mock()
  441. yield task, deps, cb
  442. def test_chord_part_return_propagate_set(self):
  443. with self._chord_part_context(self.b) as (task, deps, _):
  444. self.b.on_chord_part_return(task.request, 'SUCCESS', 10)
  445. self.b.expire.assert_not_called()
  446. deps.delete.assert_called_with()
  447. deps.join_native.assert_called_with(propagate=True, timeout=3.0)
  448. def test_chord_part_return_propagate_default(self):
  449. with self._chord_part_context(self.b) as (task, deps, _):
  450. self.b.on_chord_part_return(task.request, 'SUCCESS', 10)
  451. self.b.expire.assert_not_called()
  452. deps.delete.assert_called_with()
  453. deps.join_native.assert_called_with(propagate=True, timeout=3.0)
  454. def test_chord_part_return_join_raises_internal(self):
  455. with self._chord_part_context(self.b) as (task, deps, callback):
  456. deps._failed_join_report = lambda: iter([])
  457. deps.join_native.side_effect = KeyError('foo')
  458. self.b.on_chord_part_return(task.request, 'SUCCESS', 10)
  459. self.b.fail_from_current_stack.assert_called()
  460. args = self.b.fail_from_current_stack.call_args
  461. exc = args[1]['exc']
  462. assert isinstance(exc, ChordError)
  463. assert 'foo' in str(exc)
  464. def test_chord_part_return_join_raises_task(self):
  465. b = KVBackend(serializer='pickle', app=self.app)
  466. with self._chord_part_context(b) as (task, deps, callback):
  467. deps._failed_join_report = lambda: iter([
  468. self.app.AsyncResult('culprit'),
  469. ])
  470. deps.join_native.side_effect = KeyError('foo')
  471. b.on_chord_part_return(task.request, 'SUCCESS', 10)
  472. b.fail_from_current_stack.assert_called()
  473. args = b.fail_from_current_stack.call_args
  474. exc = args[1]['exc']
  475. assert isinstance(exc, ChordError)
  476. assert 'Dependency culprit raised' in str(exc)
  477. def test_restore_group_from_json(self):
  478. b = KVBackend(serializer='json', app=self.app)
  479. g = self.app.GroupResult(
  480. 'group_id',
  481. [self.app.AsyncResult('a'), self.app.AsyncResult('b')],
  482. )
  483. b._save_group(g.id, g)
  484. g2 = b._restore_group(g.id)['result']
  485. assert g2 == g
  486. def test_restore_group_from_pickle(self):
  487. b = KVBackend(serializer='pickle', app=self.app)
  488. g = self.app.GroupResult(
  489. 'group_id',
  490. [self.app.AsyncResult('a'), self.app.AsyncResult('b')],
  491. )
  492. b._save_group(g.id, g)
  493. g2 = b._restore_group(g.id)['result']
  494. assert g2 == g
  495. def test_chord_apply_fallback(self):
  496. self.b.implements_incr = False
  497. self.b.fallback_chord_unlock = Mock()
  498. header_result = self.app.GroupResult(
  499. 'group_id',
  500. [self.app.AsyncResult(x) for x in range(3)],
  501. )
  502. self.b.apply_chord(
  503. header_result, 'body', foo=1,
  504. )
  505. self.b.fallback_chord_unlock.assert_called_with(
  506. header_result, 'body', foo=1,
  507. )
  508. def test_get_missing_meta(self):
  509. assert self.b.get_result('xxx-missing') is None
  510. assert self.b.get_state('xxx-missing') == states.PENDING
  511. def test_save_restore_delete_group(self):
  512. tid = uuid()
  513. tsr = self.app.GroupResult(
  514. tid, [self.app.AsyncResult(uuid()) for _ in range(10)],
  515. )
  516. self.b.save_group(tid, tsr)
  517. self.b.restore_group(tid)
  518. assert self.b.restore_group(tid) == tsr
  519. self.b.delete_group(tid)
  520. assert self.b.restore_group(tid) is None
  521. def test_restore_missing_group(self):
  522. assert self.b.restore_group('xxx-nonexistant') is None
  523. class test_KeyValueStoreBackend_interface:
  524. def test_get(self):
  525. with pytest.raises(NotImplementedError):
  526. KeyValueStoreBackend(self.app).get('a')
  527. def test_set(self):
  528. with pytest.raises(NotImplementedError):
  529. KeyValueStoreBackend(self.app).set('a', 1)
  530. def test_incr(self):
  531. with pytest.raises(NotImplementedError):
  532. KeyValueStoreBackend(self.app).incr('a')
  533. def test_cleanup(self):
  534. assert not KeyValueStoreBackend(self.app).cleanup()
  535. def test_delete(self):
  536. with pytest.raises(NotImplementedError):
  537. KeyValueStoreBackend(self.app).delete('a')
  538. def test_mget(self):
  539. with pytest.raises(NotImplementedError):
  540. KeyValueStoreBackend(self.app).mget(['a'])
  541. def test_forget(self):
  542. with pytest.raises(NotImplementedError):
  543. KeyValueStoreBackend(self.app).forget('a')
  544. class test_DisabledBackend:
  545. def test_store_result(self):
  546. DisabledBackend(self.app).store_result()
  547. def test_is_disabled(self):
  548. with pytest.raises(NotImplementedError):
  549. DisabledBackend(self.app).get_state('foo')
  550. def test_as_uri(self):
  551. assert DisabledBackend(self.app).as_uri() == 'disabled://'
  552. @pytest.mark.celery(result_backend='disabled')
  553. def test_chord_raises_error(self):
  554. with pytest.raises(NotImplementedError):
  555. chord(self.add.s(i, i) for i in range(10))(self.add.s([2]))
  556. @pytest.mark.celery(result_backend='disabled')
  557. def test_chain_with_chord_raises_error(self):
  558. with pytest.raises(NotImplementedError):
  559. (self.add.s(2, 2) |
  560. group(self.add.s(2, 2),
  561. self.add.s(5, 6)) | self.add.s()).delay()
  562. class test_as_uri:
  563. def setup(self):
  564. self.b = BaseBackend(
  565. app=self.app,
  566. url='sch://uuuu:pwpw@hostname.dom'
  567. )
  568. def test_as_uri_include_password(self):
  569. assert self.b.as_uri(True) == self.b.url
  570. def test_as_uri_exclude_password(self):
  571. assert self.b.as_uri() == 'sch://uuuu:**@hostname.dom/'