test_request.py 35 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043
  1. # -*- coding: utf-8 -*-
  2. from __future__ import absolute_import, unicode_literals
  3. import numbers
  4. import os
  5. import signal
  6. import socket
  7. import sys
  8. from datetime import datetime, timedelta
  9. from time import time
  10. import pytest
  11. from billiard.einfo import ExceptionInfo
  12. from case import Mock, patch
  13. from kombu.utils.encoding import (default_encode, from_utf8, safe_repr,
  14. safe_str)
  15. from kombu.utils.uuid import uuid
  16. from celery import states
  17. from celery.app.trace import (TraceInfo, _trace_task_ret, build_tracer,
  18. mro_lookup, reset_worker_optimizations,
  19. setup_worker_optimizations, trace_task)
  20. from celery.exceptions import (Ignore, InvalidTaskError, Reject, Retry,
  21. TaskRevokedError, Terminated, WorkerLostError)
  22. from celery.five import monotonic
  23. from celery.signals import task_revoked
  24. from celery.worker import request as module
  25. from celery.worker import strategy
  26. from celery.worker.request import Request, create_request_cls
  27. from celery.worker.request import logger as req_logger
  28. from celery.worker.state import revoked
  29. class RequestCase:
  30. def setup(self):
  31. self.app.conf.result_serializer = 'pickle'
  32. @self.app.task(shared=False)
  33. def add(x, y, **kw_):
  34. return x + y
  35. self.add = add
  36. @self.app.task(shared=False)
  37. def mytask(i, **kwargs):
  38. return i ** i
  39. self.mytask = mytask
  40. @self.app.task(shared=False)
  41. def mytask_raising(i):
  42. raise KeyError(i)
  43. self.mytask_raising = mytask_raising
  44. def xRequest(self, name=None, id=None, args=None, kwargs=None,
  45. on_ack=None, on_reject=None, Request=Request, **head):
  46. args = [1] if args is None else args
  47. kwargs = {'f': 'x'} if kwargs is None else kwargs
  48. on_ack = on_ack or Mock(name='on_ack')
  49. on_reject = on_reject or Mock(name='on_reject')
  50. message = self.TaskMessage(
  51. name or self.mytask.name, id, args=args, kwargs=kwargs, **head
  52. )
  53. return Request(message, app=self.app,
  54. on_ack=on_ack, on_reject=on_reject)
  55. class test_mro_lookup:
  56. def test_order(self):
  57. class A(object):
  58. pass
  59. class B(A):
  60. pass
  61. class C(B):
  62. pass
  63. class D(C):
  64. @classmethod
  65. def mro(cls):
  66. return ()
  67. A.x = 10
  68. assert mro_lookup(C, 'x') == A
  69. assert mro_lookup(C, 'x', stop={A}) is None
  70. B.x = 10
  71. assert mro_lookup(C, 'x') == B
  72. C.x = 10
  73. assert mro_lookup(C, 'x') == C
  74. assert mro_lookup(D, 'x') is None
  75. def jail(app, task_id, name, args, kwargs):
  76. request = {'id': task_id}
  77. task = app.tasks[name]
  78. task.__trace__ = None # rebuild
  79. return trace_task(
  80. task, task_id, args, kwargs, request=request, eager=False, app=app,
  81. ).retval
  82. @pytest.mark.skipif(sys.version_info[0] > 3, reason='Py2 only')
  83. class test_default_encode:
  84. def test_jython(self):
  85. prev, sys.platform = sys.platform, 'java 1.6.1'
  86. try:
  87. assert default_encode(b'foo') == b'foo'
  88. finally:
  89. sys.platform = prev
  90. def test_cpython(self):
  91. prev, sys.platform = sys.platform, 'darwin'
  92. gfe, sys.getfilesystemencoding = (
  93. sys.getfilesystemencoding,
  94. lambda: 'utf-8',
  95. )
  96. try:
  97. assert default_encode(b'foo') == b'foo'
  98. finally:
  99. sys.platform = prev
  100. sys.getfilesystemencoding = gfe
  101. class test_Retry:
  102. def test_retry_semipredicate(self):
  103. try:
  104. raise Exception('foo')
  105. except Exception as exc:
  106. ret = Retry('Retrying task', exc)
  107. assert ret.exc == exc
  108. class test_trace_task(RequestCase):
  109. def test_process_cleanup_fails(self, patching):
  110. _logger = patching('celery.app.trace.logger')
  111. self.mytask.backend = Mock()
  112. self.mytask.backend.process_cleanup = Mock(side_effect=KeyError())
  113. tid = uuid()
  114. ret = jail(self.app, tid, self.mytask.name, [2], {})
  115. assert ret == 4
  116. self.mytask.backend.mark_as_done.assert_called()
  117. assert 'Process cleanup failed' in _logger.error.call_args[0][0]
  118. def test_process_cleanup_BaseException(self):
  119. self.mytask.backend = Mock()
  120. self.mytask.backend.process_cleanup = Mock(side_effect=SystemExit())
  121. with pytest.raises(SystemExit):
  122. jail(self.app, uuid(), self.mytask.name, [2], {})
  123. def test_execute_jail_success(self):
  124. ret = jail(self.app, uuid(), self.mytask.name, [2], {})
  125. assert ret == 4
  126. def test_marked_as_started(self):
  127. _started = []
  128. def store_result(tid, meta, state, **kwargs):
  129. if state == states.STARTED:
  130. _started.append(tid)
  131. self.mytask.backend.store_result = Mock(name='store_result')
  132. self.mytask.backend.store_result.side_effect = store_result
  133. self.mytask.track_started = True
  134. tid = uuid()
  135. jail(self.app, tid, self.mytask.name, [2], {})
  136. assert tid in _started
  137. self.mytask.ignore_result = True
  138. tid = uuid()
  139. jail(self.app, tid, self.mytask.name, [2], {})
  140. assert tid not in _started
  141. def test_execute_jail_failure(self):
  142. ret = jail(
  143. self.app, uuid(), self.mytask_raising.name, [4], {},
  144. )
  145. assert isinstance(ret, ExceptionInfo)
  146. assert ret.exception.args == (4,)
  147. def test_execute_ignore_result(self):
  148. @self.app.task(shared=False, ignore_result=True)
  149. def ignores_result(i):
  150. return i ** i
  151. task_id = uuid()
  152. ret = jail(self.app, task_id, ignores_result.name, [4], {})
  153. assert ret == 256
  154. assert not self.app.AsyncResult(task_id).ready()
  155. class test_Request(RequestCase):
  156. def get_request(self, sig, Request=Request, **kwargs):
  157. return Request(
  158. self.task_message_from_sig(self.app, sig),
  159. on_ack=Mock(name='on_ack'),
  160. on_reject=Mock(name='on_reject'),
  161. eventer=Mock(name='eventer'),
  162. app=self.app,
  163. connection_errors=(socket.error,),
  164. task=sig.type,
  165. **kwargs
  166. )
  167. def test_shadow(self):
  168. assert self.get_request(
  169. self.add.s(2, 2).set(shadow='fooxyz')).name == 'fooxyz'
  170. def test_invalid_eta_raises_InvalidTaskError(self):
  171. with pytest.raises(InvalidTaskError):
  172. self.get_request(self.add.s(2, 2).set(eta='12345'))
  173. def test_invalid_expires_raises_InvalidTaskError(self):
  174. with pytest.raises(InvalidTaskError):
  175. self.get_request(self.add.s(2, 2).set(expires='12345'))
  176. def test_valid_expires_with_utc_makes_aware(self):
  177. with patch('celery.worker.request.maybe_make_aware') as mma:
  178. self.get_request(self.add.s(2, 2).set(expires=10),
  179. maybe_make_aware=mma)
  180. mma.assert_called()
  181. def test_maybe_expire_when_expires_is_None(self):
  182. req = self.get_request(self.add.s(2, 2))
  183. assert not req.maybe_expire()
  184. def test_on_retry_acks_if_late(self):
  185. self.add.acks_late = True
  186. req = self.get_request(self.add.s(2, 2))
  187. req.on_retry(Mock())
  188. req.on_ack.assert_called_with(req_logger, req.connection_errors)
  189. def test_on_failure_Termianted(self):
  190. einfo = None
  191. try:
  192. raise Terminated('9')
  193. except Terminated:
  194. einfo = ExceptionInfo()
  195. assert einfo is not None
  196. req = self.get_request(self.add.s(2, 2))
  197. req.on_failure(einfo)
  198. req.eventer.send.assert_called_with(
  199. 'task-revoked',
  200. uuid=req.id, terminated=True, signum='9', expired=False,
  201. )
  202. def test_on_failure_propagates_MemoryError(self):
  203. einfo = None
  204. try:
  205. raise MemoryError()
  206. except MemoryError:
  207. einfo = ExceptionInfo(internal=True)
  208. assert einfo is not None
  209. req = self.get_request(self.add.s(2, 2))
  210. with pytest.raises(MemoryError):
  211. req.on_failure(einfo)
  212. def test_on_failure_Ignore_acknowledges(self):
  213. einfo = None
  214. try:
  215. raise Ignore()
  216. except Ignore:
  217. einfo = ExceptionInfo(internal=True)
  218. assert einfo is not None
  219. req = self.get_request(self.add.s(2, 2))
  220. req.on_failure(einfo)
  221. req.on_ack.assert_called_with(req_logger, req.connection_errors)
  222. def test_on_failure_Reject_rejects(self):
  223. einfo = None
  224. try:
  225. raise Reject()
  226. except Reject:
  227. einfo = ExceptionInfo(internal=True)
  228. assert einfo is not None
  229. req = self.get_request(self.add.s(2, 2))
  230. req.on_failure(einfo)
  231. req.on_reject.assert_called_with(
  232. req_logger, req.connection_errors, False,
  233. )
  234. def test_on_failure_Reject_rejects_with_requeue(self):
  235. einfo = None
  236. try:
  237. raise Reject(requeue=True)
  238. except Reject:
  239. einfo = ExceptionInfo(internal=True)
  240. assert einfo is not None
  241. req = self.get_request(self.add.s(2, 2))
  242. req.on_failure(einfo)
  243. req.on_reject.assert_called_with(
  244. req_logger, req.connection_errors, True,
  245. )
  246. def test_on_failure_WorkerLostError_rejects_with_requeue(self):
  247. einfo = None
  248. try:
  249. raise WorkerLostError()
  250. except:
  251. einfo = ExceptionInfo(internal=True)
  252. req = self.get_request(self.add.s(2, 2))
  253. req.task.acks_late = True
  254. req.task.reject_on_worker_lost = True
  255. req.delivery_info['redelivered'] = False
  256. req.on_failure(einfo)
  257. req.on_reject.assert_called_with(
  258. req_logger, req.connection_errors, True)
  259. def test_on_failure_WorkerLostError_redelivered_None(self):
  260. einfo = None
  261. try:
  262. raise WorkerLostError()
  263. except:
  264. einfo = ExceptionInfo(internal=True)
  265. req = self.get_request(self.add.s(2, 2))
  266. req.task.acks_late = True
  267. req.task.reject_on_worker_lost = True
  268. req.delivery_info['redelivered'] = None
  269. req.on_failure(einfo)
  270. req.on_reject.assert_called_with(
  271. req_logger, req.connection_errors, True)
  272. def test_tzlocal_is_cached(self):
  273. req = self.get_request(self.add.s(2, 2))
  274. req._tzlocal = 'foo'
  275. assert req.tzlocal == 'foo'
  276. def test_task_wrapper_repr(self):
  277. assert repr(self.xRequest())
  278. def test_sets_store_errors(self):
  279. self.mytask.ignore_result = True
  280. job = self.xRequest()
  281. assert not job.store_errors
  282. self.mytask.store_errors_even_if_ignored = True
  283. job = self.xRequest()
  284. assert job.store_errors
  285. def test_send_event(self):
  286. job = self.xRequest()
  287. job.eventer = Mock(name='.eventer')
  288. job.send_event('task-frobulated')
  289. job.eventer.send.assert_called_with('task-frobulated', uuid=job.id)
  290. def test_send_events__disabled_at_task_level(self):
  291. job = self.xRequest()
  292. job.task.send_events = False
  293. job.eventer = Mock(name='.eventer')
  294. job.send_event('task-frobulated')
  295. job.eventer.send.assert_not_called()
  296. def test_on_retry(self):
  297. job = self.get_request(self.mytask.s(1, f='x'))
  298. job.eventer = Mock(name='.eventer')
  299. try:
  300. raise Retry('foo', KeyError('moofoobar'))
  301. except:
  302. einfo = ExceptionInfo()
  303. job.on_failure(einfo)
  304. job.eventer.send.assert_called_with(
  305. 'task-retried',
  306. uuid=job.id,
  307. exception=safe_repr(einfo.exception.exc),
  308. traceback=safe_str(einfo.traceback),
  309. )
  310. prev, module._does_info = module._does_info, False
  311. try:
  312. job.on_failure(einfo)
  313. finally:
  314. module._does_info = prev
  315. einfo.internal = True
  316. job.on_failure(einfo)
  317. def test_compat_properties(self):
  318. job = self.xRequest()
  319. assert job.task_id == job.id
  320. assert job.task_name == job.name
  321. job.task_id = 'ID'
  322. assert job.id == 'ID'
  323. job.task_name = 'NAME'
  324. assert job.name == 'NAME'
  325. def test_terminate__pool_ref(self):
  326. pool = Mock()
  327. signum = signal.SIGTERM
  328. job = self.get_request(self.mytask.s(1, f='x'))
  329. job._apply_result = Mock(name='_apply_result')
  330. with self.assert_signal_called(
  331. task_revoked, sender=job.task, request=job,
  332. terminated=True, expired=False, signum=signum):
  333. job.time_start = monotonic()
  334. job.worker_pid = 314
  335. job.terminate(pool, signal='TERM')
  336. job._apply_result().terminate.assert_called_with(signum)
  337. job._apply_result = Mock(name='_apply_result2')
  338. job._apply_result.return_value = None
  339. job.terminate(pool, signal='TERM')
  340. def test_terminate__task_started(self):
  341. pool = Mock()
  342. signum = signal.SIGTERM
  343. job = self.get_request(self.mytask.s(1, f='x'))
  344. with self.assert_signal_called(
  345. task_revoked, sender=job.task, request=job,
  346. terminated=True, expired=False, signum=signum):
  347. job.time_start = monotonic()
  348. job.worker_pid = 313
  349. job.terminate(pool, signal='TERM')
  350. pool.terminate_job.assert_called_with(job.worker_pid, signum)
  351. def test_terminate__task_reserved(self):
  352. pool = Mock()
  353. job = self.get_request(self.mytask.s(1, f='x'))
  354. job.time_start = None
  355. job.terminate(pool, signal='TERM')
  356. pool.terminate_job.assert_not_called()
  357. assert job._terminate_on_ack == (pool, 15)
  358. job.terminate(pool, signal='TERM')
  359. def test_revoked_expires_expired(self):
  360. job = self.get_request(self.mytask.s(1, f='x').set(
  361. expires=datetime.utcnow() - timedelta(days=1)
  362. ))
  363. with self.assert_signal_called(
  364. task_revoked, sender=job.task, request=job,
  365. terminated=False, expired=True, signum=None):
  366. job.revoked()
  367. assert job.id in revoked
  368. assert self.mytask.backend.get_status(job.id) == states.REVOKED
  369. def test_revoked_expires_not_expired(self):
  370. job = self.xRequest(
  371. expires=datetime.utcnow() + timedelta(days=1),
  372. )
  373. job.revoked()
  374. assert job.id not in revoked
  375. assert self.mytask.backend.get_status(job.id) != states.REVOKED
  376. def test_revoked_expires_ignore_result(self):
  377. self.mytask.ignore_result = True
  378. job = self.xRequest(
  379. expires=datetime.utcnow() - timedelta(days=1),
  380. )
  381. job.revoked()
  382. assert job.id in revoked
  383. assert self.mytask.backend.get_status(job.id) != states.REVOKED
  384. def test_already_revoked(self):
  385. job = self.xRequest()
  386. job._already_revoked = True
  387. assert job.revoked()
  388. def test_revoked(self):
  389. job = self.xRequest()
  390. with self.assert_signal_called(
  391. task_revoked, sender=job.task, request=job,
  392. terminated=False, expired=False, signum=None):
  393. revoked.add(job.id)
  394. assert job.revoked()
  395. assert job._already_revoked
  396. assert job.acknowledged
  397. def test_execute_does_not_execute_revoked(self):
  398. job = self.xRequest()
  399. revoked.add(job.id)
  400. job.execute()
  401. def test_execute_acks_late(self):
  402. self.mytask_raising.acks_late = True
  403. job = self.xRequest(
  404. name=self.mytask_raising.name,
  405. kwargs={},
  406. )
  407. job.execute()
  408. assert job.acknowledged
  409. job.execute()
  410. def test_execute_using_pool_does_not_execute_revoked(self):
  411. job = self.xRequest()
  412. revoked.add(job.id)
  413. with pytest.raises(TaskRevokedError):
  414. job.execute_using_pool(None)
  415. def test_on_accepted_acks_early(self):
  416. job = self.xRequest()
  417. job.on_accepted(pid=os.getpid(), time_accepted=monotonic())
  418. assert job.acknowledged
  419. prev, module._does_debug = module._does_debug, False
  420. try:
  421. job.on_accepted(pid=os.getpid(), time_accepted=monotonic())
  422. finally:
  423. module._does_debug = prev
  424. def test_on_accepted_acks_late(self):
  425. job = self.xRequest()
  426. self.mytask.acks_late = True
  427. job.on_accepted(pid=os.getpid(), time_accepted=monotonic())
  428. assert not job.acknowledged
  429. def test_on_accepted_terminates(self):
  430. signum = signal.SIGTERM
  431. pool = Mock()
  432. job = self.xRequest()
  433. with self.assert_signal_called(
  434. task_revoked, sender=job.task, request=job,
  435. terminated=True, expired=False, signum=signum):
  436. job.terminate(pool, signal='TERM')
  437. assert not pool.terminate_job.call_count
  438. job.on_accepted(pid=314, time_accepted=monotonic())
  439. pool.terminate_job.assert_called_with(314, signum)
  440. def test_on_accepted_time_start(self):
  441. job = self.xRequest()
  442. job.on_accepted(pid=os.getpid(), time_accepted=monotonic())
  443. assert time() - job.time_start < 1
  444. def test_on_success_acks_early(self):
  445. job = self.xRequest()
  446. job.time_start = 1
  447. job.on_success((0, 42, 0.001))
  448. prev, module._does_info = module._does_info, False
  449. try:
  450. job.on_success((0, 42, 0.001))
  451. assert not job.acknowledged
  452. finally:
  453. module._does_info = prev
  454. def test_on_success_BaseException(self):
  455. job = self.xRequest()
  456. job.time_start = 1
  457. with pytest.raises(SystemExit):
  458. try:
  459. raise SystemExit()
  460. except SystemExit:
  461. job.on_success((1, ExceptionInfo(), 0.01))
  462. else:
  463. assert False
  464. def test_on_success_eventer(self):
  465. job = self.xRequest()
  466. job.time_start = 1
  467. job.eventer = Mock()
  468. job.eventer.send = Mock()
  469. job.on_success((0, 42, 0.001))
  470. job.eventer.send.assert_called()
  471. def test_on_success_when_failure(self):
  472. job = self.xRequest()
  473. job.time_start = 1
  474. job.on_failure = Mock()
  475. try:
  476. raise KeyError('foo')
  477. except Exception:
  478. job.on_success((1, ExceptionInfo(), 0.001))
  479. job.on_failure.assert_called()
  480. def test_on_success_acks_late(self):
  481. job = self.xRequest()
  482. job.time_start = 1
  483. self.mytask.acks_late = True
  484. job.on_success((0, 42, 0.001))
  485. assert job.acknowledged
  486. def test_on_failure_WorkerLostError(self):
  487. def get_ei():
  488. try:
  489. raise WorkerLostError('do re mi')
  490. except WorkerLostError:
  491. return ExceptionInfo()
  492. job = self.xRequest()
  493. exc_info = get_ei()
  494. job.on_failure(exc_info)
  495. assert self.mytask.backend.get_status(job.id) == states.FAILURE
  496. self.mytask.ignore_result = True
  497. exc_info = get_ei()
  498. job = self.xRequest()
  499. job.on_failure(exc_info)
  500. assert self.mytask.backend.get_status(job.id) == states.PENDING
  501. def test_on_failure_acks_late(self):
  502. job = self.xRequest()
  503. job.time_start = 1
  504. self.mytask.acks_late = True
  505. try:
  506. raise KeyError('foo')
  507. except KeyError:
  508. exc_info = ExceptionInfo()
  509. job.on_failure(exc_info)
  510. assert job.acknowledged
  511. def test_from_message_invalid_kwargs(self):
  512. m = self.TaskMessage(self.mytask.name, args=(), kwargs='foo')
  513. req = Request(m, app=self.app)
  514. with pytest.raises(InvalidTaskError):
  515. raise req.execute().exception
  516. def test_on_hard_timeout(self, patching):
  517. error = patching('celery.worker.request.error')
  518. job = self.xRequest()
  519. job.acknowledge = Mock(name='ack')
  520. job.task.acks_late = True
  521. job.on_timeout(soft=False, timeout=1337)
  522. assert 'Hard time limit' in error.call_args[0][0]
  523. assert self.mytask.backend.get_status(job.id) == states.FAILURE
  524. job.acknowledge.assert_called_with()
  525. job = self.xRequest()
  526. job.acknowledge = Mock(name='ack')
  527. job.task.acks_late = False
  528. job.on_timeout(soft=False, timeout=1335)
  529. job.acknowledge.assert_not_called()
  530. def test_on_soft_timeout(self, patching):
  531. warn = patching('celery.worker.request.warn')
  532. job = self.xRequest()
  533. job.acknowledge = Mock(name='ack')
  534. job.task.acks_late = True
  535. job.on_timeout(soft=True, timeout=1337)
  536. assert 'Soft time limit' in warn.call_args[0][0]
  537. assert self.mytask.backend.get_status(job.id) == states.PENDING
  538. job.acknowledge.assert_not_called()
  539. self.mytask.ignore_result = True
  540. job = self.xRequest()
  541. job.on_timeout(soft=True, timeout=1336)
  542. assert self.mytask.backend.get_status(job.id) == states.PENDING
  543. def test_fast_trace_task(self):
  544. from celery.app import trace
  545. setup_worker_optimizations(self.app)
  546. assert trace.trace_task_ret is trace._fast_trace_task
  547. tid = uuid()
  548. message = self.TaskMessage(self.mytask.name, tid, args=[4])
  549. assert len(message.payload) == 3
  550. try:
  551. self.mytask.__trace__ = build_tracer(
  552. self.mytask.name, self.mytask, self.app.loader, 'test',
  553. app=self.app,
  554. )
  555. failed, res, runtime = trace.trace_task_ret(
  556. self.mytask.name, tid, message.headers, message.body,
  557. message.content_type, message.content_encoding)
  558. assert not failed
  559. assert res == repr(4 ** 4)
  560. assert runtime is not None
  561. assert isinstance(runtime, numbers.Real)
  562. finally:
  563. reset_worker_optimizations()
  564. assert trace.trace_task_ret is trace._trace_task_ret
  565. delattr(self.mytask, '__trace__')
  566. failed, res, runtime = trace.trace_task_ret(
  567. self.mytask.name, tid, message.headers, message.body,
  568. message.content_type, message.content_encoding, app=self.app,
  569. )
  570. assert not failed
  571. assert res == repr(4 ** 4)
  572. assert runtime is not None
  573. assert isinstance(runtime, numbers.Real)
  574. def test_trace_task_ret(self):
  575. self.mytask.__trace__ = build_tracer(
  576. self.mytask.name, self.mytask, self.app.loader, 'test',
  577. app=self.app,
  578. )
  579. tid = uuid()
  580. message = self.TaskMessage(self.mytask.name, tid, args=[4])
  581. _, R, _ = _trace_task_ret(
  582. self.mytask.name, tid, message.headers,
  583. message.body, message.content_type,
  584. message.content_encoding, app=self.app,
  585. )
  586. assert R == repr(4 ** 4)
  587. def test_trace_task_ret__no_trace(self):
  588. try:
  589. delattr(self.mytask, '__trace__')
  590. except AttributeError:
  591. pass
  592. tid = uuid()
  593. message = self.TaskMessage(self.mytask.name, tid, args=[4])
  594. _, R, _ = _trace_task_ret(
  595. self.mytask.name, tid, message.headers,
  596. message.body, message.content_type,
  597. message.content_encoding, app=self.app,
  598. )
  599. assert R == repr(4 ** 4)
  600. def test_trace_catches_exception(self):
  601. @self.app.task(request=None, shared=False)
  602. def raising():
  603. raise KeyError('baz')
  604. with pytest.warns(RuntimeWarning):
  605. res = trace_task(raising, uuid(), [], {}, app=self.app)[0]
  606. assert isinstance(res, ExceptionInfo)
  607. def test_worker_task_trace_handle_retry(self):
  608. tid = uuid()
  609. self.mytask.push_request(id=tid)
  610. try:
  611. raise ValueError('foo')
  612. except Exception as exc:
  613. try:
  614. raise Retry(str(exc), exc=exc)
  615. except Retry as exc:
  616. w = TraceInfo(states.RETRY, exc)
  617. w.handle_retry(
  618. self.mytask, self.mytask.request, store_errors=False,
  619. )
  620. assert self.mytask.backend.get_status(tid) == states.PENDING
  621. w.handle_retry(
  622. self.mytask, self.mytask.request, store_errors=True,
  623. )
  624. assert self.mytask.backend.get_status(tid) == states.RETRY
  625. finally:
  626. self.mytask.pop_request()
  627. def test_worker_task_trace_handle_failure(self):
  628. tid = uuid()
  629. self.mytask.push_request()
  630. try:
  631. self.mytask.request.id = tid
  632. try:
  633. raise ValueError('foo')
  634. except Exception as exc:
  635. w = TraceInfo(states.FAILURE, exc)
  636. w.handle_failure(
  637. self.mytask, self.mytask.request, store_errors=False,
  638. )
  639. assert self.mytask.backend.get_status(tid) == states.PENDING
  640. w.handle_failure(
  641. self.mytask, self.mytask.request, store_errors=True,
  642. )
  643. assert self.mytask.backend.get_status(tid) == states.FAILURE
  644. finally:
  645. self.mytask.pop_request()
  646. def test_from_message(self):
  647. us = 'æØåveéðƒeæ'
  648. tid = uuid()
  649. m = self.TaskMessage(
  650. self.mytask.name, tid, args=[2], kwargs={us: 'bar'},
  651. )
  652. job = Request(m, app=self.app)
  653. assert isinstance(job, Request)
  654. assert job.name == self.mytask.name
  655. assert job.id == tid
  656. assert job.message is m
  657. def test_from_message_empty_args(self):
  658. tid = uuid()
  659. m = self.TaskMessage(self.mytask.name, tid, args=[], kwargs={})
  660. job = Request(m, app=self.app)
  661. assert isinstance(job, Request)
  662. def test_from_message_missing_required_fields(self):
  663. m = self.TaskMessage(self.mytask.name)
  664. m.headers.clear()
  665. with pytest.raises(KeyError):
  666. Request(m, app=self.app)
  667. def test_from_message_nonexistant_task(self):
  668. m = self.TaskMessage(
  669. 'cu.mytask.doesnotexist',
  670. args=[2], kwargs={'æØåveéðƒeæ': 'bar'},
  671. )
  672. with pytest.raises(KeyError):
  673. Request(m, app=self.app)
  674. def test_execute(self):
  675. tid = uuid()
  676. job = self.xRequest(id=tid, args=[4], kwargs={})
  677. assert job.execute() == 256
  678. meta = self.mytask.backend.get_task_meta(tid)
  679. assert meta['status'] == states.SUCCESS
  680. assert meta['result'] == 256
  681. def test_execute_success_no_kwargs(self):
  682. @self.app.task # traverses coverage for decorator without parens
  683. def mytask_no_kwargs(i):
  684. return i ** i
  685. tid = uuid()
  686. job = self.xRequest(
  687. name=mytask_no_kwargs.name,
  688. id=tid,
  689. args=[4],
  690. kwargs={},
  691. )
  692. assert job.execute() == 256
  693. meta = mytask_no_kwargs.backend.get_task_meta(tid)
  694. assert meta['result'] == 256
  695. assert meta['status'] == states.SUCCESS
  696. def test_execute_ack(self):
  697. scratch = {'ACK': False}
  698. def on_ack(*args, **kwargs):
  699. scratch['ACK'] = True
  700. tid = uuid()
  701. job = self.xRequest(id=tid, args=[4], on_ack=on_ack)
  702. assert job.execute() == 256
  703. meta = self.mytask.backend.get_task_meta(tid)
  704. assert scratch['ACK']
  705. assert meta['result'] == 256
  706. assert meta['status'] == states.SUCCESS
  707. def test_execute_fail(self):
  708. tid = uuid()
  709. job = self.xRequest(
  710. name=self.mytask_raising.name,
  711. id=tid,
  712. args=[4],
  713. kwargs={},
  714. )
  715. assert isinstance(job.execute(), ExceptionInfo)
  716. assert self.mytask_raising.backend.serializer == 'pickle'
  717. meta = self.mytask_raising.backend.get_task_meta(tid)
  718. assert meta['status'] == states.FAILURE
  719. assert isinstance(meta['result'], KeyError)
  720. def test_execute_using_pool(self):
  721. tid = uuid()
  722. job = self.xRequest(id=tid, args=[4])
  723. p = Mock()
  724. job.execute_using_pool(p)
  725. p.apply_async.assert_called_once()
  726. args = p.apply_async.call_args[1]['args']
  727. assert args[0] == self.mytask.name
  728. assert args[1] == tid
  729. assert args[2] == job.request_dict
  730. assert args[3] == job.message.body
  731. def _test_on_failure(self, exception, **kwargs):
  732. tid = uuid()
  733. job = self.xRequest(id=tid, args=[4])
  734. job.send_event = Mock(name='send_event')
  735. job.task.backend.mark_as_failure = Mock(name='mark_as_failure')
  736. try:
  737. raise exception
  738. except type(exception):
  739. exc_info = ExceptionInfo()
  740. job.on_failure(exc_info, **kwargs)
  741. job.send_event.assert_called()
  742. return job
  743. def test_on_failure(self):
  744. self._test_on_failure(Exception('Inside unit tests'))
  745. def test_on_failure__unicode_exception(self):
  746. self._test_on_failure(Exception('Бобры атакуют'))
  747. def test_on_failure__utf8_exception(self):
  748. self._test_on_failure(Exception(
  749. from_utf8('Бобры атакуют')))
  750. def test_on_failure__WorkerLostError(self):
  751. exc = WorkerLostError()
  752. job = self._test_on_failure(exc)
  753. job.task.backend.mark_as_failure.assert_called_with(
  754. job.id, exc, request=job, store_result=True,
  755. )
  756. def test_on_failure__return_ok(self):
  757. self._test_on_failure(KeyError(), return_ok=True)
  758. def test_reject(self):
  759. job = self.xRequest(id=uuid())
  760. job.on_reject = Mock(name='on_reject')
  761. job.reject(requeue=True)
  762. job.on_reject.assert_called_with(
  763. req_logger, job.connection_errors, True,
  764. )
  765. assert job.acknowledged
  766. job.on_reject.reset_mock()
  767. job.reject(requeue=True)
  768. job.on_reject.assert_not_called()
  769. def test_group(self):
  770. gid = uuid()
  771. job = self.xRequest(id=uuid(), group=gid)
  772. assert job.group == gid
  773. class test_create_request_class(RequestCase):
  774. def setup(self):
  775. self.task = Mock(name='task')
  776. self.pool = Mock(name='pool')
  777. self.eventer = Mock(name='eventer')
  778. RequestCase.setup(self)
  779. def create_request_cls(self, **kwargs):
  780. return create_request_cls(
  781. Request, self.task, self.pool, 'foo', self.eventer, **kwargs
  782. )
  783. def zRequest(self, Request=None, revoked_tasks=None, ref=None, **kwargs):
  784. return self.xRequest(
  785. Request=Request or self.create_request_cls(
  786. ref=ref,
  787. revoked_tasks=revoked_tasks,
  788. ),
  789. **kwargs)
  790. def test_on_success(self):
  791. self.zRequest(id=uuid()).on_success((False, 'hey', 3.1222))
  792. def test_on_success__SystemExit(self,
  793. errors=(SystemExit, KeyboardInterrupt)):
  794. for exc in errors:
  795. einfo = None
  796. try:
  797. raise exc()
  798. except exc:
  799. einfo = ExceptionInfo()
  800. with pytest.raises(exc):
  801. self.zRequest(id=uuid()).on_success((True, einfo, 1.0))
  802. def test_on_success__calls_failure(self):
  803. job = self.zRequest(id=uuid())
  804. einfo = Mock(name='einfo')
  805. job.on_failure = Mock(name='on_failure')
  806. job.on_success((True, einfo, 1.0))
  807. job.on_failure.assert_called_with(einfo, return_ok=True)
  808. def test_on_success__acks_late_enabled(self):
  809. self.task.acks_late = True
  810. job = self.zRequest(id=uuid())
  811. job.acknowledge = Mock(name='ack')
  812. job.on_success((False, 'foo', 1.0))
  813. job.acknowledge.assert_called_with()
  814. def test_on_success__acks_late_disabled(self):
  815. self.task.acks_late = False
  816. job = self.zRequest(id=uuid())
  817. job.acknowledge = Mock(name='ack')
  818. job.on_success((False, 'foo', 1.0))
  819. job.acknowledge.assert_not_called()
  820. def test_on_success__no_events(self):
  821. self.eventer = None
  822. job = self.zRequest(id=uuid())
  823. job.send_event = Mock(name='send_event')
  824. job.on_success((False, 'foo', 1.0))
  825. job.send_event.assert_not_called()
  826. def test_on_success__with_events(self):
  827. job = self.zRequest(id=uuid())
  828. job.send_event = Mock(name='send_event')
  829. job.on_success((False, 'foo', 1.0))
  830. job.send_event.assert_called_with(
  831. 'task-succeeded', result='foo', runtime=1.0,
  832. )
  833. def test_execute_using_pool__revoked(self):
  834. tid = uuid()
  835. job = self.zRequest(id=tid, revoked_tasks={tid})
  836. job.revoked = Mock()
  837. job.revoked.return_value = True
  838. with pytest.raises(TaskRevokedError):
  839. job.execute_using_pool(self.pool)
  840. def test_execute_using_pool__expired(self):
  841. tid = uuid()
  842. job = self.zRequest(id=tid, revoked_tasks=set())
  843. job.expires = 1232133
  844. job.revoked = Mock()
  845. job.revoked.return_value = True
  846. with pytest.raises(TaskRevokedError):
  847. job.execute_using_pool(self.pool)
  848. def test_execute_using_pool(self):
  849. from celery.app.trace import trace_task_ret as trace
  850. weakref_ref = Mock(name='weakref.ref')
  851. job = self.zRequest(id=uuid(), revoked_tasks=set(), ref=weakref_ref)
  852. job.execute_using_pool(self.pool)
  853. self.pool.apply_async.assert_called_with(
  854. trace,
  855. args=(job.type, job.id, job.request_dict, job.body,
  856. job.content_type, job.content_encoding),
  857. accept_callback=job.on_accepted,
  858. timeout_callback=job.on_timeout,
  859. callback=job.on_success,
  860. error_callback=job.on_failure,
  861. soft_timeout=self.task.soft_time_limit,
  862. timeout=self.task.time_limit,
  863. correlation_id=job.id,
  864. )
  865. assert job._apply_result
  866. weakref_ref.assert_called_with(self.pool.apply_async())
  867. assert job._apply_result is weakref_ref()
  868. def test_execute_using_pool_with_none_timelimit_header(self):
  869. from celery.app.trace import trace_task_ret as trace
  870. weakref_ref = Mock(name='weakref.ref')
  871. job = self.zRequest(id=uuid(),
  872. revoked_tasks=set(),
  873. ref=weakref_ref,
  874. headers={'timelimit': None})
  875. job.execute_using_pool(self.pool)
  876. self.pool.apply_async.assert_called_with(
  877. trace,
  878. args=(job.type, job.id, job.request_dict, job.body,
  879. job.content_type, job.content_encoding),
  880. accept_callback=job.on_accepted,
  881. timeout_callback=job.on_timeout,
  882. callback=job.on_success,
  883. error_callback=job.on_failure,
  884. soft_timeout=self.task.soft_time_limit,
  885. timeout=self.task.time_limit,
  886. correlation_id=job.id,
  887. )
  888. assert job._apply_result
  889. weakref_ref.assert_called_with(self.pool.apply_async())
  890. assert job._apply_result is weakref_ref()
  891. def test_execute_using_pool__defaults_of_hybrid_to_proto2(self):
  892. weakref_ref = Mock(name='weakref.ref')
  893. headers = strategy.hybrid_to_proto2('', {'id': uuid(),
  894. 'task': self.mytask.name})[1]
  895. job = self.zRequest(revoked_tasks=set(), ref=weakref_ref, **headers)
  896. job.execute_using_pool(self.pool)
  897. assert job._apply_result
  898. weakref_ref.assert_called_with(self.pool.apply_async())
  899. assert job._apply_result is weakref_ref()