#NEWLINE# @ pytest . mark . parametrize ( [ 'p_query' ] , OPTS ) #NEWLINE# def test_cancel_by_path( worker , p_query ) : #NEWLINE# #INDENT# expected_action_jobs = p_query [ 1 ] #NEWLINE# worker . start ( flags = '--greenlets 1' , queues = 'default q1 q2' ) #NEWLINE# job_ids = [ ] #NEWLINE# job_ids . append ( worker . send_task ( 'tests.tasks.general.Add' , { #NEWLINE# #INDENT# 'a' : 41 , #NEWLINE# 'b' : 1 , #NEWLINE# 'sleep' : 2 , #NEWLINE# #UNINDENT# } , queue = 'default' , block = False ) ) #NEWLINE# params = { #NEWLINE# #INDENT# 'action' : 'cancel' , #NEWLINE# 'status' : 'queued' , #NEWLINE# #UNINDENT# } #NEWLINE# params . update ( p_query [ 0 ] ) #NEWLINE# requeue_job = worker . send_task ( 'mrq.basetasks.utils.JobAction' , p_query , block = False ) #NEWLINE# job_ids . append ( worker . send_task ( 'tests.tasks.general.MongoInsert' , { #NEWLINE# #INDENT# 'a' : 42 , #NEWLINE# #UNINDENT# } , queue = 'q1' , block = False ) ) #NEWLINE# job_ids . append ( worker . send_task ( 'tests.tasks.general.MongoInsert' , { #NEWLINE# #INDENT# 'a' : 42 , #NEWLINE# #UNINDENT# } , queue = 'q2' , block = False ) ) #NEWLINE# job_ids . append ( worker . send_task ( 'tests.tasks.general.MongoInsert' , { #NEWLINE# #INDENT# 'a' : 43 , #NEWLINE# #UNINDENT# } , queue = 'q2' , block = False ) ) #NEWLINE# job_ids . append ( worker . send_task ( 'tests.tasks.general.MongoInsert2' , { #NEWLINE# #INDENT# 'a' : 44 , #NEWLINE# #UNINDENT# } , queue = 'q1' , block = False ) ) #NEWLINE# Job ( job_ids [ ( - 1 ) ] ) . wait ( poll_interval = 0.01 ) #NEWLINE# time . sleep ( 1 ) #NEWLINE# worker . stop ( deps = False ) #NEWLINE# jobs = [ Job ( job_id ) . fetch ( ) . data for job_id in job_ids ] #NEWLINE# assert ( jobs [ 0 ] [ 'status' ] == 'success' ) #NEWLINE# assert ( jobs [ 0 ] [ 'result' ] == 42 ) #NEWLINE# assert ( Job ( requeue_job ) . fetch ( ) . data [ 'result' ] [ 'cancelled' ] == expected_action_jobs ) #NEWLINE# assert ( worker . mongodb_jobs . tests_inserts . count ( ) == ( ( len ( job_ids ) - 1 ) - expected_action_jobs ) ) #NEWLINE# action_jobs = list ( worker . mongodb_jobs . mrq_jobs . find ( { #NEWLINE# #INDENT# 'status' : 'cancel' , #NEWLINE# #UNINDENT# } ) ) #NEWLINE# assert ( len ( action_jobs ) == expected_action_jobs ) #NEWLINE# assert ( set ( [ x . get ( 'result' ) for x in action_jobs ] ) == set ( [ None ] ) ) #NEWLINE# assert ( Queue ( 'default' ) . size ( ) == 0 ) #NEWLINE# assert ( Queue ( 'q1' ) . size ( ) == 0 ) #NEWLINE# assert ( Queue ( 'q2' ) . size ( ) == 0 ) #NEWLINE# worker . mongodb_jobs . tests_inserts . remove ( { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# params = { #NEWLINE# #INDENT# 'action' : 'requeue' , #NEWLINE# #UNINDENT# } #NEWLINE# params . update ( p_query [ 0 ] ) #NEWLINE# worker . start ( flags = '--gevent 1' , start_deps = False , queues = 'default' , flush = False ) #NEWLINE# ret = worker . send_task ( 'mrq.basetasks.utils.JobAction' , params , block = True ) #NEWLINE# assert ( ret [ 'requeued' ] == expected_action_jobs ) #NEWLINE# worker . stop ( deps = False ) #NEWLINE# assert ( worker . mongodb_jobs . mrq_jobs . find ( { #NEWLINE# #INDENT# 'status' : 'queued' , #NEWLINE# #UNINDENT# } ) . count ( ) == expected_action_jobs ) #NEWLINE# assert ( ( ( Queue ( 'default' ) . size ( ) + Queue ( 'q1' ) . size ( ) ) + Queue ( 'q2' ) . size ( ) ) == expected_action_jobs ) #NEWLINE# worker . stop_deps ( )