#NEWLINE# def test_import_task_created_with_plugin( self ) : #NEWLINE# #INDENT# class ToSingletonPlugin ( plugins . BeetsPlugin ) : #NEWLINE# #INDENT# def __init__( self ) : #NEWLINE# #INDENT# super ( ToSingletonPlugin , self ) . __init__ ( ) #NEWLINE# self . register_listener ( 'import_task_created' , self . import_task_created_event ) #NEWLINE# #UNINDENT# def import_task_created_event( self , session , task ) : #NEWLINE# #INDENT# if ( isinstance ( task , SingletonImportTask ) or isinstance ( task , SentinelImportTask ) or isinstance ( task , ArchiveImportTask ) ) : #NEWLINE# #INDENT# return task #NEWLINE# #UNINDENT# new_tasks = [ ] #NEWLINE# for item in task . items : #NEWLINE# #INDENT# new_tasks . append ( SingletonImportTask ( task . toppath , item ) ) #NEWLINE# #UNINDENT# return new_tasks #NEWLINE# #UNINDENT# #UNINDENT# to_singleton_plugin = ToSingletonPlugin #NEWLINE# self . register_plugin ( to_singleton_plugin ) #NEWLINE# import_files = [ self . import_dir ] #NEWLINE# self . _setup_import_session ( singletons = False ) #NEWLINE# self . importer . paths = import_files #NEWLINE# with helper . capture_log ( ) as logs : #NEWLINE# #INDENT# self . importer . run ( ) #NEWLINE# #UNINDENT# self . unload_plugins ( ) #NEWLINE# self . assertEqual ( logs . count ( 'Sending event: import_task_created' ) , 1 ) #NEWLINE# logs = [ line for line in logs if ( not line . startswith ( 'Sending event:' ) ) ] #NEWLINE# self . assertEqual ( logs , [ 'Singleton: {0}' . format ( self . file_paths [ 0 ] ) , 'Singleton: {0}' . format ( self . file_paths [ 1 ] ) ] )