#NEWLINE# @ pytest . mark . django_db #NEWLINE# def test_scheduler( broker ) : #NEWLINE# #INDENT# broker . list_key = 'scheduler_test:q' #NEWLINE# broker . delete_queue ( ) #NEWLINE# schedule = create_schedule ( 'math.copysign' , 1 , ( - 1 ) , name = 'test math' , hook = 'django_q.tests.tasks.result' , schedule_type = Schedule . HOURLY , repeats = 1 ) #NEWLINE# assert ( schedule . last_run ( ) is None ) #NEWLINE# with pytest . raises ( IntegrityError ) : #NEWLINE# #INDENT# schedule = create_schedule ( 'math.copysign' , 1 , ( - 1 ) , name = 'test math' , hook = 'django_q.tests.tasks.result' , schedule_type = Schedule . HOURLY , repeats = 1 ) #NEWLINE# #UNINDENT# scheduler ( broker = broker ) #NEWLINE# task_queue = Queue ( ) #NEWLINE# stop_event = Event ( ) #NEWLINE# stop_event . set ( ) #NEWLINE# pusher ( task_queue , stop_event , broker = broker ) #NEWLINE# assert ( task_queue . qsize ( ) == 1 ) #NEWLINE# assert ( broker . queue_size ( ) == 0 ) #NEWLINE# task_queue . put ( 'STOP' ) #NEWLINE# result_queue = Queue ( ) #NEWLINE# worker ( task_queue , result_queue , Value ( 'b' , ( - 1 ) ) ) #NEWLINE# assert ( result_queue . qsize ( ) == 1 ) #NEWLINE# result_queue . put ( 'STOP' ) #NEWLINE# monitor ( result_queue ) #NEWLINE# assert ( result_queue . qsize ( ) == 0 ) #NEWLINE# schedule = Schedule . objects . get ( pk = schedule . pk ) #NEWLINE# assert ( schedule . repeats == 0 ) #NEWLINE# assert ( schedule . last_run ( ) is not None ) #NEWLINE# assert ( schedule . success ( ) is True ) #NEWLINE# assert ( schedule . next_run < arrow . get ( timezone . now ( ) ) . replace ( hours = ( + 1 ) ) ) #NEWLINE# task = fetch ( schedule . task ) #NEWLINE# assert ( task is not None ) #NEWLINE# assert ( task . success is True ) #NEWLINE# assert ( task . result < 0 ) #NEWLINE# once_schedule = create_schedule ( 'django_q.tests.tasks.word_multiply' , 2 , word = 'django' , schedule_type = Schedule . ONCE , repeats = ( - 1 ) , hook = 'django_q.tests.tasks.result' ) #NEWLINE# assert ( hasattr ( once_schedule , 'pk' ) is True ) #NEWLINE# always_schedule = create_schedule ( 'django_q.tests.tasks.word_multiply' , 2 , word = 'django' , schedule_type = Schedule . DAILY , repeats = ( - 1 ) , hook = 'django_q.tests.tasks.result' ) #NEWLINE# assert ( hasattr ( always_schedule , 'pk' ) is True ) #NEWLINE# minute_schedule = create_schedule ( 'django_q.tests.tasks.word_multiply' , 2 , word = 'django' , schedule_type = Schedule . MINUTES , minutes = 10 ) #NEWLINE# assert ( hasattr ( minute_schedule , 'pk' ) is True ) #NEWLINE# for t in Schedule . TYPE : #NEWLINE# #INDENT# schedule = create_schedule ( 'django_q.tests.tasks.word_multiply' , 2 , word = 'django' , schedule_type = t [ 0 ] , repeats = 1 , hook = 'django_q.tests.tasks.result' ) #NEWLINE# assert ( schedule is not None ) #NEWLINE# assert ( schedule . last_run ( ) is None ) #NEWLINE# scheduler ( broker = broker ) #NEWLINE# #UNINDENT# Schedule . objects . create ( func = 'django_q.tests.tasks.word_multiply' , args = '2' , kwargs = 'word="django"' , schedule_type = Schedule . DAILY ) #NEWLINE# scheduler ( broker = broker ) #NEWLINE# assert ( Schedule . objects . filter ( pk = once_schedule . pk ) . exists ( ) is False ) #NEWLINE# Conf . CATCH_UP = True #NEWLINE# now = timezone . now ( ) #NEWLINE# schedule = create_schedule ( 'django_q.tests.tasks.word_multiply' , 2 , word = 'catch_up' , schedule_type = Schedule . HOURLY , next_run = ( timezone . now ( ) - timedelta ( hours = 12 ) ) , repeats = ( - 1 ) ) #NEWLINE# scheduler ( broker = broker ) #NEWLINE# schedule = Schedule . objects . get ( pk = schedule . pk ) #NEWLINE# assert ( schedule . next_run < now ) #NEWLINE# Conf . CATCH_UP = False #NEWLINE# scheduler ( broker = broker ) #NEWLINE# schedule = Schedule . objects . get ( pk = schedule . pk ) #NEWLINE# assert ( schedule . next_run > now ) #NEWLINE# broker . delete_queue ( )