<s> """<STR_LIT>""" <EOL> import Queue <EOL> import logging <EOL> import sys <EOL> import time <EOL> import unittest <EOL> import gflags <EOL> FLAGS = gflags . FLAGS <EOL> from dpxdt . client import workers <EOL> from dpxdt . client import fetch_worker <EOL> from dpxdt . client import timer_worker <EOL> import test_utils <EOL> class EchoThread ( workers . WorkerThread ) : <EOL> def handle_item ( self , item ) : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> if item . should_die : <EOL> raise Exception ( '<STR_LIT>' % item . input_number ) <EOL> item . output_number = item . input_number <EOL> return item <EOL> class EchoItem ( workers . WorkItem ) : <EOL> def __init__ ( self , number , should_die = False ) : <EOL> workers . WorkItem . __init__ ( self ) <EOL> self . input_number = number <EOL> self . output_number = None <EOL> self . should_die = should_die <EOL> class EchoChild ( workers . WorkflowItem ) : <EOL> def run ( self , number , should_die = False , wait_seconds = <NUM_LIT:0> ) : <EOL> if wait_seconds > <NUM_LIT:0> : <EOL> yield timer_worker . TimerItem ( wait_seconds ) <EOL> item = yield EchoItem ( number , should_die = should_die ) <EOL> raise workers . Return ( item . output_number ) <EOL> class EchoChildWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self , number , should_die = False , wait_seconds = <NUM_LIT:0> ) : <EOL> if wait_seconds > <NUM_LIT:0> : <EOL> yield timer_worker . TimerItem ( wait_seconds ) <EOL> if should_die : <EOL> try : <EOL> yield EchoChild ( number , should_die = should_die ) <EOL> except Exception , e : <EOL> raise e <EOL> raise workers . Return ( number ) <EOL> class RootWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self , child_count , die_on = - <NUM_LIT:1> ) : <EOL> total = <NUM_LIT:0> <EOL> for i in xrange ( child_count ) : <EOL> number = yield EchoChild ( i , should_die = ( die_on == i ) ) <EOL> assert number is i <EOL> total += number <EOL> self . result = total <EOL> class GeneratorExceptionChild ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> number = yield EchoChild ( <NUM_LIT:4> , should_die = False ) <EOL> raise Exception ( '<STR_LIT>' % number ) <EOL> class GeneratorExceptionReraiseParent ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> try : <EOL> yield GeneratorExceptionChild ( ) <EOL> except Exception , e : <EOL> assert str ( e ) == '<STR_LIT>' , str ( e ) <EOL> raise Exception ( '<STR_LIT>' ) <EOL> class RootGeneratorExceptionWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> try : <EOL> yield GeneratorExceptionReraiseParent ( ) <EOL> except Exception , e : <EOL> assert str ( e ) == '<STR_LIT>' , str ( e ) <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> else : <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> class RootWaitAllWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self , child_count ) : <EOL> wait_all = [ EchoItem ( i ) for i in xrange ( child_count ) ] <EOL> output = yield wait_all <EOL> raise workers . Return ( sum ( x . output_number for x in output ) ) <EOL> class RootWaitAnyWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> output = yield workers . WaitAny ( [ <EOL> EchoItem ( <NUM_LIT:10> ) , <EOL> EchoChild ( <NUM_LIT> ) , <EOL> EchoItem ( <NUM_LIT:2> ) , <EOL> EchoItem ( <NUM_LIT> ) , <EOL> ] ) <EOL> assert len ( [ x for x in output if x . done ] ) >= <NUM_LIT:1> <EOL> assert not output [ <NUM_LIT:1> ] . done <EOL> yield timer_worker . TimerItem ( <NUM_LIT:2> ) <EOL> results = yield output <EOL> assert len ( [ x for x in output if x . done ] ) >= <NUM_LIT:1> <EOL> assert results [ <NUM_LIT:0> ] . done and results [ <NUM_LIT:0> ] . output_number == <NUM_LIT:10> <EOL> assert results [ <NUM_LIT:1> ] == <NUM_LIT> <EOL> assert results [ <NUM_LIT:2> ] . done and results [ <NUM_LIT:2> ] . output_number == <NUM_LIT:2> <EOL> assert results [ <NUM_LIT:3> ] . done and results [ <NUM_LIT:3> ] . output_number == <NUM_LIT> <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> class RootWaitAnyExceptionWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> output = yield workers . WaitAny ( [ <EOL> EchoChild ( <NUM_LIT> , should_die = True ) , <EOL> EchoItem ( <NUM_LIT:10> ) , <EOL> EchoItem ( <NUM_LIT> ) , <EOL> ] ) <EOL> assert len ( [ x for x in output if x . done ] ) == <NUM_LIT:1> <EOL> assert not output [ <NUM_LIT:0> ] . done <EOL> assert output [ <NUM_LIT:1> ] . done and output [ <NUM_LIT:1> ] . output_number == <NUM_LIT:10> <EOL> assert not output [ <NUM_LIT:2> ] . done <EOL> yield timer_worker . TimerItem ( <NUM_LIT:2> ) <EOL> assert output [ <NUM_LIT:0> ] . done <EOL> assert output [ <NUM_LIT:2> ] . done and output [ <NUM_LIT:2> ] . output_number == <NUM_LIT> <EOL> try : <EOL> yield output <EOL> except Exception , e : <EOL> raise workers . Return ( str ( e ) ) <EOL> else : <EOL> assert False , '<STR_LIT>' <EOL> class FireAndForgetEchoItem ( EchoItem ) : <EOL> fire_and_forget = True <EOL> class RootFireAndForgetWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> job1 = FireAndForgetEchoItem ( <NUM_LIT:10> ) <EOL> result = yield job1 <EOL> print result <EOL> assert result is job1 <EOL> assert not result . done <EOL> result = yield EchoItem ( <NUM_LIT> ) <EOL> assert result . done <EOL> assert result . output_number == <NUM_LIT> <EOL> job2 = EchoItem ( <NUM_LIT:30> ) <EOL> job2 . fire_and_forget = True <EOL> result = yield job2 <EOL> assert result is job2 <EOL> assert not result . done <EOL> job3 = FireAndForgetEchoItem ( <NUM_LIT> ) <EOL> job3 . fire_and_forget = False <EOL> result = yield job3 <EOL> assert result is job3 <EOL> assert result . done <EOL> assert result . output_number == <NUM_LIT> <EOL> job4 = EchoChild ( <NUM_LIT> ) <EOL> job4 . fire_and_forget = True <EOL> result = yield job4 <EOL> assert result is job4 <EOL> assert not result . done <EOL> yield timer_worker . TimerItem ( <NUM_LIT:2> ) <EOL> assert job1 . done <EOL> assert job1 . output_number == <NUM_LIT:10> <EOL> assert job2 . done <EOL> assert job2 . output_number == <NUM_LIT:30> <EOL> assert job4 . done <EOL> assert job4 . result == <NUM_LIT> <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> class RootFireAndForgetExceptionWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> job = EchoChild ( <NUM_LIT> , should_die = True ) <EOL> job . fire_and_forget = True <EOL> result = yield job <EOL> assert result is job <EOL> assert not result . done <EOL> assert not result . error <EOL> result = yield EchoItem ( <NUM_LIT> ) <EOL> assert result . done <EOL> assert result . output_number == <NUM_LIT> <EOL> yield timer_worker . TimerItem ( <NUM_LIT:2> ) <EOL> assert job . done <EOL> assert str ( job . error [ <NUM_LIT:1> ] ) == '<STR_LIT>' <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> class RootFireAndForgetMultipleExceptionWorkflow ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> jobs = [ ] <EOL> for i in xrange ( <NUM_LIT:3> ) : <EOL> job = EchoChildWorkflow ( <NUM_LIT> , should_die = True , wait_seconds = i * <NUM_LIT:0.5> ) <EOL> job . fire_and_forget = True <EOL> result = yield job <EOL> assert result is job <EOL> assert not result . done <EOL> assert not result . error <EOL> jobs . append ( job ) <EOL> yield timer_worker . TimerItem ( <NUM_LIT:0.5> ) <EOL> assert jobs [ <NUM_LIT:0> ] . done <EOL> assert jobs [ <NUM_LIT:1> ] . done is False <EOL> assert jobs [ <NUM_LIT:2> ] . done is False <EOL> yield timer_worker . TimerItem ( <NUM_LIT> ) <EOL> assert jobs [ <NUM_LIT:0> ] . done <EOL> assert jobs [ <NUM_LIT:1> ] . done <EOL> assert jobs [ <NUM_LIT:2> ] . done <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> class RootWaitAnyFireAndForget ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> output = yield workers . WaitAny ( [ <EOL> FireAndForgetEchoItem ( <NUM_LIT> ) , <EOL> EchoItem ( <NUM_LIT> ) , <EOL> EchoChild ( <NUM_LIT> ) , <EOL> ] ) <EOL> assert output [ <NUM_LIT:0> ] . done <EOL> assert output [ <NUM_LIT:1> ] . done <EOL> assert not output [ <NUM_LIT:2> ] . done <EOL> result = yield output <EOL> assert result [ <NUM_LIT:2> ] == <NUM_LIT> <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> class RootWaitAllFireAndForget ( workers . WorkflowItem ) : <EOL> def run ( self ) : <EOL> output = yield [ <EOL> FireAndForgetEchoItem ( <NUM_LIT> ) , <EOL> EchoItem ( <NUM_LIT> ) , <EOL> EchoChild ( <NUM_LIT> ) , <EOL> ] <EOL> assert output [ <NUM_LIT:0> ] . done <EOL> assert output [ <NUM_LIT:1> ] . done <EOL> assert output [ <NUM_LIT:2> ] == <NUM_LIT> <EOL> raise workers . Return ( '<STR_LIT>' ) <EOL> class WorkflowThreadTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> FLAGS . fetch_frequency = <NUM_LIT:100> <EOL> FLAGS . polltime = <NUM_LIT> <EOL> self . coordinator = workers . get_coordinator ( ) <EOL> self . echo_queue = Queue . Queue ( ) <EOL> self . coordinator . register ( EchoItem , self . echo_queue ) <EOL> self . coordinator . register ( FireAndForgetEchoItem , self . echo_queue ) <EOL> self . coordinator . worker_threads . append ( <EOL> EchoThread ( self . echo_queue , self . coordinator . input_queue ) ) <EOL> self . timer_queue = Queue . Queue ( ) <EOL> self . coordinator . register ( timer_worker . TimerItem , self . timer_queue ) <EOL> self . coordinator . worker_threads . append ( <EOL> timer_worker . TimerThread ( <EOL> self . timer_queue , self . coordinator . input_queue ) ) <EOL> self . coordinator . start ( ) <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> self . coordinator . stop ( ) <EOL> self . coordinator . join ( ) <EOL> self . assertEquals ( <NUM_LIT:0> , len ( self . coordinator . pending ) ) <EOL> def testMultiLevelWorkflow ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootWorkflow ( <NUM_LIT:5> ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( <NUM_LIT:4> + <NUM_LIT:3> + <NUM_LIT:2> + <NUM_LIT:1> + <NUM_LIT:0> , work . result ) <EOL> def testMultiLevelWorkflowException ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootWorkflow ( <NUM_LIT:5> , die_on = <NUM_LIT:3> ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> try : <EOL> finished . check_result ( ) <EOL> except Exception , e : <EOL> self . assertEquals ( '<STR_LIT>' , str ( e ) ) <EOL> def testWorkflowExceptionPropagation ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootGeneratorExceptionWorkflow ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def testWaitAll ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootWaitAllWorkflow ( <NUM_LIT:4> ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( <NUM_LIT:6> , work . result ) <EOL> def testWaitAny ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootWaitAnyWorkflow ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def testWaitAnyException ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootWaitAnyExceptionWorkflow ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def testFireAndForget ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootFireAndForgetWorkflow ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def testFireAndForgetException ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootFireAndForgetExceptionWorkflow ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def testFireAndForgetException_MultiLevel ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootFireAndForgetMultipleExceptionWorkflow ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def testWaitAnyFireAndForget ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootWaitAnyFireAndForget ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def testWaitAllFireAndForget ( self ) : <EOL> """<STR_LIT>""" <EOL> work = RootWaitAllFireAndForget ( ) <EOL> work . root = True <EOL> self . coordinator . input_queue . put ( work ) <EOL> finished = self . coordinator . output_queue . get ( ) <EOL> self . assertTrue ( work is finished ) <EOL> finished . check_result ( ) <EOL> self . assertEquals ( '<STR_LIT>' , work . result ) <EOL> def main ( argv ) : <EOL> test_utils . debug_log_everything ( ) <EOL> argv = FLAGS ( argv ) <EOL> unittest . main ( argv = argv ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( sys . argv ) </s>
