Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use put_rule method in localstack

Best Python code snippet using localstack_python

boto_cloudwatch_event_test.py

Source:boto_cloudwatch_event_test.py

1# -*- coding: utf-8 -*-2# Import Python libs3from __future__ import absolute_import4import random5import string6# Import Salt Testing libs7from salttesting.unit import skipIf, TestCase8from salttesting.mock import NO_MOCK, NO_MOCK_REASON, patch9from salttesting.helpers import ensure_in_syspath10ensure_in_syspath('../../')11# Import Salt libs12import salt.config13import salt.loader14# Import 3rd-party libs15import logging16# Import Mock libraries17from salttesting.mock import NO_MOCK, NO_MOCK_REASON, MagicMock, patch18# pylint: disable=import-error,no-name-in-module19from unit.modules.boto_cloudwatch_event_test import BotoCloudWatchEventTestCaseMixin20# pylint: disable=unused-import21# Import 3rd-party libs22try:23    import boto324    from botocore.exceptions import ClientError25    HAS_BOTO = True26except ImportError:27    HAS_BOTO = False28# pylint: enable=unused-import29from salt.ext.six.moves import range30# pylint: enable=import-error,no-name-in-module31region = 'us-east-1'32access_key = 'GKTADJGHEIQSXMKKRBJ08H'33secret_key = 'askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs'34conn_parameters = {'region': region, 'key': access_key, 'keyid': secret_key, 'profile': {}}35error_message = 'An error occurred (101) when calling the {0} operation: Test-defined error'36error_content = {37  'Error': {38    'Code': 101,39    'Message': "Test-defined error"40  }41}42not_found_error = ClientError({43    'Error': {44        'Code': 'ResourceNotFoundException',45        'Message': "Test-defined error"46    }47}, 'msg')48rule_name = 'test_thing_type'49rule_desc = 'test_thing_type_desc'50rule_sched = 'rate(20 min)'51rule_arn = 'arn:::::rule/arn'52rule_ret = dict(53    Arn=rule_arn,54    Description=rule_desc,55    EventPattern=None,56    Name=rule_name,57    RoleArn=None,58    ScheduleExpression=rule_sched,59    State='ENABLED'60)61log = logging.getLogger(__name__)62opts = salt.config.DEFAULT_MINION_OPTS63context = {}64utils = salt.loader.utils(opts, whitelist=['boto3'], context=context)65serializers = salt.loader.serializers(opts)66funcs = salt.loader.minion_mods(opts, context=context, utils=utils, whitelist=['boto_cloudwatch_event'])67salt_states = salt.loader.states(opts=opts, functions=funcs, utils=utils, whitelist=['boto_cloudwatch_event'], serializers=serializers)68def _has_required_boto():69    '''70    Returns True/False boolean depending on if Boto is installed and correct71    version.72    '''73    if not HAS_BOTO:74        return False75    else:76        return True77class BotoCloudWatchEventStateTestCaseBase(TestCase):78    conn = None79    # Set up MagicMock to replace the boto3 session80    def setUp(self):81        context.clear()82        # connections keep getting cached from prior tests, can't find the83        # correct context object to clear it. So randomize the cache key, to prevent any84        # cache hits85        conn_parameters['key'] = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(50))86        self.patcher = patch('boto3.session.Session')87        self.addCleanup(self.patcher.stop)88        mock_session = self.patcher.start()89        session_instance = mock_session.return_value90        self.conn = MagicMock()91        session_instance.client.return_value = self.conn92@skipIf(HAS_BOTO is False, 'The boto module must be installed.')93@skipIf(NO_MOCK, NO_MOCK_REASON)94class BotoCloudWatchEventTestCase(BotoCloudWatchEventStateTestCaseBase, BotoCloudWatchEventTestCaseMixin):95    def test_present_when_failing_to_describe_rule(self):96        '''97        Tests exceptions when checking rule existence98        '''99        self.conn.list_rules.side_effect = ClientError(error_content, 'error on list rules')100        result = salt_states['boto_cloudwatch_event.present'](101                             name='test present',102                             Name=rule_name,103                             Description=rule_desc,104                             ScheduleExpression=rule_sched,105                             Targets=[{106                               'Id': 'target1',107                               'Arn': 'arn::::::*',108                             }],109                             **conn_parameters)110        self.assertEqual(result.get('result'), False)111        self.assertTrue('error on list rules' in result.get('comment', {}))112    def test_present_when_failing_to_create_a_new_rule(self):113        '''114        Tests present on a rule name that doesn't exist and115        an error is thrown on creation.116        '''117        self.conn.list_rules.return_value = {'Rules': []}118        self.conn.put_rule.side_effect = ClientError(error_content, 'put_rule')119        result = salt_states['boto_cloudwatch_event.present'](120                             name='test present',121                             Name=rule_name,122                             Description=rule_desc,123                             ScheduleExpression=rule_sched,124                             Targets=[{125                               'Id': 'target1',126                               'Arn': 'arn::::::*',127                             }],128                             **conn_parameters)129        self.assertEqual(result.get('result'), False)130        self.assertTrue('put_rule' in result.get('comment', ''))131    def test_present_when_failing_to_describe_the_new_rule(self):132        '''133        Tests present on a rule name that doesn't exist and134        an error is thrown when adding targets.135        '''136        self.conn.list_rules.return_value = {'Rules': []}137        self.conn.put_rule.return_value = rule_ret138        self.conn.describe_rule.side_effect = ClientError(error_content, 'describe_rule')139        result = salt_states['boto_cloudwatch_event.present'](140                             name='test present',141                             Name=rule_name,142                             Description=rule_desc,143                             ScheduleExpression=rule_sched,144                             Targets=[{145                               'Id': 'target1',146                               'Arn': 'arn::::::*',147                             }],148                             **conn_parameters)149        self.assertEqual(result.get('result'), False)150        self.assertTrue('describe_rule' in result.get('comment', ''))151    def test_present_when_failing_to_create_a_new_rules_targets(self):152        '''153        Tests present on a rule name that doesn't exist and154        an error is thrown when adding targets.155        '''156        self.conn.list_rules.return_value = {'Rules': []}157        self.conn.put_rule.return_value = rule_ret158        self.conn.describe_rule.return_value = rule_ret159        self.conn.put_targets.side_effect = ClientError(error_content, 'put_targets')160        result = salt_states['boto_cloudwatch_event.present'](161                             name='test present',162                             Name=rule_name,163                             Description=rule_desc,164                             ScheduleExpression=rule_sched,165                             Targets=[{166                               'Id': 'target1',167                               'Arn': 'arn::::::*',168                             }],169                             **conn_parameters)170        self.assertEqual(result.get('result'), False)171        self.assertTrue('put_targets' in result.get('comment', ''))172    def test_present_when_rule_does_not_exist(self):173        '''174        Tests the successful case of creating a new rule, and updating its175        targets176        '''177        self.conn.list_rules.return_value = {'Rules': []}178        self.conn.put_rule.return_value = rule_ret179        self.conn.describe_rule.return_value = rule_ret180        self.conn.put_targets.return_value = {'FailedEntryCount': 0}181        result = salt_states['boto_cloudwatch_event.present'](182                             name='test present',183                             Name=rule_name,184                             Description=rule_desc,185                             ScheduleExpression=rule_sched,186                             Targets=[{187                               'Id': 'target1',188                               'Arn': 'arn::::::*',189                             }],190                             **conn_parameters)191        self.assertEqual(result.get('result'), True)192    def test_present_when_failing_to_update_an_existing_rule(self):193        '''194        Tests present on an existing rule where an error is thrown on updating the pool properties.195        '''196        self.conn.list_rules.return_value = {'Rules': [rule_ret]}197        self.conn.describe_rule.side_effect = ClientError(error_content, 'describe_rule')198        result = salt_states['boto_cloudwatch_event.present'](199                             name='test present',200                             Name=rule_name,201                             Description=rule_desc,202                             ScheduleExpression=rule_sched,203                             Targets=[{204                               'Id': 'target1',205                               'Arn': 'arn::::::*',206                             }],207                             **conn_parameters)208        self.assertEqual(result.get('result'), False)209        self.assertTrue('describe_rule' in result.get('comment', ''))210    def test_present_when_failing_to_get_targets(self):211        '''212        Tests present on an existing rule where put_rule succeeded, but an error213        is thrown on getting targets214        '''215        self.conn.list_rules.return_value = {'Rules': [rule_ret]}216        self.conn.put_rule.return_value = rule_ret217        self.conn.describe_rule.return_value = rule_ret218        self.conn.list_targets_by_rule.side_effect = ClientError(error_content, 'list_targets')219        result = salt_states['boto_cloudwatch_event.present'](220                             name='test present',221                             Name=rule_name,222                             Description=rule_desc,223                             ScheduleExpression=rule_sched,224                             Targets=[{225                               'Id': 'target1',226                               'Arn': 'arn::::::*',227                             }],228                             **conn_parameters)229        self.assertEqual(result.get('result'), False)230        self.assertTrue('list_targets' in result.get('comment', ''))231    def test_present_when_failing_to_put_targets(self):232        '''233        Tests present on an existing rule where put_rule succeeded, but an error234        is thrown on putting targets235        '''236        self.conn.list_rules.return_value = {'Rules': []}237        self.conn.put_rule.return_value = rule_ret238        self.conn.describe_rule.return_value = rule_ret239        self.conn.list_targets.return_value = {'Targets': []}240        self.conn.put_targets.side_effect = ClientError(error_content, 'put_targets')241        result = salt_states['boto_cloudwatch_event.present'](242                             name='test present',243                             Name=rule_name,244                             Description=rule_desc,245                             ScheduleExpression=rule_sched,246                             Targets=[{247                               'Id': 'target1',248                               'Arn': 'arn::::::*',249                             }],250                             **conn_parameters)251        self.assertEqual(result.get('result'), False)252        self.assertTrue('put_targets' in result.get('comment', ''))253    def test_present_when_putting_targets(self):254        '''255        Tests present on an existing rule where put_rule succeeded, and targets256        must be added257        '''258        self.conn.list_rules.return_value = {'Rules': []}259        self.conn.put_rule.return_value = rule_ret260        self.conn.describe_rule.return_value = rule_ret261        self.conn.list_targets.return_value = {'Targets': []}262        self.conn.put_targets.return_value = {'FailedEntryCount': 0}263        result = salt_states['boto_cloudwatch_event.present'](264                             name='test present',265                             Name=rule_name,266                             Description=rule_desc,267                             ScheduleExpression=rule_sched,268                             Targets=[{269                               'Id': 'target1',270                               'Arn': 'arn::::::*',271                             }],272                             **conn_parameters)273        self.assertEqual(result.get('result'), True)274    def test_present_when_removing_targets(self):275        '''276        Tests present on an existing rule where put_rule succeeded, and targets277        must be removed278        '''279        self.conn.list_rules.return_value = {'Rules': []}280        self.conn.put_rule.return_value = rule_ret281        self.conn.describe_rule.return_value = rule_ret282        self.conn.list_targets.return_value = {'Targets': [{'Id': 'target1'}, {'Id': 'target2'}]}283        self.conn.put_targets.return_value = {'FailedEntryCount': 0}284        result = salt_states['boto_cloudwatch_event.present'](285                             name='test present',286                             Name=rule_name,287                             Description=rule_desc,288                             ScheduleExpression=rule_sched,289                             Targets=[{290                               'Id': 'target1',291                               'Arn': 'arn::::::*',292                             }],293                             **conn_parameters)294        self.assertEqual(result.get('result'), True)295    def test_absent_when_failing_to_describe_rule(self):296        '''297        Tests exceptions when checking rule existence298        '''299        self.conn.list_rules.side_effect = ClientError(error_content, 'error on list rules')300        result = salt_states['boto_cloudwatch_event.absent'](301                             name='test present',302                             Name=rule_name,303                             **conn_parameters)304        self.assertEqual(result.get('result'), False)305        self.assertTrue('error on list rules' in result.get('comment', {}))306    def test_absent_when_rule_does_not_exist(self):307        '''308        Tests absent on an non-existing rule309        '''310        self.conn.list_rules.return_value = {'Rules': []}311        result = salt_states['boto_cloudwatch_event.absent'](312                             name='test absent',313                             Name=rule_name,314                             **conn_parameters)315        self.assertEqual(result.get('result'), True)316        self.assertEqual(result['changes'], {})317    def test_absent_when_failing_to_list_targets(self):318        '''319        Tests absent on an rule when the list_targets call fails320        '''321        self.conn.list_rules.return_value = {'Rules': [rule_ret]}322        self.conn.list_targets_by_rule.side_effect = ClientError(error_content, 'list_targets')323        result = salt_states['boto_cloudwatch_event.absent'](324                             name='test absent',325                             Name=rule_name,326                             **conn_parameters)327        self.assertEqual(result.get('result'), False)328        self.assertTrue('list_targets' in result.get('comment', ''))329    def test_absent_when_failing_to_remove_targets_exception(self):330        '''331        Tests absent on an rule when the remove_targets call fails332        '''333        self.conn.list_rules.return_value = {'Rules': [rule_ret]}334        self.conn.list_targets_by_rule.return_value = {'Targets': [{'Id': 'target1'}]}335        self.conn.remove_targets.side_effect = ClientError(error_content, 'remove_targets')336        result = salt_states['boto_cloudwatch_event.absent'](337                             name='test absent',338                             Name=rule_name,339                             **conn_parameters)340        self.assertEqual(result.get('result'), False)341        self.assertTrue('remove_targets' in result.get('comment', ''))342    def test_absent_when_failing_to_remove_targets_nonexception(self):343        '''344        Tests absent on an rule when the remove_targets call fails345        '''346        self.conn.list_rules.return_value = {'Rules': [rule_ret]}347        self.conn.list_targets_by_rule.return_value = {'Targets': [{'Id': 'target1'}]}348        self.conn.remove_targets.return_value = {'FailedEntryCount': 1}349        result = salt_states['boto_cloudwatch_event.absent'](350                             name='test absent',351                             Name=rule_name,352                             **conn_parameters)353        self.assertEqual(result.get('result'), False)354    def test_absent_when_failing_to_delete_rule(self):355        '''356        Tests absent on an rule when the delete_rule call fails357        '''358        self.conn.list_rules.return_value = {'Rules': [rule_ret]}359        self.conn.list_targets_by_rule.return_value = {'Targets': [{'Id': 'target1'}]}360        self.conn.remove_targets.return_value = {'FailedEntryCount': 0}361        self.conn.delete_rule.side_effect = ClientError(error_content, 'delete_rule')362        result = salt_states['boto_cloudwatch_event.absent'](363                             name='test absent',364                             Name=rule_name,365                             **conn_parameters)366        self.assertEqual(result.get('result'), False)367        self.assertTrue('delete_rule' in result.get('comment', ''))368    def test_absent(self):369        '''370        Tests absent on an rule371        '''372        self.conn.list_rules.return_value = {'Rules': [rule_ret]}373        self.conn.list_targets_by_rule.return_value = {'Targets': [{'Id': 'target1'}]}374        self.conn.remove_targets.return_value = {'FailedEntryCount': 0}375        result = salt_states['boto_cloudwatch_event.absent'](376                             name='test absent',377                             Name=rule_name,378                             **conn_parameters)...

