MCPcopy
hub / github.com/celery/celery / test_on_timer_tick

Method test_on_timer_tick

t/unit/worker/test_worker.py:941–948  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

939 assert 'KeyError' in msg % args
940
941 def test_on_timer_tick(self):
942 worker = self.app.WorkController(concurrency=1, loglevel=10)
943
944 components.Timer(worker).on_timer_tick(30.0)
945 xargs = self.comp_logger.debug.call_args[0]
946 fmt, arg = xargs[0], xargs[1]
947 assert arg == 30.0
948 assert 'Next ETA %s secs' in fmt
949
950 def test_process_task(self):
951 worker = self.worker

Callers

nothing calls this directly

Calls 2

WorkControllerMethod · 0.80
on_timer_tickMethod · 0.80

Tested by

no test coverage detected