test_amqp.py 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273
  1. from __future__ import absolute_import, unicode_literals
  2. import pickle
  3. import pytest
  4. from contextlib import contextmanager
  5. from datetime import timedelta
  6. from pickle import dumps, loads
  7. from case import Mock, mock
  8. from billiard.einfo import ExceptionInfo
  9. from celery import states
  10. from celery import uuid
  11. from celery.backends.amqp import AMQPBackend
  12. from celery.five import Empty, Queue, range
  13. from celery.result import AsyncResult
  14. class SomeClass(object):
  15. def __init__(self, data):
  16. self.data = data
  17. class test_AMQPBackend:
  18. def setup(self):
  19. self.app.conf.result_cache_max = 100
  20. def create_backend(self, **opts):
  21. opts = dict(dict(serializer='pickle', persistent=True), **opts)
  22. return AMQPBackend(self.app, **opts)
  23. def test_destination_for(self):
  24. b = self.create_backend()
  25. request = Mock()
  26. assert b.destination_for('id', request) == (
  27. b.rkey('id'), request.correlation_id,
  28. )
  29. def test_store_result__no_routing_key(self):
  30. b = self.create_backend()
  31. b.destination_for = Mock()
  32. b.destination_for.return_value = None, None
  33. b.store_result('id', None, states.SUCCESS)
  34. def test_mark_as_done(self):
  35. tb1 = self.create_backend(max_cached_results=1)
  36. tb2 = self.create_backend(max_cached_results=1)
  37. tid = uuid()
  38. tb1.mark_as_done(tid, 42)
  39. assert tb2.get_state(tid) == states.SUCCESS
  40. assert tb2.get_result(tid) == 42
  41. assert tb2._cache.get(tid)
  42. assert tb2.get_result(tid), 42
  43. @pytest.mark.usefixtures('depends_on_current_app')
  44. def test_pickleable(self):
  45. assert loads(dumps(self.create_backend()))
  46. def test_revive(self):
  47. tb = self.create_backend()
  48. tb.revive(None)
  49. def test_is_pickled(self):
  50. tb1 = self.create_backend()
  51. tb2 = self.create_backend()
  52. tid2 = uuid()
  53. result = {'foo': 'baz', 'bar': SomeClass(12345)}
  54. tb1.mark_as_done(tid2, result)
  55. # is serialized properly.
  56. rindb = tb2.get_result(tid2)
  57. assert rindb.get('foo') == 'baz'
  58. assert rindb.get('bar').data == 12345
  59. def test_mark_as_failure(self):
  60. tb1 = self.create_backend()
  61. tb2 = self.create_backend()
  62. tid3 = uuid()
  63. try:
  64. raise KeyError('foo')
  65. except KeyError as exception:
  66. einfo = ExceptionInfo()
  67. tb1.mark_as_failure(tid3, exception, traceback=einfo.traceback)
  68. assert tb2.get_state(tid3) == states.FAILURE
  69. assert isinstance(tb2.get_result(tid3), KeyError)
  70. assert tb2.get_traceback(tid3) == einfo.traceback
  71. def test_repair_uuid(self):
  72. from celery.backends.amqp import repair_uuid
  73. for i in range(10):
  74. tid = uuid()
  75. assert repair_uuid(tid.replace('-', '')) == tid
  76. def test_expires_is_int(self):
  77. b = self.create_backend(expires=48)
  78. assert b.queue_arguments.get('x-expires') == 48 * 1000.0
  79. def test_expires_is_float(self):
  80. b = self.create_backend(expires=48.3)
  81. assert b.queue_arguments.get('x-expires') == 48.3 * 1000.0
  82. def test_expires_is_timedelta(self):
  83. b = self.create_backend(expires=timedelta(minutes=1))
  84. assert b.queue_arguments.get('x-expires') == 60 * 1000.0
  85. @mock.sleepdeprived()
  86. def test_store_result_retries(self):
  87. iterations = [0]
  88. stop_raising_at = [5]
  89. def publish(*args, **kwargs):
  90. if iterations[0] > stop_raising_at[0]:
  91. return
  92. iterations[0] += 1
  93. raise KeyError('foo')
  94. backend = AMQPBackend(self.app)
  95. from celery.app.amqp import Producer
  96. prod, Producer.publish = Producer.publish, publish
  97. try:
  98. with pytest.raises(KeyError):
  99. backend.retry_policy['max_retries'] = None
  100. backend.store_result('foo', 'bar', 'STARTED')
  101. with pytest.raises(KeyError):
  102. backend.retry_policy['max_retries'] = 10
  103. backend.store_result('foo', 'bar', 'STARTED')
  104. finally:
  105. Producer.publish = prod
  106. def test_poll_no_messages(self):
  107. b = self.create_backend()
  108. assert b.get_task_meta(uuid())['status'] == states.PENDING
  109. @contextmanager
  110. def _result_context(self):
  111. results = Queue()
  112. class Message(object):
  113. acked = 0
  114. requeued = 0
  115. def __init__(self, **merge):
  116. self.payload = dict({'status': states.STARTED,
  117. 'result': None}, **merge)
  118. self.properties = {'correlation_id': merge.get('task_id')}
  119. self.body = pickle.dumps(self.payload)
  120. self.content_type = 'application/x-python-serialize'
  121. self.content_encoding = 'binary'
  122. def ack(self, *args, **kwargs):
  123. self.acked += 1
  124. def requeue(self, *args, **kwargs):
  125. self.requeued += 1
  126. class MockBinding(object):
  127. def __init__(self, *args, **kwargs):
  128. self.channel = Mock()
  129. def __call__(self, *args, **kwargs):
  130. return self
  131. def declare(self):
  132. pass
  133. def get(self, no_ack=False, accept=None):
  134. try:
  135. m = results.get(block=False)
  136. if m:
  137. m.accept = accept
  138. return m
  139. except Empty:
  140. pass
  141. def is_bound(self):
  142. return True
  143. class MockBackend(AMQPBackend):
  144. Queue = MockBinding
  145. backend = MockBackend(self.app, max_cached_results=100)
  146. backend._republish = Mock()
  147. yield results, backend, Message
  148. def test_backlog_limit_exceeded(self):
  149. with self._result_context() as (results, backend, Message):
  150. for i in range(1001):
  151. results.put(Message(task_id='id', status=states.RECEIVED))
  152. with pytest.raises(backend.BacklogLimitExceeded):
  153. backend.get_task_meta('id')
  154. def test_poll_result(self):
  155. with self._result_context() as (results, backend, Message):
  156. tid = uuid()
  157. # FFWD's to the latest state.
  158. state_messages = [
  159. Message(task_id=tid, status=states.RECEIVED, seq=1),
  160. Message(task_id=tid, status=states.STARTED, seq=2),
  161. Message(task_id=tid, status=states.FAILURE, seq=3),
  162. ]
  163. for state_message in state_messages:
  164. results.put(state_message)
  165. r1 = backend.get_task_meta(tid)
  166. # FFWDs to the last state.
  167. assert r1['status'] == states.FAILURE
  168. assert r1['seq'] == 3
  169. # Caches last known state.
  170. tid = uuid()
  171. results.put(Message(task_id=tid))
  172. backend.get_task_meta(tid)
  173. assert tid, backend._cache in 'Caches last known state'
  174. assert state_messages[-1].requeued
  175. # Returns cache if no new states.
  176. results.queue.clear()
  177. assert not results.qsize()
  178. backend._cache[tid] = 'hello'
  179. # returns cache if no new states.
  180. assert backend.get_task_meta(tid) == 'hello'
  181. def test_drain_events_decodes_exceptions_in_meta(self):
  182. tid = uuid()
  183. b = self.create_backend(serializer='json')
  184. b.store_result(tid, RuntimeError('aap'), states.FAILURE)
  185. result = AsyncResult(tid, backend=b)
  186. with pytest.raises(Exception) as excinfo:
  187. result.get()
  188. assert excinfo.value.__class__.__name__ == 'RuntimeError'
  189. assert str(excinfo.value) == 'aap'
  190. def test_no_expires(self):
  191. b = self.create_backend(expires=None)
  192. app = self.app
  193. app.conf.result_expires = None
  194. b = self.create_backend(expires=None)
  195. with pytest.raises(KeyError):
  196. b.queue_arguments['x-expires']
  197. def test_process_cleanup(self):
  198. self.create_backend().process_cleanup()
  199. def test_reload_task_result(self):
  200. with pytest.raises(NotImplementedError):
  201. self.create_backend().reload_task_result('x')
  202. def test_reload_group_result(self):
  203. with pytest.raises(NotImplementedError):
  204. self.create_backend().reload_group_result('x')
  205. def test_save_group(self):
  206. with pytest.raises(NotImplementedError):
  207. self.create_backend().save_group('x', 'x')
  208. def test_restore_group(self):
  209. with pytest.raises(NotImplementedError):
  210. self.create_backend().restore_group('x')
  211. def test_delete_group(self):
  212. with pytest.raises(NotImplementedError):
  213. self.create_backend().delete_group('x')