#NEWLINE# def make_t( wf_name , wf ) : #NEWLINE# #INDENT# def test( self ) : #NEWLINE# #INDENT# lw = LocalWorkflow ( wf , activity_workers = 16 , workflow_workers = 2 , executor = ThreadPoolExecutor ) #NEWLINE# lw . conf_activity ( 'a' , examples . activity ) #NEWLINE# start = time . time ( ) #NEWLINE# result = lw . run ( TIME_SCALE ) #NEWLINE# duration = ( time . time ( ) - start ) #NEWLINE# lines = [ l . strip ( ) for l in wf . __doc__ . split ( '\n' ) ] #NEWLINE# expected = None #NEWLINE# for line in lines : #NEWLINE# #INDENT# if line . startswith ( 'R ' ) : #NEWLINE# #INDENT# expected = int ( line . split ( ) [ ( - 1 ) ] . split ( '-' ) [ ( - 1 ) ] ) #NEWLINE# break #NEWLINE# #UNINDENT# #UNINDENT# self . assertEquals ( expected , result ) #NEWLINE# for line in lines : #NEWLINE# #INDENT# if line . startswith ( 'Duration:' ) : #NEWLINE# #INDENT# expected_duration = ( ( int ( line . split ( ) [ ( - 1 ) ] ) * TIME_SCALE ) * 0.1 ) #NEWLINE# break #NEWLINE# #UNINDENT# #UNINDENT# print ( expected_duration , duration ) #NEWLINE# self . assertTrue ( ( abs ( ( expected_duration - duration ) ) < ( TIME_SCALE * 0.9 ) ) ) #NEWLINE# #UNINDENT# return test