test_beat.py 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768
  1. from __future__ import absolute_import, unicode_literals
  2. import errno
  3. import pytz
  4. from datetime import datetime, timedelta
  5. from pickle import dumps, loads
  6. import pytest
  7. from case import Mock, call, patch, skip
  8. from celery import __version__, beat, uuid
  9. from celery.beat import event_t
  10. from celery.five import keys, string_t
  11. from celery.schedules import crontab, schedule
  12. from celery.utils.objects import Bunch
  13. class MockShelve(dict):
  14. closed = False
  15. synced = False
  16. def close(self):
  17. self.closed = True
  18. def sync(self):
  19. self.synced = True
  20. class MockService(object):
  21. started = False
  22. stopped = False
  23. def __init__(self, *args, **kwargs):
  24. pass
  25. def start(self, **kwargs):
  26. self.started = True
  27. def stop(self, **kwargs):
  28. self.stopped = True
  29. class test_ScheduleEntry:
  30. Entry = beat.ScheduleEntry
  31. def create_entry(self, **kwargs):
  32. entry = {
  33. 'name': 'celery.unittest.add',
  34. 'schedule': timedelta(seconds=10),
  35. 'args': (2, 2),
  36. 'options': {'routing_key': 'cpu'},
  37. 'app': self.app,
  38. }
  39. return self.Entry(**dict(entry, **kwargs))
  40. def test_next(self):
  41. entry = self.create_entry(schedule=10)
  42. assert entry.last_run_at
  43. assert isinstance(entry.last_run_at, datetime)
  44. assert entry.total_run_count == 0
  45. next_run_at = entry.last_run_at + timedelta(seconds=10)
  46. next_entry = entry.next(next_run_at)
  47. assert next_entry.last_run_at >= next_run_at
  48. assert next_entry.total_run_count == 1
  49. def test_is_due(self):
  50. entry = self.create_entry(schedule=timedelta(seconds=10))
  51. assert entry.app is self.app
  52. assert entry.schedule.app is self.app
  53. due1, next_time_to_run1 = entry.is_due()
  54. assert not due1
  55. assert next_time_to_run1 > 9
  56. next_run_at = entry.last_run_at - timedelta(seconds=10)
  57. next_entry = entry.next(next_run_at)
  58. due2, next_time_to_run2 = next_entry.is_due()
  59. assert due2
  60. assert next_time_to_run2 > 9
  61. def test_repr(self):
  62. entry = self.create_entry()
  63. assert '<ScheduleEntry:' in repr(entry)
  64. def test_reduce(self):
  65. entry = self.create_entry(schedule=timedelta(seconds=10))
  66. fun, args = entry.__reduce__()
  67. res = fun(*args)
  68. assert res.schedule == entry.schedule
  69. def test_lt(self):
  70. e1 = self.create_entry(schedule=timedelta(seconds=10))
  71. e2 = self.create_entry(schedule=timedelta(seconds=2))
  72. # order doesn't matter, see comment in __lt__
  73. res1 = e1 < e2 # noqa
  74. try:
  75. res2 = e1 < object() # noqa
  76. except TypeError:
  77. pass
  78. def test_update(self):
  79. entry = self.create_entry()
  80. assert entry.schedule == timedelta(seconds=10)
  81. assert entry.args == (2, 2)
  82. assert entry.kwargs == {}
  83. assert entry.options == {'routing_key': 'cpu'}
  84. entry2 = self.create_entry(schedule=timedelta(minutes=20),
  85. args=(16, 16),
  86. kwargs={'callback': 'foo.bar.baz'},
  87. options={'routing_key': 'urgent'})
  88. entry.update(entry2)
  89. assert entry.schedule == schedule(timedelta(minutes=20))
  90. assert entry.args == (16, 16)
  91. assert entry.kwargs == {'callback': 'foo.bar.baz'}
  92. assert entry.options == {'routing_key': 'urgent'}
  93. class mScheduler(beat.Scheduler):
  94. def __init__(self, *args, **kwargs):
  95. self.sent = []
  96. beat.Scheduler.__init__(self, *args, **kwargs)
  97. def send_task(self, name=None, args=None, kwargs=None, **options):
  98. self.sent.append({'name': name,
  99. 'args': args,
  100. 'kwargs': kwargs,
  101. 'options': options})
  102. return self.app.AsyncResult(uuid())
  103. class mSchedulerSchedulingError(mScheduler):
  104. def send_task(self, *args, **kwargs):
  105. raise beat.SchedulingError('Could not apply task')
  106. class mSchedulerRuntimeError(mScheduler):
  107. def is_due(self, *args, **kwargs):
  108. raise RuntimeError('dict modified while itervalues')
  109. class mocked_schedule(schedule):
  110. def __init__(self, is_due, next_run_at, nowfun=datetime.utcnow):
  111. self._is_due = is_due
  112. self._next_run_at = next_run_at
  113. self.run_every = timedelta(seconds=1)
  114. self.nowfun = nowfun
  115. self.default_now = self.nowfun
  116. def is_due(self, last_run_at):
  117. return self._is_due, self._next_run_at
  118. always_due = mocked_schedule(True, 1)
  119. always_pending = mocked_schedule(False, 1)
  120. class test_Scheduler:
  121. def test_custom_schedule_dict(self):
  122. custom = {'foo': 'bar'}
  123. scheduler = mScheduler(app=self.app, schedule=custom, lazy=True)
  124. assert scheduler.data is custom
  125. def test_apply_async_uses_registered_task_instances(self):
  126. @self.app.task(shared=False)
  127. def foo():
  128. pass
  129. foo.apply_async = Mock(name='foo.apply_async')
  130. assert foo.name in foo._get_app().tasks
  131. scheduler = mScheduler(app=self.app)
  132. scheduler.apply_async(scheduler.Entry(task=foo.name, app=self.app))
  133. foo.apply_async.assert_called()
  134. def test_should_sync(self):
  135. @self.app.task(shared=False)
  136. def not_sync():
  137. pass
  138. not_sync.apply_async = Mock()
  139. s = mScheduler(app=self.app)
  140. s._do_sync = Mock()
  141. s.should_sync = Mock()
  142. s.should_sync.return_value = True
  143. s.apply_async(s.Entry(task=not_sync.name, app=self.app))
  144. s._do_sync.assert_called_with()
  145. s._do_sync = Mock()
  146. s.should_sync.return_value = False
  147. s.apply_async(s.Entry(task=not_sync.name, app=self.app))
  148. s._do_sync.assert_not_called()
  149. def test_should_sync_increments_sync_every_counter(self):
  150. self.app.conf.beat_sync_every = 2
  151. @self.app.task(shared=False)
  152. def not_sync():
  153. pass
  154. not_sync.apply_async = Mock()
  155. s = mScheduler(app=self.app)
  156. assert s.sync_every_tasks == 2
  157. s._do_sync = Mock()
  158. s.apply_async(s.Entry(task=not_sync.name, app=self.app))
  159. assert s._tasks_since_sync == 1
  160. s.apply_async(s.Entry(task=not_sync.name, app=self.app))
  161. s._do_sync.assert_called_with()
  162. self.app.conf.beat_sync_every = 0
  163. def test_sync_task_counter_resets_on_do_sync(self):
  164. self.app.conf.beat_sync_every = 1
  165. @self.app.task(shared=False)
  166. def not_sync():
  167. pass
  168. not_sync.apply_async = Mock()
  169. s = mScheduler(app=self.app)
  170. assert s.sync_every_tasks == 1
  171. s.apply_async(s.Entry(task=not_sync.name, app=self.app))
  172. assert s._tasks_since_sync == 0
  173. self.app.conf.beat_sync_every = 0
  174. @patch('celery.app.base.Celery.send_task')
  175. def test_send_task(self, send_task):
  176. b = beat.Scheduler(app=self.app)
  177. b.send_task('tasks.add', countdown=10)
  178. send_task.assert_called_with('tasks.add', countdown=10)
  179. def test_info(self):
  180. scheduler = mScheduler(app=self.app)
  181. assert isinstance(scheduler.info, string_t)
  182. def test_maybe_entry(self):
  183. s = mScheduler(app=self.app)
  184. entry = s.Entry(name='add every', task='tasks.add', app=self.app)
  185. assert s._maybe_entry(entry.name, entry) is entry
  186. assert s._maybe_entry('add every', {'task': 'tasks.add'})
  187. def test_set_schedule(self):
  188. s = mScheduler(app=self.app)
  189. s.schedule = {'foo': 'bar'}
  190. assert s.data == {'foo': 'bar'}
  191. @patch('kombu.connection.Connection.ensure_connection')
  192. def test_ensure_connection_error_handler(self, ensure):
  193. s = mScheduler(app=self.app)
  194. assert s._ensure_connected()
  195. ensure.assert_called()
  196. callback = ensure.call_args[0][0]
  197. callback(KeyError(), 5)
  198. def test_install_default_entries(self):
  199. self.app.conf.result_expires = None
  200. self.app.conf.beat_schedule = {}
  201. s = mScheduler(app=self.app)
  202. s.install_default_entries({})
  203. assert 'celery.backend_cleanup' not in s.data
  204. self.app.backend.supports_autoexpire = False
  205. self.app.conf.result_expires = 30
  206. s = mScheduler(app=self.app)
  207. s.install_default_entries({})
  208. assert 'celery.backend_cleanup' in s.data
  209. self.app.backend.supports_autoexpire = True
  210. self.app.conf.result_expires = 31
  211. s = mScheduler(app=self.app)
  212. s.install_default_entries({})
  213. assert 'celery.backend_cleanup' not in s.data
  214. def test_due_tick(self):
  215. scheduler = mScheduler(app=self.app)
  216. scheduler.add(name='test_due_tick',
  217. schedule=always_due,
  218. args=(1, 2),
  219. kwargs={'foo': 'bar'})
  220. assert scheduler.tick() == 0
  221. @patch('celery.beat.error')
  222. def test_due_tick_SchedulingError(self, error):
  223. scheduler = mSchedulerSchedulingError(app=self.app)
  224. scheduler.add(name='test_due_tick_SchedulingError',
  225. schedule=always_due)
  226. assert scheduler.tick() == 0
  227. error.assert_called()
  228. def test_pending_tick(self):
  229. scheduler = mScheduler(app=self.app)
  230. scheduler.add(name='test_pending_tick',
  231. schedule=always_pending)
  232. assert scheduler.tick() == 1 - 0.010
  233. def test_honors_max_interval(self):
  234. scheduler = mScheduler(app=self.app)
  235. maxi = scheduler.max_interval
  236. scheduler.add(name='test_honors_max_interval',
  237. schedule=mocked_schedule(False, maxi * 4))
  238. assert scheduler.tick() == maxi
  239. def test_ticks(self):
  240. scheduler = mScheduler(app=self.app)
  241. nums = [600, 300, 650, 120, 250, 36]
  242. s = {'test_ticks%s' % i: {'schedule': mocked_schedule(False, j)}
  243. for i, j in enumerate(nums)}
  244. scheduler.update_from_dict(s)
  245. assert scheduler.tick() == min(nums) - 0.010
  246. def test_ticks_microseconds(self):
  247. scheduler = mScheduler(app=self.app)
  248. now_ts = 1514797200.2
  249. now = datetime.fromtimestamp(now_ts)
  250. schedule_half = schedule(timedelta(seconds=0.5), nowfun=lambda: now)
  251. scheduler.add(name='half_second_schedule', schedule=schedule_half)
  252. scheduler.tick()
  253. # ensure those 0.2 seconds on now_ts don't get dropped
  254. expected_time = now_ts + 0.5 - 0.010
  255. assert scheduler._heap[0].time == expected_time
  256. def test_ticks_schedule_change(self):
  257. # initialise schedule and check heap is not initialized
  258. scheduler = mScheduler(app=self.app)
  259. assert scheduler._heap is None
  260. # set initial schedule and check heap is updated
  261. schedule_5 = schedule(5)
  262. scheduler.add(name='test_schedule', schedule=schedule_5)
  263. scheduler.tick()
  264. assert scheduler._heap[0].entry.schedule == schedule_5
  265. # update schedule and check heap is updated
  266. schedule_10 = schedule(10)
  267. scheduler.add(name='test_schedule', schedule=schedule(10))
  268. scheduler.tick()
  269. assert scheduler._heap[0].entry.schedule == schedule_10
  270. def test_schedule_no_remain(self):
  271. scheduler = mScheduler(app=self.app)
  272. scheduler.add(name='test_schedule_no_remain',
  273. schedule=mocked_schedule(False, None))
  274. assert scheduler.tick() == scheduler.max_interval
  275. def test_interface(self):
  276. scheduler = mScheduler(app=self.app)
  277. scheduler.sync()
  278. scheduler.setup_schedule()
  279. scheduler.close()
  280. def test_merge_inplace(self):
  281. a = mScheduler(app=self.app)
  282. b = mScheduler(app=self.app)
  283. a.update_from_dict({'foo': {'schedule': mocked_schedule(True, 10)},
  284. 'bar': {'schedule': mocked_schedule(True, 20)}})
  285. b.update_from_dict({'bar': {'schedule': mocked_schedule(True, 40)},
  286. 'baz': {'schedule': mocked_schedule(True, 10)}})
  287. a.merge_inplace(b.schedule)
  288. assert 'foo' not in a.schedule
  289. assert 'baz' in a.schedule
  290. assert a.schedule['bar'].schedule._next_run_at == 40
  291. def test_when(self):
  292. now_time_utc = datetime(2000, 10, 10, 10, 10, 10, 10, tzinfo=pytz.utc)
  293. now_time_casey = now_time_utc.astimezone(
  294. pytz.timezone('Antarctica/Casey')
  295. )
  296. scheduler = mScheduler(app=self.app)
  297. result_utc = scheduler._when(
  298. mocked_schedule(True, 10, lambda: now_time_utc),
  299. 10
  300. )
  301. result_casey = scheduler._when(
  302. mocked_schedule(True, 10, lambda: now_time_casey),
  303. 10
  304. )
  305. assert result_utc == result_casey
  306. @patch('celery.beat.Scheduler._when', return_value=1)
  307. def test_populate_heap(self, _when):
  308. scheduler = mScheduler(app=self.app)
  309. scheduler.update_from_dict(
  310. {'foo': {'schedule': mocked_schedule(True, 10)}}
  311. )
  312. scheduler.populate_heap()
  313. assert scheduler._heap == [event_t(1, 5, scheduler.schedule['foo'])]
  314. def create_schedule_entry(self, schedule=None, args=(), kwargs={},
  315. options={}, task=None):
  316. entry = {
  317. 'name': 'celery.unittest.add',
  318. 'schedule': schedule,
  319. 'app': self.app,
  320. 'args': args,
  321. 'kwargs': kwargs,
  322. 'options': options,
  323. 'task': task
  324. }
  325. return beat.ScheduleEntry(**dict(entry))
  326. def test_schedule_equal_schedule_vs_schedule_success(self):
  327. scheduler = beat.Scheduler(app=self.app)
  328. a = {'a': self.create_schedule_entry(schedule=schedule(5))}
  329. b = {'a': self.create_schedule_entry(schedule=schedule(5))}
  330. assert scheduler.schedules_equal(a, b)
  331. def test_schedule_equal_schedule_vs_schedule_fail(self):
  332. scheduler = beat.Scheduler(app=self.app)
  333. a = {'a': self.create_schedule_entry(schedule=schedule(5))}
  334. b = {'a': self.create_schedule_entry(schedule=schedule(10))}
  335. assert not scheduler.schedules_equal(a, b)
  336. def test_schedule_equal_crontab_vs_crontab_success(self):
  337. scheduler = beat.Scheduler(app=self.app)
  338. a = {'a': self.create_schedule_entry(schedule=crontab(minute=5))}
  339. b = {'a': self.create_schedule_entry(schedule=crontab(minute=5))}
  340. assert scheduler.schedules_equal(a, b)
  341. def test_schedule_equal_crontab_vs_crontab_fail(self):
  342. scheduler = beat.Scheduler(app=self.app)
  343. a = {'a': self.create_schedule_entry(schedule=crontab(minute=5))}
  344. b = {'a': self.create_schedule_entry(schedule=crontab(minute=10))}
  345. assert not scheduler.schedules_equal(a, b)
  346. def test_schedule_equal_crontab_vs_schedule_fail(self):
  347. scheduler = beat.Scheduler(app=self.app)
  348. a = {'a': self.create_schedule_entry(schedule=crontab(minute=5))}
  349. b = {'a': self.create_schedule_entry(schedule=schedule(5))}
  350. assert not scheduler.schedules_equal(a, b)
  351. def test_schedule_equal_different_key_fail(self):
  352. scheduler = beat.Scheduler(app=self.app)
  353. a = {'a': self.create_schedule_entry(schedule=schedule(5))}
  354. b = {'b': self.create_schedule_entry(schedule=schedule(5))}
  355. assert not scheduler.schedules_equal(a, b)
  356. def test_schedule_equal_args_vs_args_success(self):
  357. scheduler = beat.Scheduler(app=self.app)
  358. a = {'a': self.create_schedule_entry(args='a')}
  359. b = {'a': self.create_schedule_entry(args='a')}
  360. assert scheduler.schedules_equal(a, b)
  361. def test_schedule_equal_args_vs_args_fail(self):
  362. scheduler = beat.Scheduler(app=self.app)
  363. a = {'a': self.create_schedule_entry(args='a')}
  364. b = {'a': self.create_schedule_entry(args='b')}
  365. assert not scheduler.schedules_equal(a, b)
  366. def test_schedule_equal_kwargs_vs_kwargs_success(self):
  367. scheduler = beat.Scheduler(app=self.app)
  368. a = {'a': self.create_schedule_entry(kwargs={'a': 'a'})}
  369. b = {'a': self.create_schedule_entry(kwargs={'a': 'a'})}
  370. assert scheduler.schedules_equal(a, b)
  371. def test_schedule_equal_kwargs_vs_kwargs_fail(self):
  372. scheduler = beat.Scheduler(app=self.app)
  373. a = {'a': self.create_schedule_entry(kwargs={'a': 'a'})}
  374. b = {'a': self.create_schedule_entry(kwargs={'b': 'b'})}
  375. assert not scheduler.schedules_equal(a, b)
  376. def test_schedule_equal_options_vs_options_success(self):
  377. scheduler = beat.Scheduler(app=self.app)
  378. a = {'a': self.create_schedule_entry(options={'a': 'a'})}
  379. b = {'a': self.create_schedule_entry(options={'a': 'a'})}
  380. assert scheduler.schedules_equal(a, b)
  381. def test_schedule_equal_options_vs_options_fail(self):
  382. scheduler = beat.Scheduler(app=self.app)
  383. a = {'a': self.create_schedule_entry(options={'a': 'a'})}
  384. b = {'a': self.create_schedule_entry(options={'b': 'b'})}
  385. assert not scheduler.schedules_equal(a, b)
  386. def test_schedule_equal_task_vs_task_success(self):
  387. scheduler = beat.Scheduler(app=self.app)
  388. a = {'a': self.create_schedule_entry(task='a')}
  389. b = {'a': self.create_schedule_entry(task='a')}
  390. assert scheduler.schedules_equal(a, b)
  391. def test_schedule_equal_task_vs_task_fail(self):
  392. scheduler = beat.Scheduler(app=self.app)
  393. a = {'a': self.create_schedule_entry(task='a')}
  394. b = {'a': self.create_schedule_entry(task='b')}
  395. assert not scheduler.schedules_equal(a, b)
  396. def create_persistent_scheduler(shelv=None):
  397. if shelv is None:
  398. shelv = MockShelve()
  399. class MockPersistentScheduler(beat.PersistentScheduler):
  400. sh = shelv
  401. persistence = Bunch(
  402. open=lambda *a, **kw: shelv,
  403. )
  404. tick_raises_exit = False
  405. shutdown_service = None
  406. def tick(self):
  407. if self.tick_raises_exit:
  408. raise SystemExit()
  409. if self.shutdown_service:
  410. self.shutdown_service._is_shutdown.set()
  411. return 0.0
  412. return MockPersistentScheduler, shelv
  413. def create_persistent_scheduler_w_call_logging(shelv=None):
  414. if shelv is None:
  415. shelv = MockShelve()
  416. class MockPersistentScheduler(beat.PersistentScheduler):
  417. sh = shelv
  418. persistence = Bunch(
  419. open=lambda *a, **kw: shelv,
  420. )
  421. def __init__(self, *args, **kwargs):
  422. self.sent = []
  423. beat.PersistentScheduler.__init__(self, *args, **kwargs)
  424. def send_task(self, task=None, args=None, kwargs=None, **options):
  425. self.sent.append({'task': task,
  426. 'args': args,
  427. 'kwargs': kwargs,
  428. 'options': options})
  429. return self.app.AsyncResult(uuid())
  430. return MockPersistentScheduler, shelv
  431. class test_PersistentScheduler:
  432. @patch('os.remove')
  433. def test_remove_db(self, remove):
  434. s = create_persistent_scheduler()[0](app=self.app,
  435. schedule_filename='schedule')
  436. s._remove_db()
  437. remove.assert_has_calls(
  438. [call('schedule' + suffix) for suffix in s.known_suffixes]
  439. )
  440. err = OSError()
  441. err.errno = errno.ENOENT
  442. remove.side_effect = err
  443. s._remove_db()
  444. err.errno = errno.EPERM
  445. with pytest.raises(OSError):
  446. s._remove_db()
  447. def test_setup_schedule(self):
  448. s = create_persistent_scheduler()[0](app=self.app,
  449. schedule_filename='schedule')
  450. opens = s.persistence.open = Mock()
  451. s._remove_db = Mock()
  452. def effect(*args, **kwargs):
  453. if opens.call_count > 1:
  454. return s.sh
  455. raise OSError()
  456. opens.side_effect = effect
  457. s.setup_schedule()
  458. s._remove_db.assert_called_with()
  459. s._store = {str('__version__'): 1}
  460. s.setup_schedule()
  461. s._store.clear = Mock()
  462. op = s.persistence.open = Mock()
  463. op.return_value = s._store
  464. s._store[str('tz')] = 'FUNKY'
  465. s.setup_schedule()
  466. op.assert_called_with(s.schedule_filename, writeback=True)
  467. s._store.clear.assert_called_with()
  468. s._store[str('utc_enabled')] = False
  469. s._store.clear = Mock()
  470. s.setup_schedule()
  471. s._store.clear.assert_called_with()
  472. def test_get_schedule(self):
  473. s = create_persistent_scheduler()[0](
  474. schedule_filename='schedule', app=self.app,
  475. )
  476. s._store = {str('entries'): {}}
  477. s.schedule = {'foo': 'bar'}
  478. assert s.schedule == {'foo': 'bar'}
  479. assert s._store[str('entries')] == s.schedule
  480. def test_run_all_due_tasks_after_restart(self):
  481. scheduler_class, shelve = create_persistent_scheduler_w_call_logging()
  482. shelve['tz'] = 'UTC'
  483. shelve['utc_enabled'] = True
  484. shelve['__version__'] = __version__
  485. cur_seconds = 20
  486. def now_func():
  487. return datetime(2018, 1, 1, 1, 11, cur_seconds)
  488. app_schedule = {
  489. 'first_missed': {'schedule': crontab(
  490. minute='*/10', nowfun=now_func), 'task': 'first_missed'},
  491. 'second_missed': {'schedule': crontab(
  492. minute='*/1', nowfun=now_func), 'task': 'second_missed'},
  493. 'non_missed': {'schedule': crontab(
  494. minute='*/13', nowfun=now_func), 'task': 'non_missed'}
  495. }
  496. shelve['entries'] = {
  497. 'first_missed': beat.ScheduleEntry(
  498. 'first_missed', 'first_missed',
  499. last_run_at=now_func() - timedelta(minutes=2),
  500. total_run_count=10,
  501. schedule=app_schedule['first_missed']['schedule']),
  502. 'second_missed': beat.ScheduleEntry(
  503. 'second_missed', 'second_missed',
  504. last_run_at=now_func() - timedelta(minutes=2),
  505. total_run_count=10,
  506. schedule=app_schedule['second_missed']['schedule']),
  507. 'non_missed': beat.ScheduleEntry(
  508. 'non_missed', 'non_missed',
  509. last_run_at=now_func() - timedelta(minutes=2),
  510. total_run_count=10,
  511. schedule=app_schedule['non_missed']['schedule']),
  512. }
  513. self.app.conf.beat_schedule = app_schedule
  514. scheduler = scheduler_class(self.app)
  515. max_iter_number = 5
  516. for i in range(max_iter_number):
  517. delay = scheduler.tick()
  518. if delay > 0:
  519. break
  520. assert {'first_missed', 'second_missed'} == {
  521. item['task'] for item in scheduler.sent}
  522. # ensure next call on the beginning of next min
  523. assert abs(60 - cur_seconds - delay) < 1
  524. class test_Service:
  525. def get_service(self):
  526. Scheduler, mock_shelve = create_persistent_scheduler()
  527. return beat.Service(app=self.app, scheduler_cls=Scheduler), mock_shelve
  528. def test_pickleable(self):
  529. s = beat.Service(app=self.app, scheduler_cls=Mock)
  530. assert loads(dumps(s))
  531. def test_start(self):
  532. s, sh = self.get_service()
  533. schedule = s.scheduler.schedule
  534. assert isinstance(schedule, dict)
  535. assert isinstance(s.scheduler, beat.Scheduler)
  536. scheduled = list(schedule.keys())
  537. for task_name in keys(sh[str('entries')]):
  538. assert task_name in scheduled
  539. s.sync()
  540. assert sh.closed
  541. assert sh.synced
  542. assert s._is_stopped.isSet()
  543. s.sync()
  544. s.stop(wait=False)
  545. assert s._is_shutdown.isSet()
  546. s.stop(wait=True)
  547. assert s._is_shutdown.isSet()
  548. p = s.scheduler._store
  549. s.scheduler._store = None
  550. try:
  551. s.scheduler.sync()
  552. finally:
  553. s.scheduler._store = p
  554. def test_start_embedded_process(self):
  555. s, sh = self.get_service()
  556. s._is_shutdown.set()
  557. s.start(embedded_process=True)
  558. def test_start_thread(self):
  559. s, sh = self.get_service()
  560. s._is_shutdown.set()
  561. s.start(embedded_process=False)
  562. def test_start_tick_raises_exit_error(self):
  563. s, sh = self.get_service()
  564. s.scheduler.tick_raises_exit = True
  565. s.start()
  566. assert s._is_shutdown.isSet()
  567. def test_start_manages_one_tick_before_shutdown(self):
  568. s, sh = self.get_service()
  569. s.scheduler.shutdown_service = s
  570. s.start()
  571. assert s._is_shutdown.isSet()
  572. class test_EmbeddedService:
  573. @skip.unless_module('_multiprocessing', name='multiprocessing')
  574. def xxx_start_stop_process(self):
  575. from billiard.process import Process
  576. s = beat.EmbeddedService(self.app)
  577. assert isinstance(s, Process)
  578. assert isinstance(s.service, beat.Service)
  579. s.service = MockService()
  580. class _Popen(object):
  581. terminated = False
  582. def terminate(self):
  583. self.terminated = True
  584. with patch('celery.platforms.close_open_fds'):
  585. s.run()
  586. assert s.service.started
  587. s._popen = _Popen()
  588. s.stop()
  589. assert s.service.stopped
  590. assert s._popen.terminated
  591. def test_start_stop_threaded(self):
  592. s = beat.EmbeddedService(self.app, thread=True)
  593. from threading import Thread
  594. assert isinstance(s, Thread)
  595. assert isinstance(s.service, beat.Service)
  596. s.service = MockService()
  597. s.run()
  598. assert s.service.started
  599. s.stop()
  600. assert s.service.stopped
  601. class test_schedule:
  602. def test_maybe_make_aware(self):
  603. x = schedule(10, app=self.app)
  604. x.utc_enabled = True
  605. d = x.maybe_make_aware(datetime.utcnow())
  606. assert d.tzinfo
  607. x.utc_enabled = False
  608. d2 = x.maybe_make_aware(datetime.utcnow())
  609. assert d2.tzinfo
  610. def test_to_local(self):
  611. x = schedule(10, app=self.app)
  612. x.utc_enabled = True
  613. d = x.to_local(datetime.utcnow())
  614. assert d.tzinfo is None
  615. x.utc_enabled = False
  616. d = x.to_local(datetime.utcnow())
  617. assert d.tzinfo