#NEWLINE# def test_listen_direct_exit( self , stream ) : #NEWLINE# #INDENT# io_in = io . StringIO ( ) #NEWLINE# io_out = io . StringIO ( ) #NEWLINE# io_err = io . StringIO ( ) #NEWLINE# stream_thread = Thread ( target = stream . listen ( io_in , io_out , io_err ) ) #NEWLINE# stream_thread . start ( ) #NEWLINE# io_in . write ( 'EXIT\n' ) #NEWLINE# stream_thread . join ( ) #NEWLINE# io_out . seek ( 0 ) #NEWLINE# io_err . seek ( 0 ) #NEWLINE# assert ( len ( io_out . read ( ) ) == 0 ) #NEWLINE# assert ( len ( io_err . read ( ) ) == 0 ) #NEWLINE# assert ( stream . predict_service . predict . call_count == 0 )