#NEWLINE# @ pytest . mark . parametrize ( 'task_name,timeout' , [ ( 'tests.instances.mul' , 20 ) , ( 'tests.instances.add' , 300 ) , ( 'tests.instances.add2' , 70 ) , ( 'tests.instances.add3' , 80 ) ] ) #NEWLINE# def test_instances( task_name , timeout ) : #NEWLINE# #INDENT# manager_class = _select_manager ( celery . backend . __class__ . __name__ ) #NEWLINE# manager_instance = list ( ) #NEWLINE# task = celery . tasks [ task_name ] #NEWLINE# original_exit = manager_class . __exit__ #NEWLINE# def new_exit( self , * _ ) : #NEWLINE# #INDENT# manager_instance . append ( self ) #NEWLINE# return original_exit ( self , * _ ) #NEWLINE# #UNINDENT# setattr ( manager_class , '__exit__' , new_exit ) #NEWLINE# task . apply_async ( args = ( 4 , 4 ) ) . get ( ) #NEWLINE# setattr ( manager_class , '__exit__' , original_exit ) #NEWLINE# assert ( timeout == manager_instance [ 0 ] . timeout )