/usr/lib/python2.7/dist-packages/celery/tests/tasks/test_chord.py is in python-celery 3.1.20-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 | from __future__ import absolute_import
from contextlib import contextmanager
from celery import group
from celery import canvas
from celery import result
from celery.exceptions import ChordError, Retry
from celery.five import range
from celery.result import AsyncResult, GroupResult, EagerResult
from celery.tests.case import AppCase, Mock
def passthru(x):
return x
class ChordCase(AppCase):
def setup(self):
@self.app.task(shared=False)
def add(x, y):
return x + y
self.add = add
class TSR(GroupResult):
is_ready = True
value = None
def ready(self):
return self.is_ready
def join(self, propagate=True, **kwargs):
if propagate:
for value in self.value:
if isinstance(value, Exception):
raise value
return self.value
join_native = join
def _failed_join_report(self):
for value in self.value:
if isinstance(value, Exception):
yield EagerResult('some_id', value, 'FAILURE')
class TSRNoReport(TSR):
def _failed_join_report(self):
return iter([])
@contextmanager
def patch_unlock_retry(app):
unlock = app.tasks['celery.chord_unlock']
retry = Mock()
retry.return_value = Retry()
prev, unlock.retry = unlock.retry, retry
try:
yield unlock, retry
finally:
unlock.retry = prev
class test_unlock_chord_task(ChordCase):
def test_unlock_ready(self):
class AlwaysReady(TSR):
is_ready = True
value = [2, 4, 8, 6]
with self._chord_context(AlwaysReady) as (cb, retry, _):
cb.type.apply_async.assert_called_with(
([2, 4, 8, 6], ), {}, task_id=cb.id,
)
# did not retry
self.assertFalse(retry.call_count)
def test_callback_fails(self):
class AlwaysReady(TSR):
is_ready = True
value = [2, 4, 8, 6]
def setup(callback):
callback.apply_async.side_effect = IOError()
with self._chord_context(AlwaysReady, setup) as (cb, retry, fail):
self.assertTrue(fail.called)
self.assertEqual(
fail.call_args[0][0], cb.id,
)
self.assertIsInstance(
fail.call_args[1]['exc'], ChordError,
)
def test_unlock_ready_failed(self):
class Failed(TSR):
is_ready = True
value = [2, KeyError('foo'), 8, 6]
with self._chord_context(Failed) as (cb, retry, fail_current):
self.assertFalse(cb.type.apply_async.called)
# did not retry
self.assertFalse(retry.call_count)
self.assertTrue(fail_current.called)
self.assertEqual(
fail_current.call_args[0][0], cb.id,
)
self.assertIsInstance(
fail_current.call_args[1]['exc'], ChordError,
)
self.assertIn('some_id', str(fail_current.call_args[1]['exc']))
def test_unlock_ready_failed_no_culprit(self):
class Failed(TSRNoReport):
is_ready = True
value = [2, KeyError('foo'), 8, 6]
with self._chord_context(Failed) as (cb, retry, fail_current):
self.assertTrue(fail_current.called)
self.assertEqual(
fail_current.call_args[0][0], cb.id,
)
self.assertIsInstance(
fail_current.call_args[1]['exc'], ChordError,
)
@contextmanager
def _chord_context(self, ResultCls, setup=None, **kwargs):
@self.app.task(shared=False)
def callback(*args, **kwargs):
pass
self.app.finalize()
pts, result.GroupResult = result.GroupResult, ResultCls
callback.apply_async = Mock()
callback_s = callback.s()
callback_s.id = 'callback_id'
fail_current = self.app.backend.fail_from_current_stack = Mock()
try:
with patch_unlock_retry(self.app) as (unlock, retry):
subtask, canvas.maybe_signature = (
canvas.maybe_signature, passthru,
)
if setup:
setup(callback)
try:
assert self.app.tasks['celery.chord_unlock'] is unlock
try:
unlock(
'group_id', callback_s,
result=[
self.app.AsyncResult(r) for r in ['1', 2, 3]
],
GroupResult=ResultCls, **kwargs
)
except Retry:
pass
finally:
canvas.maybe_signature = subtask
yield callback_s, retry, fail_current
finally:
result.GroupResult = pts
def test_when_not_ready(self):
class NeverReady(TSR):
is_ready = False
with self._chord_context(NeverReady, interval=10, max_retries=30) \
as (cb, retry, _):
self.assertFalse(cb.type.apply_async.called)
# did retry
retry.assert_called_with(countdown=10, max_retries=30)
def test_is_in_registry(self):
self.assertIn('celery.chord_unlock', self.app.tasks)
class test_chord(ChordCase):
def test_eager(self):
from celery import chord
@self.app.task(shared=False)
def addX(x, y):
return x + y
@self.app.task(shared=False)
def sumX(n):
return sum(n)
self.app.conf.CELERY_ALWAYS_EAGER = True
x = chord(addX.s(i, i) for i in range(10))
body = sumX.s()
result = x(body)
self.assertEqual(result.get(), sum(i + i for i in range(10)))
def test_apply(self):
self.app.conf.CELERY_ALWAYS_EAGER = False
from celery import chord
m = Mock()
m.app.conf.CELERY_ALWAYS_EAGER = False
m.AsyncResult = AsyncResult
prev, chord._type = chord._type, m
try:
x = chord(self.add.s(i, i) for i in range(10))
body = self.add.s(2)
result = x(body)
self.assertTrue(result.id)
# does not modify original subtask
with self.assertRaises(KeyError):
body.options['task_id']
self.assertTrue(chord._type.called)
finally:
chord._type = prev
class test_Chord_task(ChordCase):
def test_run(self):
self.app.backend = Mock()
self.app.backend.cleanup = Mock()
self.app.backend.cleanup.__name__ = 'cleanup'
Chord = self.app.tasks['celery.chord']
body = dict()
Chord(group(self.add.subtask((i, i)) for i in range(5)), body)
Chord([self.add.subtask((j, j)) for j in range(5)], body)
self.assertEqual(self.app.backend.apply_chord.call_count, 2)
|