#NEWLINE# @ contextmanager #NEWLINE# def mock_worker( **imap_spec ) : #NEWLINE# #INDENT# from mock import Mock , patch #NEWLINE# from monocle.callback import defer #NEWLINE# from tinymail.imap_worker import ImapWorker #NEWLINE# worker = Mock ( spec = ImapWorker ) #NEWLINE# worker . done = Mock ( ) #NEWLINE# state = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# folders = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# for ( name , mbox_spec ) in imap_spec . iteritems ( ) : #NEWLINE# #INDENT# folder = folders [ name ] = { #NEWLINE# #INDENT# 'flags' : { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , #NEWLINE# 'headers' : { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , #NEWLINE# 'index' : { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } #NEWLINE# folder [ 'uidvalidity' ] = mbox_spec . pop ( 'UIDVALIDITY' , 123456 ) #NEWLINE# for ( i , ( uid , msg_spec ) ) in enumerate ( mbox_spec . iteritems ( ) ) : #NEWLINE# #INDENT# if ( msg_spec is None ) : #NEWLINE# #INDENT# msg_spec = ( uid , set ( ) , 'PLACEHOLDER HEADER' ) #NEWLINE# #UNINDENT# folder [ 'flags' ] [ uid ] = msg_spec [ 1 ] #NEWLINE# folder [ 'index' ] [ uid ] = i #NEWLINE# folder [ 'headers' ] [ i ] = msg_spec [ 2 ] #NEWLINE# #UNINDENT# #UNINDENT# worker . connect . return_value = defer ( None ) #NEWLINE# worker . get_mailbox_names . return_value = defer ( list ( folders ) ) #NEWLINE# def select_mailbox( name , readonly = True ) : #NEWLINE# #INDENT# state [ 'name' ] = name #NEWLINE# return defer ( { #NEWLINE# #INDENT# 'MESSAGES' : len ( folders [ name ] [ 'flags' ] ) , #NEWLINE# 'UIDVALIDITY' : folders [ name ] [ 'uidvalidity' ] , #NEWLINE# #UNINDENT# } ) #NEWLINE# #UNINDENT# worker . select_mailbox = select_mailbox #NEWLINE# worker . get_message_flags = ( lambda : defer ( folders [ state [ 'name' ] ] [ 'flags' ] ) ) #NEWLINE# def get_message_headers( uid_list ) : #NEWLINE# #INDENT# name = state [ 'name' ] #NEWLINE# message_headers = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# for uid in uid_list : #NEWLINE# #INDENT# i = folders [ name ] [ 'index' ] [ uid ] #NEWLINE# message_headers [ uid ] = folders [ name ] [ 'headers' ] [ i ] #NEWLINE# #UNINDENT# return defer ( message_headers ) #NEWLINE# #UNINDENT# worker . get_message_headers . side_effect = get_message_headers #NEWLINE# def copy_messages( src_uids , dst_folder_name ) : #NEWLINE# #INDENT# src_folder = folders [ state [ 'name' ] ] #NEWLINE# dst_folder = folders [ dst_folder_name ] #NEWLINE# next_uid = ( max ( ( [ 0 ] + dst_folder [ 'index' ] . keys ( ) ) ) + 1 ) #NEWLINE# next_index = ( max ( ( [ 0 ] + dst_folder [ 'index' ] . values ( ) ) ) + 1 ) #NEWLINE# uid_map = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# for uid in src_uids : #NEWLINE# #INDENT# uid_map [ uid ] = next_uid #NEWLINE# i = src_folder [ 'index' ] [ uid ] #NEWLINE# dst_folder [ 'flags' ] [ next_uid ] = src_folder [ 'flags' ] [ uid ] #NEWLINE# dst_folder [ 'index' ] [ next_uid ] = next_index #NEWLINE# dst_folder [ 'headers' ] [ next_index ] = src_folder [ 'headers' ] [ i ] #NEWLINE# next_uid += 1 #NEWLINE# next_index += 1 #NEWLINE# #UNINDENT# return defer ( { #NEWLINE# #INDENT# 'UIDVALIDITY' : dst_folder [ 'uidvalidity' ] , #NEWLINE# 'uid_map' : uid_map , #NEWLINE# #UNINDENT# } ) #NEWLINE# #UNINDENT# worker . copy_messages . side_effect = copy_messages #NEWLINE# worker . change_flag . return_value = defer ( None ) #NEWLINE# worker . close_mailbox . return_value = defer ( None ) #NEWLINE# worker . disconnect . return_value = defer ( None ) #NEWLINE# worker . copy_messages . return_value = defer ( None ) #NEWLINE# with patch ( 'tinymail.account._new_imap_worker' , Mock ( return_value = worker ) ) : #NEWLINE# #INDENT# ( yield worker )