#NEWLINE# def test_read( ) : #NEWLINE# #INDENT# p = sp . Popen ( args = [ 'python' , '-' ] , stdin = sp . PIPE , stdout = sp . PIPE ) #NEWLINE# p . stdin . write ( script ) #NEWLINE# p . stdin . close ( ) #NEWLINE# f = stream . Reader ( p . stdout ) #NEWLINE# result = list ( zip ( range ( 10 ) , f ) ) #NEWLINE# p . kill ( ) #NEWLINE# j = 0 #NEWLINE# for ( i , buf ) in result : #NEWLINE# #INDENT# assert ( i == j ) #NEWLINE# assert ( len ( buf ) == f . bufsize ) #NEWLINE# j += 1 #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# next ( f ) #NEWLINE# #UNINDENT# except IOError as e : #NEWLINE# #INDENT# assert ( e . args == ( 'timeout' , ) )