<s> import mock <EOL> from st2common . constants import action as action_constants <EOL> from st2common . models . db . action import LiveActionDB <EOL> from st2common . persistence . action import LiveAction <EOL> from st2common . persistence . policy import Policy <EOL> from st2common . services import action as action_service <EOL> from st2common . transport . liveaction import LiveActionPublisher <EOL> from st2common . transport . publishers import CUDPublisher <EOL> from st2tests import DbTestCase , EventletTestCase <EOL> from st2tests . fixturesloader import FixturesLoader <EOL> from tests . unit . base import MockLiveActionPublisher <EOL> from st2tests . mocks . runner import MockActionRunner <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> '<STR_LIT>' : [ <EOL> '<STR_LIT>' <EOL> ] <EOL> } <EOL> NON_EMPTY_RESULT = '<STR_LIT>' <EOL> SCHEDULED_STATES = [ <EOL> action_constants . LIVEACTION_STATUS_SCHEDULED , <EOL> action_constants . LIVEACTION_STATUS_RUNNING , <EOL> action_constants . LIVEACTION_STATUS_SUCCEEDED <EOL> ] <EOL> @ mock . patch . object ( <EOL> MockActionRunner , '<STR_LIT>' , <EOL> mock . MagicMock ( <EOL> return_value = ( action_constants . LIVEACTION_STATUS_RUNNING , NON_EMPTY_RESULT , None ) ) ) <EOL> @ mock . patch . object ( <EOL> CUDPublisher , '<STR_LIT>' , <EOL> mock . MagicMock ( side_effect = MockLiveActionPublisher . publish_update ) ) <EOL> @ mock . patch . object ( <EOL> CUDPublisher , '<STR_LIT>' , <EOL> mock . MagicMock ( return_value = None ) ) <EOL> @ mock . patch . object ( <EOL> LiveActionPublisher , '<STR_LIT>' , <EOL> mock . MagicMock ( side_effect = MockLiveActionPublisher . publish_state ) ) <EOL> class ConcurrencyByAttributePolicyTest ( EventletTestCase , DbTestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> EventletTestCase . setUpClass ( ) <EOL> DbTestCase . setUpClass ( ) <EOL> loader = FixturesLoader ( ) <EOL> loader . save_fixtures_to_db ( fixtures_pack = PACK , <EOL> fixtures_dict = TEST_FIXTURES ) <EOL> def tearDown ( self ) : <EOL> for liveaction in LiveAction . get_all ( ) : <EOL> action_service . update_status ( <EOL> liveaction , action_constants . LIVEACTION_STATUS_CANCELED ) <EOL> def test_over_threshold ( self ) : <EOL> policy_db = Policy . get_by_ref ( '<STR_LIT>' ) <EOL> self . assertGreater ( policy_db . parameters [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertIn ( '<STR_LIT>' , policy_db . parameters [ '<STR_LIT>' ] ) <EOL> for i in range ( <NUM_LIT:0> , policy_db . parameters [ '<STR_LIT>' ] ) : <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> action_service . request ( liveaction ) <EOL> scheduled = [ item for item in LiveAction . get_all ( ) if item . status in SCHEDULED_STATES ] <EOL> self . assertEqual ( len ( scheduled ) , policy_db . parameters [ '<STR_LIT>' ] ) <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> liveaction , _ = action_service . request ( liveaction ) <EOL> delayed = LiveAction . get_by_id ( str ( liveaction . id ) ) <EOL> self . assertEqual ( delayed . status , action_constants . LIVEACTION_STATUS_DELAYED ) <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT:bar>' } ) <EOL> liveaction , _ = action_service . request ( liveaction ) <EOL> liveaction = LiveAction . get_by_id ( str ( liveaction . id ) ) <EOL> self . assertIn ( liveaction . status , SCHEDULED_STATES ) <EOL> action_service . update_status ( <EOL> scheduled [ <NUM_LIT:0> ] , action_constants . LIVEACTION_STATUS_SUCCEEDED , publish = True ) <EOL> liveaction = LiveAction . get_by_id ( str ( delayed . id ) ) <EOL> self . assertIn ( liveaction . status , SCHEDULED_STATES ) <EOL> def test_on_cancellation ( self ) : <EOL> policy_db = Policy . get_by_ref ( '<STR_LIT>' ) <EOL> self . assertGreater ( policy_db . parameters [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertIn ( '<STR_LIT>' , policy_db . parameters [ '<STR_LIT>' ] ) <EOL> for i in range ( <NUM_LIT:0> , policy_db . parameters [ '<STR_LIT>' ] ) : <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> action_service . request ( liveaction ) <EOL> scheduled = [ item for item in LiveAction . get_all ( ) if item . status in SCHEDULED_STATES ] <EOL> self . assertEqual ( len ( scheduled ) , policy_db . parameters [ '<STR_LIT>' ] ) <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> liveaction , _ = action_service . request ( liveaction ) <EOL> delayed = LiveAction . get_by_id ( str ( liveaction . id ) ) <EOL> self . assertEqual ( delayed . status , action_constants . LIVEACTION_STATUS_DELAYED ) <EOL> liveaction = LiveActionDB ( action = '<STR_LIT>' , parameters = { '<STR_LIT>' : '<STR_LIT:bar>' } ) <EOL> liveaction , _ = action_service . request ( liveaction ) <EOL> liveaction = LiveAction . get_by_id ( str ( liveaction . id ) ) <EOL> self . assertIn ( liveaction . status , SCHEDULED_STATES ) <EOL> action_service . request_cancellation ( scheduled [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> liveaction = LiveAction . get_by_id ( str ( delayed . id ) ) <EOL> self . assertIn ( liveaction . status , SCHEDULED_STATES ) </s>
