

def test_job_with_repeat(self):
    '\n        Ensure jobs with repeat attribute are put back in the scheduler\n        X (repeat) number of times\n        '
    time_now = datetime.utcnow()
    interval = 10
    job = self.scheduler.schedule(time_now, say_hello, interval=interval, repeat=1)
    self.scheduler.enqueue_job(job)
    self.assertNotIn(job.id, tl(self.testconn.zrange(self.scheduler.scheduled_jobs_key, 0, 1)))
    job = self.scheduler.schedule(time_now, say_hello, interval=interval, repeat=2)
    self.scheduler.enqueue_job(job)
    self.assertIn(job.id, tl(self.testconn.zrange(self.scheduler.scheduled_jobs_key, 0, 1)))
    self.scheduler.enqueue_job(time_now)
    self.assertNotIn(job.id, tl(self.testconn.zrange(self.scheduler.scheduled_jobs_key, 0, 1)))
    time_now = datetime.utcnow()
    job = self.scheduler.enqueue_periodic(time_now, interval, 1, say_hello)
    self.scheduler.enqueue_job(job)
    self.assertNotIn(job.id, tl(self.testconn.zrange(self.scheduler.scheduled_jobs_key, 0, 1)))
    job = self.scheduler.enqueue_periodic(time_now, interval, 2, say_hello)
    self.scheduler.enqueue_job(job)
    self.assertIn(job.id, tl(self.testconn.zrange(self.scheduler.scheduled_jobs_key, 0, 1)))
    self.scheduler.enqueue_job(job)
    self.assertNotIn(job.id, tl(self.testconn.zrange(self.scheduler.scheduled_jobs_key, 0, 1)))

dataset/ETHPy150Open ui/rq-scheduler/tests/test_scheduler.py TestScheduler.test_job_with_repeat/VarMisuse@148/220 `job`->`time_now`