test_boto_cloudwatch_event.py

Source:test_boto_cloudwatch_event.py

1# -*- coding: utf-8 -*-2# Import Python libs3from __future__ import absolute_import, print_function, unicode_literals4import logging5import random6import string7# Import Salt libs8import salt.config9import salt.loader10import salt.states.boto_cloudwatch_event as boto_cloudwatch_event11from salt.ext.six.moves import range12# Import Salt Testing libs13from tests.support.mixins import LoaderModuleMockMixin14from tests.support.mock import MagicMock, patch15from tests.support.unit import TestCase, skipIf16# pylint: disable=import-error,no-name-in-module17from tests.unit.modules.test_boto_cloudwatch_event import (18    BotoCloudWatchEventTestCaseMixin,19)20# pylint: disable=unused-import21# Import 3rd-party libs22try:23    import boto324    from botocore.exceptions import ClientError25    HAS_BOTO = True26except ImportError:27    HAS_BOTO = False28# pylint: enable=unused-import29# pylint: enable=import-error,no-name-in-module30region = "us-east-1"31access_key = "GKTADJGHEIQSXMKKRBJ08H"32secret_key = "askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs"33conn_parameters = {34    "region": region,35    "key": access_key,36    "keyid": secret_key,37    "profile": {},38}39error_message = (40    "An error occurred (101) when calling the {0} operation: Test-defined error"41)42error_content = {"Error": {"Code": 101, "Message": "Test-defined error"}}43if HAS_BOTO:44    not_found_error = ClientError(45        {46            "Error": {47                "Code": "ResourceNotFoundException",48                "Message": "Test-defined error",49            }50        },51        "msg",52    )53rule_name = "test_thing_type"54rule_desc = "test_thing_type_desc"55rule_sched = "rate(20 min)"56rule_arn = "arn:::::rule/arn"57rule_ret = dict(58    Arn=rule_arn,59    Description=rule_desc,60    EventPattern=None,61    Name=rule_name,62    RoleArn=None,63    ScheduleExpression=rule_sched,64    State="ENABLED",65)66log = logging.getLogger(__name__)67def _has_required_boto():68    """69    Returns True/False boolean depending on if Boto is installed and correct70    version.71    """72    if not HAS_BOTO:73        return False74    else:75        return True76class BotoCloudWatchEventStateTestCaseBase(TestCase, LoaderModuleMockMixin):77    conn = None78    def setup_loader_modules(self):79        ctx = {}80        utils = salt.loader.utils(81            self.opts,82            whitelist=["boto3", "args", "systemd", "path", "platform"],83            context=ctx,84        )85        serializers = salt.loader.serializers(self.opts)86        self.funcs = funcs = salt.loader.minion_mods(87            self.opts, context=ctx, utils=utils, whitelist=["boto_cloudwatch_event"]88        )89        self.salt_states = salt.loader.states(90            opts=self.opts,91            functions=funcs,92            utils=utils,93            whitelist=["boto_cloudwatch_event"],94            serializers=serializers,95        )96        return {97            boto_cloudwatch_event: {98                "__opts__": self.opts,99                "__salt__": funcs,100                "__utils__": utils,101                "__states__": self.salt_states,102                "__serializers__": serializers,103            }104        }105    @classmethod106    def setUpClass(cls):107        cls.opts = salt.config.DEFAULT_MINION_OPTS.copy()108        cls.opts["grains"] = salt.loader.grains(cls.opts)109    @classmethod110    def tearDownClass(cls):111        del cls.opts112    def setUp(self):113        self.addCleanup(delattr, self, "funcs")114        self.addCleanup(delattr, self, "salt_states")115        # Set up MagicMock to replace the boto3 session116        # connections keep getting cached from prior tests, can't find the117        # correct context object to clear it. So randomize the cache key, to prevent any118        # cache hits119        conn_parameters["key"] = "".join(120            random.choice(string.ascii_lowercase + string.digits) for _ in range(50)121        )122        self.patcher = patch("boto3.session.Session")123        self.addCleanup(self.patcher.stop)124        self.addCleanup(delattr, self, "patcher")125        mock_session = self.patcher.start()126        session_instance = mock_session.return_value127        self.conn = MagicMock()128        self.addCleanup(delattr, self, "conn")129        session_instance.client.return_value = self.conn130@skipIf(HAS_BOTO is False, "The boto module must be installed.")131class BotoCloudWatchEventTestCase(132    BotoCloudWatchEventStateTestCaseBase, BotoCloudWatchEventTestCaseMixin133):134    def test_present_when_failing_to_describe_rule(self):135        """136        Tests exceptions when checking rule existence137        """138        self.conn.list_rules.side_effect = ClientError(139            error_content, "error on list rules"140        )141        result = self.salt_states["boto_cloudwatch_event.present"](142            name="test present",143            Name=rule_name,144            Description=rule_desc,145            ScheduleExpression=rule_sched,146            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],147            **conn_parameters148        )149        self.assertEqual(result.get("result"), False)150        self.assertTrue("error on list rules" in result.get("comment", {}))151    def test_present_when_failing_to_create_a_new_rule(self):152        """153        Tests present on a rule name that doesn't exist and154        an error is thrown on creation.155        """156        self.conn.list_rules.return_value = {"Rules": []}157        self.conn.put_rule.side_effect = ClientError(error_content, "put_rule")158        result = self.salt_states["boto_cloudwatch_event.present"](159            name="test present",160            Name=rule_name,161            Description=rule_desc,162            ScheduleExpression=rule_sched,163            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],164            **conn_parameters165        )166        self.assertEqual(result.get("result"), False)167        self.assertTrue("put_rule" in result.get("comment", ""))168    def test_present_when_failing_to_describe_the_new_rule(self):169        """170        Tests present on a rule name that doesn't exist and171        an error is thrown when adding targets.172        """173        self.conn.list_rules.return_value = {"Rules": []}174        self.conn.put_rule.return_value = rule_ret175        self.conn.describe_rule.side_effect = ClientError(176            error_content, "describe_rule"177        )178        result = self.salt_states["boto_cloudwatch_event.present"](179            name="test present",180            Name=rule_name,181            Description=rule_desc,182            ScheduleExpression=rule_sched,183            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],184            **conn_parameters185        )186        self.assertEqual(result.get("result"), False)187        self.assertTrue("describe_rule" in result.get("comment", ""))188    def test_present_when_failing_to_create_a_new_rules_targets(self):189        """190        Tests present on a rule name that doesn't exist and191        an error is thrown when adding targets.192        """193        self.conn.list_rules.return_value = {"Rules": []}194        self.conn.put_rule.return_value = rule_ret195        self.conn.describe_rule.return_value = rule_ret196        self.conn.put_targets.side_effect = ClientError(error_content, "put_targets")197        result = self.salt_states["boto_cloudwatch_event.present"](198            name="test present",199            Name=rule_name,200            Description=rule_desc,201            ScheduleExpression=rule_sched,202            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],203            **conn_parameters204        )205        self.assertEqual(result.get("result"), False)206        self.assertTrue("put_targets" in result.get("comment", ""))207    def test_present_when_rule_does_not_exist(self):208        """209        Tests the successful case of creating a new rule, and updating its210        targets211        """212        self.conn.list_rules.return_value = {"Rules": []}213        self.conn.put_rule.return_value = rule_ret214        self.conn.describe_rule.return_value = rule_ret215        self.conn.put_targets.return_value = {"FailedEntryCount": 0}216        result = self.salt_states["boto_cloudwatch_event.present"](217            name="test present",218            Name=rule_name,219            Description=rule_desc,220            ScheduleExpression=rule_sched,221            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],222            **conn_parameters223        )224        self.assertEqual(result.get("result"), True)225    def test_present_when_failing_to_update_an_existing_rule(self):226        """227        Tests present on an existing rule where an error is thrown on updating the pool properties.228        """229        self.conn.list_rules.return_value = {"Rules": [rule_ret]}230        self.conn.describe_rule.side_effect = ClientError(231            error_content, "describe_rule"232        )233        result = self.salt_states["boto_cloudwatch_event.present"](234            name="test present",235            Name=rule_name,236            Description=rule_desc,237            ScheduleExpression=rule_sched,238            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],239            **conn_parameters240        )241        self.assertEqual(result.get("result"), False)242        self.assertTrue("describe_rule" in result.get("comment", ""))243    def test_present_when_failing_to_get_targets(self):244        """245        Tests present on an existing rule where put_rule succeeded, but an error246        is thrown on getting targets247        """248        self.conn.list_rules.return_value = {"Rules": [rule_ret]}249        self.conn.put_rule.return_value = rule_ret250        self.conn.describe_rule.return_value = rule_ret251        self.conn.list_targets_by_rule.side_effect = ClientError(252            error_content, "list_targets"253        )254        result = self.salt_states["boto_cloudwatch_event.present"](255            name="test present",256            Name=rule_name,257            Description=rule_desc,258            ScheduleExpression=rule_sched,259            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],260            **conn_parameters261        )262        self.assertEqual(result.get("result"), False)263        self.assertTrue("list_targets" in result.get("comment", ""))264    def test_present_when_failing_to_put_targets(self):265        """266        Tests present on an existing rule where put_rule succeeded, but an error267        is thrown on putting targets268        """269        self.conn.list_rules.return_value = {"Rules": []}270        self.conn.put_rule.return_value = rule_ret271        self.conn.describe_rule.return_value = rule_ret272        self.conn.list_targets.return_value = {"Targets": []}273        self.conn.put_targets.side_effect = ClientError(error_content, "put_targets")274        result = self.salt_states["boto_cloudwatch_event.present"](275            name="test present",276            Name=rule_name,277            Description=rule_desc,278            ScheduleExpression=rule_sched,279            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],280            **conn_parameters281        )282        self.assertEqual(result.get("result"), False)283        self.assertTrue("put_targets" in result.get("comment", ""))284    def test_present_when_putting_targets(self):285        """286        Tests present on an existing rule where put_rule succeeded, and targets287        must be added288        """289        self.conn.list_rules.return_value = {"Rules": []}290        self.conn.put_rule.return_value = rule_ret291        self.conn.describe_rule.return_value = rule_ret292        self.conn.list_targets.return_value = {"Targets": []}293        self.conn.put_targets.return_value = {"FailedEntryCount": 0}294        result = self.salt_states["boto_cloudwatch_event.present"](295            name="test present",296            Name=rule_name,297            Description=rule_desc,298            ScheduleExpression=rule_sched,299            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],300            **conn_parameters301        )302        self.assertEqual(result.get("result"), True)303    def test_present_when_removing_targets(self):304        """305        Tests present on an existing rule where put_rule succeeded, and targets306        must be removed307        """308        self.conn.list_rules.return_value = {"Rules": []}309        self.conn.put_rule.return_value = rule_ret310        self.conn.describe_rule.return_value = rule_ret311        self.conn.list_targets.return_value = {312            "Targets": [{"Id": "target1"}, {"Id": "target2"}]313        }314        self.conn.put_targets.return_value = {"FailedEntryCount": 0}315        result = self.salt_states["boto_cloudwatch_event.present"](316            name="test present",317            Name=rule_name,318            Description=rule_desc,319            ScheduleExpression=rule_sched,320            Targets=[{"Id": "target1", "Arn": "arn::::::*"}],321            **conn_parameters322        )323        self.assertEqual(result.get("result"), True)324    def test_absent_when_failing_to_describe_rule(self):325        """326        Tests exceptions when checking rule existence327        """328        self.conn.list_rules.side_effect = ClientError(329            error_content, "error on list rules"330        )331        result = self.salt_states["boto_cloudwatch_event.absent"](332            name="test present", Name=rule_name, **conn_parameters333        )334        self.assertEqual(result.get("result"), False)335        self.assertTrue("error on list rules" in result.get("comment", {}))336    def test_absent_when_rule_does_not_exist(self):337        """338        Tests absent on an non-existing rule339        """340        self.conn.list_rules.return_value = {"Rules": []}341        result = self.salt_states["boto_cloudwatch_event.absent"](342            name="test absent", Name=rule_name, **conn_parameters343        )344        self.assertEqual(result.get("result"), True)345        self.assertEqual(result["changes"], {})346    def test_absent_when_failing_to_list_targets(self):347        """348        Tests absent on an rule when the list_targets call fails349        """350        self.conn.list_rules.return_value = {"Rules": [rule_ret]}351        self.conn.list_targets_by_rule.side_effect = ClientError(352            error_content, "list_targets"353        )354        result = self.salt_states["boto_cloudwatch_event.absent"](355            name="test absent", Name=rule_name, **conn_parameters356        )357        self.assertEqual(result.get("result"), False)358        self.assertTrue("list_targets" in result.get("comment", ""))359    def test_absent_when_failing_to_remove_targets_exception(self):360        """361        Tests absent on an rule when the remove_targets call fails362        """363        self.conn.list_rules.return_value = {"Rules": [rule_ret]}364        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}365        self.conn.remove_targets.side_effect = ClientError(366            error_content, "remove_targets"367        )368        result = self.salt_states["boto_cloudwatch_event.absent"](369            name="test absent", Name=rule_name, **conn_parameters370        )371        self.assertEqual(result.get("result"), False)372        self.assertTrue("remove_targets" in result.get("comment", ""))373    def test_absent_when_failing_to_remove_targets_nonexception(self):374        """375        Tests absent on an rule when the remove_targets call fails376        """377        self.conn.list_rules.return_value = {"Rules": [rule_ret]}378        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}379        self.conn.remove_targets.return_value = {"FailedEntryCount": 1}380        result = self.salt_states["boto_cloudwatch_event.absent"](381            name="test absent", Name=rule_name, **conn_parameters382        )383        self.assertEqual(result.get("result"), False)384    def test_absent_when_failing_to_delete_rule(self):385        """386        Tests absent on an rule when the delete_rule call fails387        """388        self.conn.list_rules.return_value = {"Rules": [rule_ret]}389        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}390        self.conn.remove_targets.return_value = {"FailedEntryCount": 0}391        self.conn.delete_rule.side_effect = ClientError(error_content, "delete_rule")392        result = self.salt_states["boto_cloudwatch_event.absent"](393            name="test absent", Name=rule_name, **conn_parameters394        )395        self.assertEqual(result.get("result"), False)396        self.assertTrue("delete_rule" in result.get("comment", ""))397    def test_absent(self):398        """399        Tests absent on an rule400        """401        self.conn.list_rules.return_value = {"Rules": [rule_ret]}402        self.conn.list_targets_by_rule.return_value = {"Targets": [{"Id": "target1"}]}403        self.conn.remove_targets.return_value = {"FailedEntryCount": 0}404        result = self.salt_states["boto_cloudwatch_event.absent"](405            name="test absent", Name=rule_name, **conn_parameters406        )...

put_rule.py

Source:put_rule.py

...18# Create CloudWatchEvents client19cloudwatch_events = boto3.client('events')2021# Put an event rule22response = cloudwatch_events.put_rule(23    Name='DEMO_EVENT',24    ScheduleExpression='rate(5 minutes)',25    State='ENABLED'26)27print(response['RuleArn'])28 29 30# snippet-end:[cloudwatch.python.put_rule.complete]31# snippet-comment:[These are tags for the AWS doc team's sample catalog. Do not remove.]32# snippet-sourcedescription:[put_rule.py demonstrates how to create or update the specified rule. Rules are enabled by default, or based on value of the state. You can disable a rule using DisableRule.]33# snippet-keyword:[Python]34# snippet-keyword:[AWS SDK for Python (Boto3)]35# snippet-keyword:[Code Sample]36# snippet-keyword:[Amazon Cloudwatch]
...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.