#NEWLINE# @ pytest . mark . parametrize ( 'jobstore' , [ None , 'alter' , 'third' ] , ids = [ 'all' , 'single jobstore' , 'nonexistent store' ] ) #NEWLINE# def test_remove_all_jobs( self , scheduler , jobstore ) : #NEWLINE# #INDENT# '\n        Tests that remove_all_jobs() removes all jobs from all attached\n        job stores, plus any pending jobs.\n\n        ' #NEWLINE# scheduler . _jobstores = OrderedDict ( [ ( 'default' , MagicMock ( BaseJobStore ) ) , ( 'alter' , MagicMock ( BaseJobStore ) ) ] ) #NEWLINE# scheduler . _pending_jobs = [ ( MagicMock ( Job ) , 'default' , False ) , ( MagicMock ( Job ) , 'alter' , False ) , ( MagicMock ( Job ) , 'third' , False ) ] #NEWLINE# scheduler . _dispatch_event = MagicMock ( ) #NEWLINE# scheduler . remove_all_jobs ( jobstore ) #NEWLINE# if ( jobstore == 'alter' ) : #NEWLINE# #INDENT# scheduler . _jobstores [ 'alter' ] . remove_all_jobs . assert_called_once_with ( ) #NEWLINE# assert ( scheduler . _jobstores [ 'default' ] . remove_all_jobs . call_count == 0 ) #NEWLINE# assert ( len ( scheduler . _pending_jobs ) == 2 ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( jobstore == 'third' ) : #NEWLINE# #INDENT# assert ( scheduler . _jobstores [ 'alter' ] . remove_all_jobs . call_count == 0 ) #NEWLINE# assert ( scheduler . _jobstores [ 'default' ] . remove_all_jobs . call_count == 0 ) #NEWLINE# assert ( len ( scheduler . _pending_jobs ) == 2 ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# scheduler . _jobstores [ 'alter' ] . remove_all_jobs . assert_called_once_with ( ) #NEWLINE# scheduler . _jobstores [ 'default' ] . remove_all_jobs . assert_called_once_with ( ) #NEWLINE# assert ( len ( scheduler . _pending_jobs ) == 0 ) #NEWLINE# #UNINDENT# #UNINDENT# assert ( scheduler . _dispatch_event . call_count == 1 ) #NEWLINE# event = scheduler . _dispatch_event . call_args [ 0 ] [ 0 ] #NEWLINE# assert ( event . code == EVENT_ALL_JOBS_REMOVED ) #NEWLINE# assert ( event . alias == jobstore )