<s> from __future__ import print_function <EOL> import os <EOL> import sys <EOL> import multiprocessing <EOL> import time <EOL> import json <EOL> import uuid <EOL> import random <EOL> import functools <EOL> import gzip <EOL> import vcr <EOL> import vcr . cassette <EOL> import vcr . errors <EOL> import vcr . serialize <EOL> import vcr . request <EOL> from boto . swf . layer1 import Layer1 <EOL> from flowy import restart <EOL> from flowy import wait <EOL> from flowy import TaskError <EOL> from flowy import SWFActivityConfig <EOL> from flowy import SWFActivityWorker <EOL> from flowy import SWFWorkflowConfig <EOL> from flowy import SWFWorkflowStarter <EOL> from flowy import SWFWorkflowWorker <EOL> VERSION = <NUM_LIT:2> <EOL> HERE = os . path . dirname ( os . path . realpath ( __file__ ) ) <EOL> A_CASSETTE = os . path . join ( HERE , '<STR_LIT>' ) <EOL> W_CASSETTE = os . path . join ( HERE , '<STR_LIT>' ) <EOL> DOMAIN = '<STR_LIT>' <EOL> TASKLIST = '<STR_LIT>' <EOL> IDENTITY = '<STR_LIT:test>' <EOL> RECORDING = False <EOL> exit_event = multiprocessing . Event ( ) <EOL> wf_finished_event = multiprocessing . Event ( ) <EOL> def load_cassette ( cassette_path , serializer ) : <EOL> f = gzip . open ( cassette_path , '<STR_LIT:rb>' ) <EOL> cassette_content = f . read ( ) <EOL> cassette = vcr . serialize . deserialize ( cassette_content , serializer ) <EOL> f . close ( ) <EOL> return cassette <EOL> def save_cassette ( cassette_path , cassette_dict , serializer ) : <EOL> data = vcr . serialize . serialize ( cassette_dict , serializer ) <EOL> dirname , _ = os . path . split ( cassette_path ) <EOL> if dirname and not os . path . exists ( dirname ) : <EOL> os . makedirs ( dirname ) <EOL> f = gzip . open ( cassette_path , '<STR_LIT:wb>' ) <EOL> f . write ( data ) <EOL> f . close ( ) <EOL> vcr . cassette . load_cassette = load_cassette <EOL> vcr . cassette . save_cassette = save_cassette <EOL> def requests_match ( r1 , r2 , matchers ) : <EOL> """<STR_LIT>""" <EOL> return all ( m ( r1 , r2 ) for m in matchers ) <EOL> vcr . cassette . requests_match = requests_match <EOL> try : <EOL> from functools import lru_cache <EOL> from urllib . parse import urlparse <EOL> vcr . request . urlparse = lru_cache ( maxsize = None ) ( urlparse ) <EOL> except ImportError : <EOL> pass <EOL> def fake_uuid4 ( ) : <EOL> x = <NUM_LIT:0> <EOL> while <NUM_LIT:1> : <EOL> yield '<STR_LIT>' % x <EOL> x += <NUM_LIT:1> <EOL> uuid . uuid4 = functools . partial ( next , fake_uuid4 ( ) ) <EOL> def break_loop ( self ) : <EOL> return exit_event . is_set ( ) <EOL> class TestSWFWorkflowWorker ( SWFWorkflowWorker ) : <EOL> break_loop = break_loop <EOL> class TestSWFActivityWorker ( SWFActivityWorker ) : <EOL> break_loop = break_loop <EOL> class BaseWorkflow ( object ) : <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> r = self . call ( * args , ** kwargs ) <EOL> wait ( r ) <EOL> wf_finished_event . set ( ) <EOL> return r <EOL> def call ( self , * args , ** kwargs ) : <EOL> raise NotImplementedError <EOL> a_conf = SWFActivityConfig ( default_task_list = TASKLIST , <EOL> default_schedule_to_start = <NUM_LIT:30> , <EOL> default_schedule_to_close = <NUM_LIT> , <EOL> default_start_to_close = <NUM_LIT:15> , <EOL> default_heartbeat = <NUM_LIT:10> ) <EOL> @ a_conf ( version = VERSION ) <EOL> def tactivity ( hb , a = None , b = None , sleep = None , heartbeat = False , err = None ) : <EOL> result = None <EOL> if a is not None and b is not None : <EOL> result = a + b <EOL> elif a is not None : <EOL> result = a * a <EOL> if sleep is not None and RECORDING : <EOL> time . sleep ( sleep ) <EOL> if heartbeat : <EOL> hb ( ) <EOL> if err is not None : <EOL> raise RuntimeError ( err ) <EOL> return result <EOL> empty_conf = SWFWorkflowConfig ( default_task_list = TASKLIST , <EOL> default_decision_duration = <NUM_LIT:10> , <EOL> default_workflow_duration = <NUM_LIT:20> , <EOL> default_child_policy = '<STR_LIT>' , ) <EOL> empty_conf . conf_activity ( '<STR_LIT>' , VERSION , '<STR_LIT>' ) <EOL> @ empty_conf ( version = VERSION ) <EOL> class TWorkflow ( object ) : <EOL> def __init__ ( self , activity ) : <EOL> pass <EOL> def __call__ ( self , a = None , b = None , sleep = None , heartbeat = False , err = None ) : <EOL> dummy_heartbeat = lambda : True <EOL> return tactivity ( dummy_heartbeat , a , b , sleep , heartbeat , err ) <EOL> conf_use_activities = SWFWorkflowConfig ( default_task_list = TASKLIST , <EOL> default_decision_duration = <NUM_LIT:10> , <EOL> default_workflow_duration = <NUM_LIT> , <EOL> default_child_policy = '<STR_LIT>' ) <EOL> conf_use_activities . conf_activity ( '<STR_LIT>' , VERSION , '<STR_LIT>' ) <EOL> conf_use_activities . conf_activity ( '<STR_LIT>' , VERSION , '<STR_LIT>' , <EOL> schedule_to_close = <NUM_LIT:1> , <EOL> retry = ( <NUM_LIT:0> , ) ) <EOL> conf_use_activities . conf_activity ( '<STR_LIT>' , VERSION , '<STR_LIT>' , <EOL> retry = ( <NUM_LIT:3> , ) ) <EOL> conf_use_activities . conf_activity ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> conf_use_workflow = SWFWorkflowConfig ( default_task_list = TASKLIST , <EOL> default_decision_duration = <NUM_LIT:10> , <EOL> default_workflow_duration = <NUM_LIT> , <EOL> default_child_policy = '<STR_LIT>' ) <EOL> conf_use_workflow . conf_workflow ( '<STR_LIT>' , VERSION , '<STR_LIT>' ) <EOL> conf_use_workflow . conf_workflow ( '<STR_LIT>' , VERSION , '<STR_LIT>' , <EOL> workflow_duration = <NUM_LIT:1> , <EOL> retry = ( <NUM_LIT:0> , ) ) <EOL> conf_use_workflow . conf_workflow ( '<STR_LIT>' , VERSION , '<STR_LIT>' , <EOL> retry = ( <NUM_LIT:3> , ) ) <EOL> conf_use_workflow . conf_workflow ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> @ conf_use_activities ( version = VERSION ) <EOL> @ conf_use_workflow ( version = VERSION , name = '<STR_LIT>' ) <EOL> class TestWorkflow ( BaseWorkflow ) : <EOL> def __init__ ( self , task , short_task , delayed_task , non_existing_task ) : <EOL> self . task = task <EOL> self . short_task = short_task <EOL> self . delayed_task = delayed_task <EOL> self . non_existing_task = non_existing_task <EOL> def call ( self ) : <EOL> tasks = [ self . task ( <NUM_LIT:10> ) , <EOL> self . task ( err = u'<STR_LIT>' ) , <EOL> self . task ( heartbeat = True ) , <EOL> self . short_task ( sleep = <NUM_LIT:3> ) , <EOL> self . delayed_task ( <NUM_LIT:20> ) , <EOL> self . non_existing_task ( ) , ] <EOL> last = self . task ( <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> for _ in range ( <NUM_LIT:20> ) : <EOL> last = self . task ( last , <NUM_LIT:1> ) <EOL> tasks . append ( last ) <EOL> for t in tasks : <EOL> try : <EOL> wait ( t ) <EOL> except TaskError : <EOL> pass <EOL> @ empty_conf ( version = VERSION ) <EOL> class RestartWorkflow ( BaseWorkflow ) : <EOL> def __init__ ( self , activity ) : <EOL> pass <EOL> def call ( self , should_restart = True ) : <EOL> if should_restart : <EOL> return restart ( should_restart = False ) <EOL> return <NUM_LIT:1> <EOL> @ empty_conf ( version = VERSION ) <EOL> class ExitWorkflow ( object ) : <EOL> def __init__ ( self , activity ) : <EOL> exit_event . set ( ) <EOL> wait ( activity ( ) ) <EOL> def __call__ ( self ) : <EOL> pass <EOL> wworker = TestSWFWorkflowWorker ( ) <EOL> wworker . scan ( package = sys . modules [ __name__ ] ) <EOL> aworker = TestSWFActivityWorker ( ) <EOL> aworker . scan ( package = sys . modules [ __name__ ] ) <EOL> body_cache = { } <EOL> def body_as_dict ( r1 , r2 ) : <EOL> if r1 not in body_cache : <EOL> r1b = r1 . body if isinstance ( r1 . body , str ) else r1 . body . decode ( '<STR_LIT:utf-8>' ) <EOL> body_cache [ r1 ] = json . loads ( r1b ) <EOL> if r2 not in body_cache : <EOL> r2b = r2 . body if isinstance ( r2 . body , str ) else r2 . body . decode ( '<STR_LIT:utf-8>' ) <EOL> body_cache [ r2 ] = json . loads ( r2b ) <EOL> return body_cache [ r1 ] == body_cache [ r2 ] <EOL> def escaped_headers ( r1 , r2 ) : <EOL> import urllib <EOL> r1h = dict ( ( h , urllib . unquote ( v ) ) for h , v in r1 . headers . items ( ) ) <EOL> r2h = dict ( ( h , urllib . unquote ( v ) ) for h , v in r1 . headers . items ( ) ) <EOL> return r1 == r2 <EOL> vcr . default_vcr . register_matcher ( '<STR_LIT>' , body_as_dict ) <EOL> vcr . default_vcr . register_matcher ( '<STR_LIT>' , body_as_dict ) <EOL> cassette_args = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:host>' , <EOL> '<STR_LIT:port>' , '<STR_LIT:path>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> } <EOL> def test_activity_integration ( ) : <EOL> with vcr . use_cassette ( A_CASSETTE , <EOL> record_mode = '<STR_LIT:none>' , ** cassette_args ) as cass : <EOL> try : <EOL> l1 = Layer1 ( aws_access_key_id = '<STR_LIT:x>' , aws_secret_access_key = '<STR_LIT:x>' ) <EOL> aworker . run_forever ( DOMAIN , TASKLIST , <EOL> identity = IDENTITY , <EOL> layer1 = l1 , <EOL> setup_log = False ) <EOL> except vcr . errors . CannotOverwriteExistingCassetteException : <EOL> pass <EOL> assert cass . all_played <EOL> def test_workflow_integration ( ) : <EOL> with vcr . use_cassette ( W_CASSETTE , <EOL> record_mode = '<STR_LIT:none>' , ** cassette_args ) as cass : <EOL> try : <EOL> l1 = Layer1 ( aws_access_key_id = '<STR_LIT:x>' , aws_secret_access_key = '<STR_LIT:x>' ) <EOL> wworker . run_forever ( DOMAIN , TASKLIST , <EOL> identity = IDENTITY , <EOL> layer1 = l1 , <EOL> setup_log = False ) <EOL> except vcr . errors . CannotOverwriteExistingCassetteException : <EOL> pass <EOL> assert cass . all_played <EOL> def start_activity_worker ( ) : <EOL> with vcr . use_cassette ( A_CASSETTE , <EOL> record_mode = '<STR_LIT:all>' , ** cassette_args ) as cass : <EOL> try : <EOL> aworker . run_forever ( DOMAIN , TASKLIST , identity = IDENTITY ) <EOL> except vcr . errors . CannotOverwriteExistingCassetteException : <EOL> pass <EOL> def start_workflow_worker ( ) : <EOL> with vcr . use_cassette ( W_CASSETTE , <EOL> record_mode = '<STR_LIT:all>' , ** cassette_args ) as cass : <EOL> try : <EOL> wworker . run_forever ( DOMAIN , TASKLIST , identity = IDENTITY ) <EOL> except vcr . errors . CannotOverwriteExistingCassetteException : <EOL> pass <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> RECORDING = True <EOL> try : <EOL> os . remove ( A_CASSETTE ) <EOL> except : <EOL> pass <EOL> try : <EOL> os . remove ( W_CASSETTE ) <EOL> except : <EOL> pass <EOL> a_worker = multiprocessing . Process ( target = start_activity_worker ) <EOL> w_worker = multiprocessing . Process ( target = start_workflow_worker ) <EOL> a_worker . start ( ) <EOL> w_worker . start ( ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> wfs = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for wf in wfs : <EOL> print ( '<STR_LIT>' , wf ) <EOL> SWFWorkflowStarter ( DOMAIN , wf , VERSION ) ( ) <EOL> wf_finished_event . wait ( ) <EOL> wf_finished_event . clear ( ) <EOL> print ( '<STR_LIT>' ) <EOL> SWFWorkflowStarter ( DOMAIN , '<STR_LIT>' , VERSION ) ( ) <EOL> a_worker . join ( ) <EOL> w_worker . join ( ) </s>
