#NEWLINE# def test_graph_flow_conditional_history( self ) : #NEWLINE# #INDENT# def even_odd_decider( history , allowed ) : #NEWLINE# #INDENT# total = sum ( six . itervalues ( history ) ) #NEWLINE# if ( total == allowed ) : #NEWLINE# #INDENT# return True #NEWLINE# #UNINDENT# return False #NEWLINE# #UNINDENT# flow = gf . Flow ( 'root' ) #NEWLINE# task1 = utils . TaskMultiArgOneReturn ( name = 'task1' ) #NEWLINE# task2 = utils . ProgressingTask ( name = 'task2' ) #NEWLINE# task2_2 = utils . ProgressingTask ( name = 'task2_2' ) #NEWLINE# task3 = utils . ProgressingTask ( name = 'task3' ) #NEWLINE# task3_3 = utils . ProgressingTask ( name = 'task3_3' ) #NEWLINE# flow . add ( task1 , task2 , task2_2 , task3 , task3_3 ) #NEWLINE# flow . link ( task1 , task2 , decider = functools . partial ( even_odd_decider , allowed = 2 ) ) #NEWLINE# flow . link ( task2 , task2_2 ) #NEWLINE# flow . link ( task1 , task3 , decider = functools . partial ( even_odd_decider , allowed = 1 ) ) #NEWLINE# flow . link ( task3 , task3_3 ) #NEWLINE# engine = self . _make_engine ( flow ) #NEWLINE# engine . storage . inject ( { #NEWLINE# #INDENT# 'x' : 0 , #NEWLINE# 'y' : 1 , #NEWLINE# 'z' : 1 , #NEWLINE# #UNINDENT# } ) #NEWLINE# with utils . CaptureListener ( engine , capture_flow = False ) as capturer : #NEWLINE# #INDENT# engine . run ( ) #NEWLINE# #UNINDENT# expected = set ( [ 'task1.t RUNNING' , 'task1.t SUCCESS(2)' , 'task3.t IGNORE' , 'task3_3.t IGNORE' , 'task2.t RUNNING' , 'task2.t SUCCESS(5)' , 'task2_2.t RUNNING' , 'task2_2.t SUCCESS(5)' ] ) #NEWLINE# self . assertEqual ( expected , set ( capturer . values ) ) #NEWLINE# engine = self . _make_engine ( task2_2 ) #NEWLINE# engine . storage . inject ( { #NEWLINE# #INDENT# 'x' : 0 , #NEWLINE# 'y' : 0 , #NEWLINE# 'z' : 1 , #NEWLINE# #UNINDENT# } ) #NEWLINE# with utils . CaptureListener ( engine , capture_flow = False ) as capturer : #NEWLINE# #INDENT# engine . run ( ) #NEWLINE# #UNINDENT# expected = set ( [ 'task1.t RUNNING' , 'task1.t SUCCESS(1)' , 'task2.t IGNORE' , 'task2_2.t IGNORE' , 'task3.t RUNNING' , 'task3.t SUCCESS(5)' , 'task3_3.t RUNNING' , 'task3_3.t SUCCESS(5)' ] ) #NEWLINE# self . assertEqual ( expected , set ( capturer . values ) )