#NEWLINE# def testMultipartUploads( self ) : #NEWLINE# #INDENT# "\n            This test is meant to cover multi-part uploads in the AWSJobStore but it doesn't hurt\n            running it against the other job stores as well.\n            " #NEWLINE# random_device = '/dev/urandom' #NEWLINE# bufSize = 65536 #NEWLINE# partSize = self . _partSize ( ) #NEWLINE# self . assertEquals ( ( partSize % bufSize ) , 0 ) #NEWLINE# job = self . master . create ( '1' , 2 , 3 , 4 , preemptable = False ) #NEWLINE# for partsPerFile in ( 1 , 2.33 ) : #NEWLINE# #INDENT# checksum = hashlib . md5 ( ) #NEWLINE# checksumQueue = Queue ( 2 ) #NEWLINE# def checksumThreadFn( ) : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# _buf = checksumQueue . get ( ) #NEWLINE# if ( _buf is None ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# checksum . update ( _buf ) #NEWLINE# #UNINDENT# #UNINDENT# checksumThread = Thread ( target = checksumThreadFn ) #NEWLINE# checksumThread . start ( ) #NEWLINE# try : #NEWLINE# #INDENT# with open ( random_device ) as readable : #NEWLINE# #INDENT# with self . master . writeFileStream ( job . jobStoreID ) as ( writable , fileId ) : #NEWLINE# #INDENT# for i in range ( int ( ( ( partSize * partsPerFile ) / bufSize ) ) ) : #NEWLINE# #INDENT# buf = readable . read ( bufSize ) #NEWLINE# checksumQueue . put ( buf ) #NEWLINE# writable . write ( buf ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# finally : #NEWLINE# #INDENT# checksumQueue . put ( None ) #NEWLINE# checksumThread . join ( ) #NEWLINE# #UNINDENT# before = checksum . hexdigest ( ) #NEWLINE# checksum = hashlib . md5 ( ) #NEWLINE# with self . master . readFileStream ( fileId ) as readable : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# buf = readable . read ( bufSize ) #NEWLINE# if ( not buf ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# checksum . update ( buf ) #NEWLINE# #UNINDENT# #UNINDENT# after = checksum . hexdigest ( ) #NEWLINE# self . assertEquals ( before , after ) #NEWLINE# checksum = hashlib . md5 ( ) #NEWLINE# ( fh , path ) = tempfile . mkstemp ( ) #NEWLINE# try : #NEWLINE# #INDENT# with os . fdopen ( fh , 'r+' ) as writable : #NEWLINE# #INDENT# with open ( random_device ) as readable : #NEWLINE# #INDENT# for i in range ( int ( ( ( partSize * partsPerFile ) / bufSize ) ) ) : #NEWLINE# #INDENT# buf = readable . read ( bufSize ) #NEWLINE# writable . write ( buf ) #NEWLINE# checksum . update ( buf ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# fileId = self . master . writeFile ( path , job . jobStoreID ) #NEWLINE# #UNINDENT# finally : #NEWLINE# #INDENT# os . unlink ( path ) #NEWLINE# #UNINDENT# before = checksum . hexdigest ( ) #NEWLINE# checksum = hashlib . md5 ( ) #NEWLINE# with self . master . readFileStream ( fileId ) as readable : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# buf = readable . read ( bufSize ) #NEWLINE# if ( not buf ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# checksum . update ( buf ) #NEWLINE# #UNINDENT# #UNINDENT# after = checksum . hexdigest ( ) #NEWLINE# self . assertEquals ( before , after ) #NEWLINE# #UNINDENT# self . master . delete ( job . jobStoreID )