#NEWLINE# @ classmethod #NEWLINE# def _getUrlForTestFile( cls , size = None ) : #NEWLINE# #INDENT# fileName = ( 'testfile_%s' % uuid . uuid4 ( ) ) #NEWLINE# bucket = cls . _externalStore ( ) #NEWLINE# url = ( 's3://%s/%s' % ( bucket . name , fileName ) ) #NEWLINE# if ( size is None ) : #NEWLINE# #INDENT# return url #NEWLINE# #UNINDENT# with open ( '/dev/urandom' , 'r' ) as readable : #NEWLINE# #INDENT# if ( size < cls . mpTestPartSize ) : #NEWLINE# #INDENT# bucket . new_key ( fileName ) . set_contents_from_string ( readable . read ( size ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# mp = bucket . initiate_multipart_upload ( key_name = fileName ) #NEWLINE# start = 0 #NEWLINE# partNum = itertools . count ( ) #NEWLINE# partSize = cls . mpTestPartSize #NEWLINE# try : #NEWLINE# #INDENT# while ( start < size ) : #NEWLINE# #INDENT# end = min ( ( start + partSize ) , size ) #NEWLINE# part = io . BytesIO ( readable . read ( partSize ) ) #NEWLINE# mp . upload_part_from_file ( fp = part , part_num = ( next ( partNum ) + 1 ) , size = ( end - start ) ) #NEWLINE# start = end #NEWLINE# if ( start == size ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# assert ( start == size ) #NEWLINE# #UNINDENT# except : #NEWLINE# #INDENT# mp . cancel_upload ( ) #NEWLINE# raise #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# mp . complete_upload ( ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# return ( url , hashlib . md5 ( bucket . get_key ( fileName ) . get_contents_as_string ( ) ) . hexdigest ( ) )