<s> import copy <EOL> import mock <EOL> import six <EOL> from st2common . constants . rules import RULE_TYPE_STANDARD , RULE_TYPE_BACKSTOP <EOL> from st2common . constants . pack import DEFAULT_PACK_NAME <EOL> from st2common . persistence . trigger import Trigger <EOL> from st2common . models . system . common import ResourceReference <EOL> from st2common . transport . publishers import PoolPublisher <EOL> from st2tests . fixturesloader import FixturesLoader <EOL> from tests import FunctionalTest <EOL> http_client = six . moves . http_client <EOL> TEST_FIXTURES = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] <EOL> } <EOL> FIXTURES_PACK = '<STR_LIT>' <EOL> @ mock . patch . object ( PoolPublisher , '<STR_LIT>' , mock . MagicMock ( ) ) <EOL> class TestRuleController ( FunctionalTest ) : <EOL> fixtures_loader = FixturesLoader ( ) <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( TestRuleController , cls ) . setUpClass ( ) <EOL> models = TestRuleController . fixtures_loader . save_fixtures_to_db ( <EOL> fixtures_pack = FIXTURES_PACK , fixtures_dict = TEST_FIXTURES ) <EOL> TestRuleController . RUNNER_TYPE = models [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> TestRuleController . ACTION = models [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> TestRuleController . TRIGGER = models [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_1 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_2 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_3 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_4 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_5 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_6 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_7 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> file_name = '<STR_LIT>' <EOL> TestRuleController . RULE_8 = TestRuleController . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> TestRuleController . fixtures_loader . delete_fixtures_from_db ( <EOL> fixtures_pack = FIXTURES_PACK , fixtures_dict = TEST_FIXTURES ) <EOL> super ( TestRuleController , cls ) . setUpClass ( ) <EOL> def test_get_all ( self ) : <EOL> resp = self . app . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , http_client . OK ) <EOL> def test_get_one_by_id ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> rule_id = self . __get_rule_id ( post_resp ) <EOL> get_resp = self . __do_get_one ( rule_id ) <EOL> self . assertEqual ( get_resp . status_int , http_client . OK ) <EOL> self . assertEqual ( self . __get_rule_id ( get_resp ) , rule_id ) <EOL> self . __do_delete ( rule_id ) <EOL> def test_get_one_by_ref ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> rule_name = post_resp . json [ '<STR_LIT:name>' ] <EOL> rule_pack = post_resp . json [ '<STR_LIT>' ] <EOL> ref = ResourceReference . to_string_reference ( name = rule_name , pack = rule_pack ) <EOL> rule_id = post_resp . json [ '<STR_LIT:id>' ] <EOL> get_resp = self . __do_get_one ( ref ) <EOL> self . assertEqual ( get_resp . json [ '<STR_LIT:name>' ] , rule_name ) <EOL> self . assertEqual ( get_resp . status_int , http_client . OK ) <EOL> self . __do_delete ( rule_id ) <EOL> def test_get_one_fail ( self ) : <EOL> resp = self . app . get ( '<STR_LIT>' , expect_errors = True ) <EOL> self . assertEqual ( resp . status_int , http_client . NOT_FOUND ) <EOL> def test_post ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> self . __do_delete ( self . __get_rule_id ( post_resp ) ) <EOL> def test_post_duplicate ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> org_id = self . __get_rule_id ( post_resp ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> post_resp_2 = self . __do_post ( TestRuleController . RULE_1 ) <EOL> self . assertEqual ( post_resp_2 . status_int , http_client . CONFLICT ) <EOL> self . assertEqual ( post_resp_2 . json [ '<STR_LIT>' ] , org_id ) <EOL> self . __do_delete ( org_id ) <EOL> def test_post_trigger_parameter_schema_validation_fails ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_2 ) <EOL> self . assertEqual ( post_resp . status_int , http_client . BAD_REQUEST ) <EOL> expected_msg = '<STR_LIT>' <EOL> self . assertTrue ( expected_msg in post_resp . body ) <EOL> def test_post_trigger_parameter_schema_validation_fails_missing_required_param ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_5 ) <EOL> self . assertEqual ( post_resp . status_int , http_client . BAD_REQUEST ) <EOL> expected_msg = '<STR_LIT>' <EOL> self . assertTrue ( expected_msg in post_resp . body ) <EOL> def test_post_invalid_crontimer_trigger_parameters ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_6 ) <EOL> self . assertEqual ( post_resp . status_int , http_client . BAD_REQUEST ) <EOL> expected_msg = '<STR_LIT>' <EOL> self . assertTrue ( expected_msg in post_resp . body ) <EOL> post_resp = self . __do_post ( TestRuleController . RULE_7 ) <EOL> self . assertEqual ( post_resp . status_int , http_client . BAD_REQUEST ) <EOL> expected_msg = '<STR_LIT>' <EOL> self . assertTrue ( expected_msg in post_resp . body ) <EOL> post_resp = self . __do_post ( TestRuleController . RULE_8 ) <EOL> self . assertEqual ( post_resp . status_int , http_client . BAD_REQUEST ) <EOL> expected_msg = '<STR_LIT>' <EOL> self . assertTrue ( expected_msg in post_resp . body ) <EOL> def test_post_no_enabled_attribute_disabled_by_default ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_3 ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> self . assertFalse ( post_resp . json [ '<STR_LIT>' ] ) <EOL> self . __do_delete ( self . __get_rule_id ( post_resp ) ) <EOL> def test_put ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> update_input = post_resp . json <EOL> update_input [ '<STR_LIT>' ] = not update_input [ '<STR_LIT>' ] <EOL> put_resp = self . __do_put ( self . __get_rule_id ( post_resp ) , update_input ) <EOL> self . assertEqual ( put_resp . status_int , http_client . OK ) <EOL> self . __do_delete ( self . __get_rule_id ( put_resp ) ) <EOL> def test_post_no_pack_info ( self ) : <EOL> rule = copy . deepcopy ( TestRuleController . RULE_1 ) <EOL> del rule [ '<STR_LIT>' ] <EOL> post_resp = self . __do_post ( rule ) <EOL> self . assertEqual ( post_resp . json [ '<STR_LIT>' ] , DEFAULT_PACK_NAME ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> self . __do_delete ( self . __get_rule_id ( post_resp ) ) <EOL> def test_put_no_pack_info ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> test_rule = post_resp . json <EOL> if '<STR_LIT>' in test_rule : <EOL> del test_rule [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT>' not in test_rule ) <EOL> put_resp = self . __do_put ( self . __get_rule_id ( post_resp ) , test_rule ) <EOL> self . assertEqual ( put_resp . json [ '<STR_LIT>' ] , DEFAULT_PACK_NAME ) <EOL> self . assertEqual ( put_resp . status_int , http_client . OK ) <EOL> self . __do_delete ( self . __get_rule_id ( put_resp ) ) <EOL> def test_put_fail ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> update_input = post_resp . json <EOL> put_resp = self . __do_put ( <NUM_LIT:1> , update_input ) <EOL> self . assertEqual ( put_resp . status_int , http_client . NOT_FOUND ) <EOL> self . __do_delete ( self . __get_rule_id ( post_resp ) ) <EOL> def test_delete ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> del_resp = self . __do_delete ( self . __get_rule_id ( post_resp ) ) <EOL> self . assertEqual ( del_resp . status_int , http_client . NO_CONTENT ) <EOL> def test_rule_with_tags ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> rule_id = self . __get_rule_id ( post_resp ) <EOL> get_resp = self . __do_get_one ( rule_id ) <EOL> self . assertEqual ( get_resp . status_int , http_client . OK ) <EOL> self . assertEqual ( self . __get_rule_id ( get_resp ) , rule_id ) <EOL> self . assertEqual ( get_resp . json [ '<STR_LIT>' ] , TestRuleController . RULE_1 [ '<STR_LIT>' ] ) <EOL> self . __do_delete ( rule_id ) <EOL> def test_rule_without_type ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_1 ) <EOL> rule_id = self . __get_rule_id ( post_resp ) <EOL> get_resp = self . __do_get_one ( rule_id ) <EOL> self . assertEqual ( get_resp . status_int , http_client . OK ) <EOL> self . assertEqual ( self . __get_rule_id ( get_resp ) , rule_id ) <EOL> assigned_rule_type = get_resp . json [ '<STR_LIT:type>' ] <EOL> self . assertTrue ( assigned_rule_type , '<STR_LIT>' ) <EOL> self . assertEqual ( assigned_rule_type [ '<STR_LIT>' ] , RULE_TYPE_STANDARD , <EOL> '<STR_LIT>' ) <EOL> self . __do_delete ( rule_id ) <EOL> def test_rule_with_type ( self ) : <EOL> post_resp = self . __do_post ( TestRuleController . RULE_4 ) <EOL> rule_id = self . __get_rule_id ( post_resp ) <EOL> get_resp = self . __do_get_one ( rule_id ) <EOL> self . assertEqual ( get_resp . status_int , http_client . OK ) <EOL> self . assertEqual ( self . __get_rule_id ( get_resp ) , rule_id ) <EOL> assigned_rule_type = get_resp . json [ '<STR_LIT:type>' ] <EOL> self . assertTrue ( assigned_rule_type , '<STR_LIT>' ) <EOL> self . assertEqual ( assigned_rule_type [ '<STR_LIT>' ] , RULE_TYPE_BACKSTOP , <EOL> '<STR_LIT>' ) <EOL> self . __do_delete ( rule_id ) <EOL> @ staticmethod <EOL> def __get_rule_id ( resp ) : <EOL> return resp . json [ '<STR_LIT:id>' ] <EOL> def __do_get_one ( self , rule_id ) : <EOL> return self . app . get ( '<STR_LIT>' % rule_id , expect_errors = True ) <EOL> @ mock . patch . object ( PoolPublisher , '<STR_LIT>' , mock . MagicMock ( ) ) <EOL> def __do_post ( self , rule ) : <EOL> return self . app . post_json ( '<STR_LIT>' , rule , expect_errors = True ) <EOL> @ mock . patch . object ( PoolPublisher , '<STR_LIT>' , mock . MagicMock ( ) ) <EOL> def __do_put ( self , rule_id , rule ) : <EOL> return self . app . put_json ( '<STR_LIT>' % rule_id , rule , expect_errors = True ) <EOL> @ mock . patch . object ( PoolPublisher , '<STR_LIT>' , mock . MagicMock ( ) ) <EOL> def __do_delete ( self , rule_id ) : <EOL> return self . app . delete ( '<STR_LIT>' % rule_id ) <EOL> TEST_FIXTURES_2 = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] <EOL> } <EOL> @ mock . patch . object ( PoolPublisher , '<STR_LIT>' , mock . MagicMock ( ) ) <EOL> class TestRuleControllerTriggerCreator ( FunctionalTest ) : <EOL> fixtures_loader = FixturesLoader ( ) <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( TestRuleControllerTriggerCreator , cls ) . setUpClass ( ) <EOL> cls . models = cls . fixtures_loader . save_fixtures_to_db ( <EOL> fixtures_pack = FIXTURES_PACK , fixtures_dict = TEST_FIXTURES_2 ) <EOL> file_name = '<STR_LIT>' <EOL> cls . RULE_1 = cls . fixtures_loader . load_fixtures ( <EOL> fixtures_pack = FIXTURES_PACK , <EOL> fixtures_dict = { '<STR_LIT>' : [ file_name ] } ) [ '<STR_LIT>' ] [ file_name ] <EOL> def test_ref_count_trigger_increment ( self ) : <EOL> post_resp = self . __do_post ( self . RULE_1 ) <EOL> rule_1_id = self . __get_rule_id ( post_resp ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> triggers = Trigger . get_all ( ** { '<STR_LIT:type>' : post_resp . json [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] } ) <EOL> self . assertEqual ( len ( triggers ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( triggers [ <NUM_LIT:0> ] . ref_count , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> rule_2 = copy . copy ( self . RULE_1 ) <EOL> rule_2 [ '<STR_LIT:name>' ] = rule_2 [ '<STR_LIT:name>' ] + '<STR_LIT>' <EOL> post_resp = self . __do_post ( rule_2 ) <EOL> rule_2_id = self . __get_rule_id ( post_resp ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> triggers = Trigger . get_all ( ** { '<STR_LIT:type>' : post_resp . json [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] } ) <EOL> self . assertEqual ( len ( triggers ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( triggers [ <NUM_LIT:0> ] . ref_count , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> self . __do_delete ( rule_1_id ) <EOL> self . __do_delete ( rule_2_id ) <EOL> def test_ref_count_trigger_decrement ( self ) : <EOL> post_resp = self . __do_post ( self . RULE_1 ) <EOL> rule_1_id = self . __get_rule_id ( post_resp ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> rule_2 = copy . copy ( self . RULE_1 ) <EOL> rule_2 [ '<STR_LIT:name>' ] = rule_2 [ '<STR_LIT:name>' ] + '<STR_LIT>' <EOL> post_resp = self . __do_post ( rule_2 ) <EOL> rule_2_id = self . __get_rule_id ( post_resp ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> self . __do_delete ( rule_1_id ) <EOL> triggers = Trigger . get_all ( ** { '<STR_LIT:type>' : post_resp . json [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] } ) <EOL> self . assertEqual ( len ( triggers ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( triggers [ <NUM_LIT:0> ] . ref_count , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . __do_delete ( rule_2_id ) <EOL> def test_trigger_cleanup ( self ) : <EOL> post_resp = self . __do_post ( self . RULE_1 ) <EOL> rule_1_id = self . __get_rule_id ( post_resp ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> rule_2 = copy . copy ( self . RULE_1 ) <EOL> rule_2 [ '<STR_LIT:name>' ] = rule_2 [ '<STR_LIT:name>' ] + '<STR_LIT>' <EOL> post_resp = self . __do_post ( rule_2 ) <EOL> rule_2_id = self . __get_rule_id ( post_resp ) <EOL> self . assertEqual ( post_resp . status_int , http_client . CREATED ) <EOL> triggers = Trigger . get_all ( ** { '<STR_LIT:type>' : post_resp . json [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] } ) <EOL> self . assertEqual ( len ( triggers ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( triggers [ <NUM_LIT:0> ] . ref_count , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> self . __do_delete ( rule_1_id ) <EOL> self . __do_delete ( rule_2_id ) <EOL> triggers = Trigger . get_all ( ** { '<STR_LIT:type>' : post_resp . json [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] } ) <EOL> self . assertEqual ( len ( triggers ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def __get_rule_id ( resp ) : <EOL> return resp . json [ '<STR_LIT:id>' ] <EOL> def __do_get_one ( self , rule_id ) : <EOL> return self . app . get ( '<STR_LIT>' % rule_id , expect_errors = True ) <EOL> def __do_post ( self , rule ) : <EOL> return self . app . post_json ( '<STR_LIT>' , rule , expect_errors = True ) <EOL> def __do_put ( self , rule_id , rule ) : <EOL> return self . app . put_json ( '<STR_LIT>' % rule_id , rule , expect_errors = True ) <EOL> def __do_delete ( self , rule_id ) : <EOL> return self . app . delete ( '<STR_LIT>' % rule_id ) </s>
