#NEWLINE# @ defer . inlineCallbacks #NEWLINE# def test_using_processor_with_a_failure( self ) : #NEWLINE# #INDENT# mocked_log = self . mocker . replace ( log ) #NEWLINE# mocked_log . log_traceback ( mocker . ARGS , mocker . KWARGS ) #NEWLINE# self . mocker . count ( 1 ) #NEWLINE# self . mocker . replay ( ) #NEWLINE# processed_ok = list ( ) #NEWLINE# processed_all = list ( ) #NEWLINE# def consider_adding_to_processed( baton ) : #NEWLINE# #INDENT# processed_all . append ( baton ) #NEWLINE# if ( len ( processed_all ) == 2 ) : #NEWLINE# #INDENT# raise Exception ( 'forcibly raised' ) #NEWLINE# #UNINDENT# processed_ok . append ( baton ) #NEWLINE# #UNINDENT# registered = list ( ) #NEWLINE# stub_poller = StubSocketPoller ( ( lambda feeder , socket : registered . append ( feeder ) ) ) #NEWLINE# class Provider ( ) : #NEWLINE# #INDENT# def add_consumer( self , resource_dependency ) : #NEWLINE# #INDENT# resource_dependency . on_resource_ready ( consider_adding_to_processed ) #NEWLINE# #UNINDENT# #UNINDENT# provider = Provider ( ) #NEWLINE# self . resource_manager . register ( 'pipeline.pipeline_name' , provider ) #NEWLINE# self . resource_manager . register ( 'zmq.socket.socket_name' , provider ) #NEWLINE# feeder = providers . ZMQProcessorFeeder ( stub_poller , processor_config = 'pipeline.pipeline_name' , socket_name = 'socket_name' ) #NEWLINE# feeder . configure ( self . runtime_environment ) #NEWLINE# self . dependency_manager . resolve_initial_states ( ) #NEWLINE# self . assertIn ( feeder , registered ) #NEWLINE# del registered [ : ] #NEWLINE# expected_processed = [ 'foo' , 'baz' ] #NEWLINE# message_batch = [ 'foo' , 'bar' , 'baz' ] #NEWLINE# with mock . patch . object ( providers , 'logger' ) as mocked_logger : #NEWLINE# #INDENT# ( yield feeder . handle_messages ( message_batch ) ) #NEWLINE# #UNINDENT# self . assertEquals ( processed_ok , expected_processed ) #NEWLINE# self . assertIn ( feeder , registered )