#NEWLINE# def test_predict_error( self , stream ) : #NEWLINE# #INDENT# from palladium.interfaces import PredictError #NEWLINE# io_in = io . StringIO ( ) #NEWLINE# io_out = io . StringIO ( ) #NEWLINE# io_err = io . StringIO ( ) #NEWLINE# line = '[{"hey": "1"}]\n' #NEWLINE# io_in . write ( stream ) #NEWLINE# io_in . write ( 'EXIT\n' ) #NEWLINE# io_in . seek ( 0 ) #NEWLINE# stream . predict_service . predict . side_effect = PredictError ( 'error' ) #NEWLINE# stream_thread = Thread ( target = stream . listen ( io_in , io_out , io_err ) ) #NEWLINE# stream_thread . start ( ) #NEWLINE# stream_thread . join ( ) #NEWLINE# io_out . seek ( 0 ) #NEWLINE# io_err . seek ( 0 ) #NEWLINE# assert ( io_out . read ( ) == '[]\n' ) #NEWLINE# assert ( io_err . read ( ) == "Error while processing input row: {}<class 'palladium.interfaces.PredictError'>: error (-1)\n" . format ( line ) ) #NEWLINE# assert ( stream . predict_service . predict . call_count == 1 )