#NEWLINE# def test_failing_task_with_notifications( self ) : #NEWLINE# #INDENT# values = [ ] #NEWLINE# flow = utils . FailingTask ( 'fail' ) #NEWLINE# engine = self . _make_engine ( flow ) #NEWLINE# expected = [ 'fail.f RUNNING' , 'fail.t RUNNING' , 'fail.t FAILURE(Failure: RuntimeError: Woot!)' , 'fail.t REVERTING' , 'fail.t REVERTED(None)' , 'fail.f REVERTED' ] #NEWLINE# with utils . CaptureListener ( engine , values = values ) as capturer : #NEWLINE# #INDENT# self . assertFailuresRegexp ( RuntimeError , '^Woot' , engine . run ) #NEWLINE# #UNINDENT# self . assertEqual ( expected , capturer . values ) #NEWLINE# self . assertEqual ( states . REVERTED , engine . storage . get_flow_state ( ) ) #NEWLINE# with utils . CaptureListener ( engine , values = values ) as capturer : #NEWLINE# #INDENT# self . assertFailuresRegexp ( RuntimeError , '^Woot' , engine . run ) #NEWLINE# #UNINDENT# now_expected = list ( expected ) #NEWLINE# now_expected . extend ( [ 'fail.t PENDING' , 'fail.f PENDING' ] ) #NEWLINE# now_expected . extend ( expected ) #NEWLINE# self . assertEqual ( now_expected , values ) #NEWLINE# self . assertEqual ( states . REVERTED , engine . storage . get_flow_state ( ) )