#NEWLINE# def test_upload_url_concurrency( ) : #NEWLINE# #INDENT# file_name = '/tmp/test_upload_conncurrency.db' #NEWLINE# try : #NEWLINE# #INDENT# os . unlink ( file_name ) #NEWLINE# #UNINDENT# except OSError : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# account_info = SqliteAccountInfo ( file_name ) #NEWLINE# available_urls = set ( ) #NEWLINE# for i in six . moves . range ( 3000 ) : #NEWLINE# #INDENT# url = ( 'url_%d' % i ) #NEWLINE# account_info . put_bucket_upload_url ( 'bucket-id' , url , ( 'auth-token-%d' % i ) ) #NEWLINE# available_urls . add ( url ) #NEWLINE# #UNINDENT# lock = threading . Lock ( ) #NEWLINE# def run_thread( ) : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# ( url , _ ) = account_info . take_bucket_upload_url ( 'bucket-id' ) #NEWLINE# if ( url is None ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# with lock : #NEWLINE# #INDENT# if ( url in available_urls ) : #NEWLINE# #INDENT# available_urls . remove ( url ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# print ( 'DOUBLE:' , url ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# threads = [ ] #NEWLINE# for i in six . moves . range ( 5 ) : #NEWLINE# #INDENT# thread = threading . Thread ( target = run_thread ) #NEWLINE# thread . start ( ) #NEWLINE# threads . append ( thread ) #NEWLINE# #UNINDENT# for t in threads : #NEWLINE# #INDENT# t . join ( ) #NEWLINE# #UNINDENT# if ( len ( available_urls ) != 0 ) : #NEWLINE# #INDENT# print ( 'LEAK:' , available_urls ) #NEWLINE# #UNINDENT# os . unlink ( file_name )