<s> from donkey . job import Schedule , Job , JobQueue <EOL> def test_schedule ( ) : <EOL> interval = <NUM_LIT:30> <EOL> sched = Schedule ( interval ) <EOL> assert sched . next ( <NUM_LIT:0> ) == <NUM_LIT:0> + interval <EOL> assert sched . next ( <NUM_LIT> ) == <NUM_LIT> + interval <EOL> def test_job_run ( ) : <EOL> expected_rv = <NUM_LIT> <EOL> job = Job ( lambda : expected_rv , Schedule ( <NUM_LIT:30> ) ) <EOL> assert job . run ( ) == expected_rv <EOL> def test_job_run_stats ( ) : <EOL> interval = <NUM_LIT:30> <EOL> job = Job ( lambda : '<STR_LIT>' , Schedule ( interval ) ) <EOL> assert job . last_run_at is None <EOL> assert job . next_run_at is None <EOL> job . reschedule ( <NUM_LIT:0> ) <EOL> assert job . last_run_at is None <EOL> assert job . next_run_at == <NUM_LIT:0> + interval <EOL> job . reschedule ( interval ) <EOL> assert job . last_run_at == <NUM_LIT:0> + interval <EOL> assert job . next_run_at == interval + interval <EOL> def test_job_queue_add ( ) : <EOL> q = JobQueue ( ) <EOL> q . add ( Job ( lambda : '<STR_LIT>' , Schedule ( <NUM_LIT:30> ) ) ) <EOL> assert len ( q . get_runnable_jobs ( ) ) == <NUM_LIT:1> <EOL> def test_job_queue_job_decorator ( ) : <EOL> q = JobQueue ( ) <EOL> @ q . job ( <NUM_LIT:30> ) <EOL> def nop ( ) : <EOL> pass <EOL> assert len ( q . get_runnable_jobs ( ) ) == <NUM_LIT:1> <EOL> def test_job_queue_next_run_at ( ) : <EOL> q = JobQueue ( ) <EOL> assert q . next_run_at is None <EOL> q . job ( <NUM_LIT:30> ) ( lambda : '<STR_LIT>' ) <EOL> q . reschedule_jobs ( <NUM_LIT:0> ) <EOL> assert q . next_run_at is not None </s>
