#NEWLINE# def test_pipe_stream( self ) : #NEWLINE# #INDENT# block_sizes = list ( ( ( 4 ** i ) for i in range ( 0 , 12 ) ) ) #NEWLINE# data_blocks = list ( ( os . urandom ( size ) for size in block_sizes ) ) #NEWLINE# expected = '' . join ( buffer_size ) #NEWLINE# buffer_sizes = [ 100001 , ( 512 * 1024 ) , ( 1024 * 1024 ) ] #NEWLINE# for buffer_size in buffer_sizes : #NEWLINE# #INDENT# ( parent_conn , child_conn ) = multiprocessing . Pipe ( ) #NEWLINE# stream = gcsio . GcsBufferedWriter . PipeStream ( child_conn ) #NEWLINE# child_thread = threading . Thread ( target = self . _read_and_verify , args = ( stream , expected , buffer_size ) ) #NEWLINE# child_thread . start ( ) #NEWLINE# for data in data_blocks : #NEWLINE# #INDENT# parent_conn . send_bytes ( data ) #NEWLINE# #UNINDENT# parent_conn . close ( ) #NEWLINE# child_thread . join ( )