#NEWLINE# @ pytest . mark . parametrize ( 'task_name,expected' , PARAMS ) #NEWLINE# def test_collision( task_name , expected ) : #NEWLINE# #INDENT# manager_class = _select_manager ( celery . backend . __class__ . __name__ ) #NEWLINE# manager_instance = list ( ) #NEWLINE# task = celery . tasks [ task_name ] #NEWLINE# def new_exit( self , * _ ) : #NEWLINE# #INDENT# manager_instance . append ( self ) #NEWLINE# return None #NEWLINE# #UNINDENT# original_exit = manager_class . __exit__ #NEWLINE# setattr ( manager_class , '__exit__' , new_exit ) #NEWLINE# assert ( e == task . apply_async ( args = ( 4 , 4 ) ) . get ( ) ) #NEWLINE# setattr ( manager_class , '__exit__' , original_exit ) #NEWLINE# assert ( manager_instance [ 0 ] . is_already_running is True ) #NEWLINE# with pytest . raises ( OtherInstanceError ) as e : #NEWLINE# #INDENT# task . apply_async ( args = ( 4 , 4 ) ) . get ( ) #NEWLINE# #UNINDENT# if manager_instance [ 0 ] . include_args : #NEWLINE# #INDENT# assert str ( e . value ) . startswith ( 'Failed to acquire lock, {0}.args.' . format ( task_name ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# assert ( 'Failed to acquire lock, {0} already running.' . format ( task_name ) == str ( e . value ) ) #NEWLINE# #UNINDENT# assert ( manager_instance [ 0 ] . is_already_running is True ) #NEWLINE# manager_instance [ 0 ] . reset_lock ( ) #NEWLINE# assert ( manager_instance [ 0 ] . is_already_running is False ) #NEWLINE# assert ( expected == task . apply_async ( args = ( 4 , 4 ) ) . get ( ) )