#NEWLINE# def test_body_stream_wrapper( self ) : #NEWLINE# #INDENT# data = testing . rand_string ( ( SIZE_1_KB / 2 ) , SIZE_1_KB ) #NEWLINE# expected_body = data . encode ( 'utf-8' ) #NEWLINE# expected_len = len ( expected_body ) #NEWLINE# expected_lines = [ ( line + '\n' ) . encode ( 'utf-8' ) for line in data . split ( '\n' ) ] #NEWLINE# expected_lines [ ( - 1 ) ] = expected_lines [ ( - 1 ) ] [ : ( - 1 ) ] #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . read ( ) , expected_body ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . read ( 2 ) , expected_body [ 0 : 2 ] ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . read ( ( expected_len + 1 ) ) , expected_body ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# for i in range ( ( expected_len + 1 ) ) : #NEWLINE# #INDENT# expected_value = ( expected_body [ i : ( i + 1 ) ] if ( i < expected_len ) else b'' ) #NEWLINE# self . assertEqual ( body . read ( 1 ) , expected_value ) #NEWLINE# #UNINDENT# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . readline ( ) , expected_lines [ 0 ] ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . readline ( ( - 1 ) ) , expected_lines [ 0 ] ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . readline ( ( expected_len + 1 ) ) , expected_lines [ 0 ] ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . readlines ( ) , expected_lines ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . readlines ( ( - 1 ) ) , expected_lines ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( body . readlines ( ( expected_len + 1 ) ) , expected_lines ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# self . assertEqual ( next ( body ) , expected_lines [ 0 ] ) #NEWLINE# stream = io . BytesIO ( expected_body ) #NEWLINE# body = request_helpers . Body ( stream , expected_len ) #NEWLINE# for ( i , line ) in enumerate ( body ) : #NEWLINE# #INDENT# self . assertEqual ( line , expected_lines [ i ] )