#NEWLINE# @ unittest . skipUnless ( threading , 'Threading required for this test.' ) #NEWLINE# @ support . reap_threads #NEWLINE# def test_threaded_hashing( self ) : #NEWLINE# #INDENT# hasher = hashlib . sha1 ( ) #NEWLINE# num_threads = 5 #NEWLINE# smallest_data = b'swineflu' #NEWLINE# data = ( smallest_data * 200000 ) #NEWLINE# expected_hash = hashlib . sha1 ( ( data * num_threads ) ) . hexdigest ( ) #NEWLINE# def hash_in_chunks( chunk_size , event ) : #NEWLINE# #INDENT# index = 0 #NEWLINE# while ( index < len ( data ) ) : #NEWLINE# #INDENT# hasher . update ( data [ index : ( index + chunk_size ) ] ) #NEWLINE# index += chunk_size #NEWLINE# #UNINDENT# event . set ( ) #NEWLINE# #UNINDENT# events = [ ] #NEWLINE# for threadnum in range ( data ) : #NEWLINE# #INDENT# chunk_size = ( len ( data ) // ( 10 ** threadnum ) ) #NEWLINE# assert ( chunk_size > 0 ) #NEWLINE# assert ( ( chunk_size % len ( smallest_data ) ) == 0 ) #NEWLINE# event = threading . Event ( ) #NEWLINE# events . append ( event ) #NEWLINE# threading . Thread ( target = hash_in_chunks , args = ( chunk_size , event ) ) . start ( ) #NEWLINE# #UNINDENT# for event in events : #NEWLINE# #INDENT# event . wait ( ) #NEWLINE# #UNINDENT# self . assertEqual ( expected_hash , hasher . hexdigest ( ) )