<s> import st2actions <EOL> from st2common . constants . action import LIVEACTION_STATUS_REQUESTED <EOL> from st2common . constants . action import LIVEACTION_STATUS_SUCCEEDED <EOL> from st2common . constants . action import LIVEACTION_STATUS_TIMED_OUT <EOL> from st2common . bootstrap . policiesregistrar import register_policy_types <EOL> from st2common . models . db . action import LiveActionDB <EOL> from st2common . persistence . action import LiveAction , ActionExecution <EOL> from st2common . services import action as action_service <EOL> from st2actions . policies . retry import ExecutionRetryPolicyApplicator <EOL> from st2tests . base import DbTestCase <EOL> from st2tests . base import CleanDbTestCase <EOL> from st2tests . fixturesloader import FixturesLoader <EOL> PACK = '<STR_LIT>' <EOL> TEST_FIXTURES = { <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' <EOL> ] <EOL> } <EOL> class RetryPolicyTestCase ( CleanDbTestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> DbTestCase . setUpClass ( ) <EOL> super ( RetryPolicyTestCase , cls ) . setUpClass ( ) <EOL> def setUp ( self ) : <EOL> super ( RetryPolicyTestCase , self ) . setUp ( ) <EOL> register_policy_types ( st2actions ) <EOL> loader = FixturesLoader ( ) <EOL> models = loader . save_fixtures_to_db ( fixtures_pack = PACK , <EOL> fixtures_dict = TEST_FIXTURES ) <EOL> policy_db = models [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> retry_on = policy_db . parameters [ '<STR_LIT>' ] <EOL> max_retry_count = policy_db . parameters [ '<STR_LIT>' ] <EOL> self . policy = ExecutionRetryPolicyApplicator ( policy_ref = '<STR_LIT>' , <EOL> policy_type = '<STR_LIT>' , <EOL> retry_on = retry_on , <EOL> max_retry_count = max_retry_count , <EOL> delay = <NUM_LIT:0> ) <EOL> def test_retry_on_timeout_no_retry_since_no_timeout_reached ( self ) : <EOL> self . assertSequenceEqual ( LiveAction . get_all ( ) , [ ] ) <EOL> self . assertSequenceEqual ( ActionExecution . get_all ( ) , [ ] ) <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT:foo>' } ) <EOL> live_action_db , execution_db = action_service . request ( liveaction ) <EOL> live_action_db . status = LIVEACTION_STATUS_SUCCEEDED <EOL> execution_db . status = LIVEACTION_STATUS_SUCCEEDED <EOL> LiveAction . add_or_update ( live_action_db ) <EOL> ActionExecution . add_or_update ( execution_db ) <EOL> self . policy . apply_after ( target = live_action_db ) <EOL> live_action_dbs = LiveAction . get_all ( ) <EOL> action_execution_dbs = ActionExecution . get_all ( ) <EOL> self . assertEqual ( len ( live_action_dbs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( action_execution_dbs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( action_execution_dbs [ <NUM_LIT:0> ] . status , LIVEACTION_STATUS_SUCCEEDED ) <EOL> def test_retry_on_timeout_first_retry_is_successful ( self ) : <EOL> self . assertSequenceEqual ( LiveAction . get_all ( ) , [ ] ) <EOL> self . assertSequenceEqual ( ActionExecution . get_all ( ) , [ ] ) <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT:foo>' } ) <EOL> live_action_db , execution_db = action_service . request ( liveaction ) <EOL> live_action_db . status = LIVEACTION_STATUS_TIMED_OUT <EOL> execution_db . status = LIVEACTION_STATUS_TIMED_OUT <EOL> LiveAction . add_or_update ( live_action_db ) <EOL> ActionExecution . add_or_update ( execution_db ) <EOL> self . policy . apply_after ( target = live_action_db ) <EOL> live_action_dbs = LiveAction . get_all ( ) <EOL> action_execution_dbs = ActionExecution . get_all ( ) <EOL> self . assertEqual ( len ( live_action_dbs ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( action_execution_dbs ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( action_execution_dbs [ <NUM_LIT:0> ] . status , LIVEACTION_STATUS_TIMED_OUT ) <EOL> self . assertEqual ( action_execution_dbs [ <NUM_LIT:1> ] . status , LIVEACTION_STATUS_REQUESTED ) <EOL> live_action_db = live_action_dbs [ <NUM_LIT:1> ] <EOL> live_action_db . status = LIVEACTION_STATUS_SUCCEEDED <EOL> LiveAction . add_or_update ( live_action_db ) <EOL> self . policy . apply_after ( target = live_action_db ) <EOL> live_action_dbs = LiveAction . get_all ( ) <EOL> action_execution_dbs = ActionExecution . get_all ( ) <EOL> self . assertEqual ( len ( live_action_dbs ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( action_execution_dbs ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( live_action_dbs [ <NUM_LIT:0> ] . status , LIVEACTION_STATUS_TIMED_OUT ) <EOL> self . assertEqual ( live_action_dbs [ <NUM_LIT:1> ] . status , LIVEACTION_STATUS_SUCCEEDED ) <EOL> def test_retry_on_timeout_max_retries_reached ( self ) : <EOL> self . assertSequenceEqual ( LiveAction . get_all ( ) , [ ] ) <EOL> self . assertSequenceEqual ( ActionExecution . get_all ( ) , [ ] ) <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT:foo>' } ) <EOL> live_action_db , execution_db = action_service . request ( liveaction ) <EOL> live_action_db . status = LIVEACTION_STATUS_TIMED_OUT <EOL> live_action_db . context [ '<STR_LIT>' ] = { } <EOL> live_action_db . context [ '<STR_LIT>' ] [ '<STR_LIT>' ] = { '<STR_LIT>' : <NUM_LIT:2> } <EOL> execution_db . status = LIVEACTION_STATUS_TIMED_OUT <EOL> LiveAction . add_or_update ( live_action_db ) <EOL> ActionExecution . add_or_update ( execution_db ) <EOL> self . policy . apply_after ( target = live_action_db ) <EOL> live_action_dbs = LiveAction . get_all ( ) <EOL> action_execution_dbs = ActionExecution . get_all ( ) <EOL> self . assertEqual ( len ( live_action_dbs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( action_execution_dbs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( action_execution_dbs [ <NUM_LIT:0> ] . status , LIVEACTION_STATUS_TIMED_OUT ) </s>
