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

How to use put_delivery_channel method in localstack

Best Python code snippet using localstack_python

test_config.py

Source:test_config.py

...603def test_delivery_channels():604    client = boto3.client('config', region_name='us-west-2')605    # Try without a config recorder:606    with assert_raises(ClientError) as ce:607        client.put_delivery_channel(DeliveryChannel={})608    assert ce.exception.response['Error']['Code'] == 'NoAvailableConfigurationRecorderException'609    assert ce.exception.response['Error']['Message'] == 'Configuration recorder is not available to ' \610                                                        'put delivery channel.'611    # Create a config recorder to continue testing:612    client.put_configuration_recorder(ConfigurationRecorder={613        'name': 'testrecorder',614        'roleARN': 'somearn',615        'recordingGroup': {616            'allSupported': False,617            'includeGlobalResourceTypes': False,618            'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']619        }620    })621    # Try without a name supplied:622    with assert_raises(ClientError) as ce:623        client.put_delivery_channel(DeliveryChannel={})624    assert ce.exception.response['Error']['Code'] == 'InvalidDeliveryChannelNameException'625    assert 'is not valid, blank string.' in ce.exception.response['Error']['Message']626    # Try with a really long name:627    with assert_raises(ClientError) as ce:628        client.put_delivery_channel(DeliveryChannel={'name': 'a' * 257})629    assert ce.exception.response['Error']['Code'] == 'ValidationException'630    assert 'Member must have length less than or equal to 256' in ce.exception.response['Error']['Message']631    # Without specifying a bucket name:632    with assert_raises(ClientError) as ce:633        client.put_delivery_channel(DeliveryChannel={'name': 'testchannel'})634    assert ce.exception.response['Error']['Code'] == 'NoSuchBucketException'635    assert ce.exception.response['Error']['Message'] == 'Cannot find a S3 bucket with an empty bucket name.'636    with assert_raises(ClientError) as ce:637        client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': ''})638    assert ce.exception.response['Error']['Code'] == 'NoSuchBucketException'639    assert ce.exception.response['Error']['Message'] == 'Cannot find a S3 bucket with an empty bucket name.'640    # With an empty string for the S3 key prefix:641    with assert_raises(ClientError) as ce:642        client.put_delivery_channel(DeliveryChannel={643            'name': 'testchannel', 's3BucketName': 'somebucket', 's3KeyPrefix': ''})644    assert ce.exception.response['Error']['Code'] == 'InvalidS3KeyPrefixException'645    assert 'empty s3 key prefix.' in ce.exception.response['Error']['Message']646    # With an empty string for the SNS ARN:647    with assert_raises(ClientError) as ce:648        client.put_delivery_channel(DeliveryChannel={649            'name': 'testchannel', 's3BucketName': 'somebucket', 'snsTopicARN': ''})650    assert ce.exception.response['Error']['Code'] == 'InvalidSNSTopicARNException'651    assert 'The sns topic arn' in ce.exception.response['Error']['Message']652    # With an invalid delivery frequency:653    with assert_raises(ClientError) as ce:654        client.put_delivery_channel(DeliveryChannel={655            'name': 'testchannel',656            's3BucketName': 'somebucket',657            'configSnapshotDeliveryProperties': {'deliveryFrequency': 'WRONG'}658        })659    assert ce.exception.response['Error']['Code'] == 'InvalidDeliveryFrequency'660    assert 'WRONG' in ce.exception.response['Error']['Message']661    assert 'TwentyFour_Hours' in ce.exception.response['Error']['Message']662    # Create a proper one:663    client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})664    result = client.describe_delivery_channels()['DeliveryChannels']665    assert len(result) == 1666    assert len(result[0].keys()) == 2667    assert result[0]['name'] == 'testchannel'668    assert result[0]['s3BucketName'] == 'somebucket'669    # Overwrite it with another proper configuration:670    client.put_delivery_channel(DeliveryChannel={671        'name': 'testchannel',672        's3BucketName': 'somebucket',673        'snsTopicARN': 'sometopicarn',674        'configSnapshotDeliveryProperties': {'deliveryFrequency': 'TwentyFour_Hours'}675    })676    result = client.describe_delivery_channels()['DeliveryChannels']677    assert len(result) == 1678    assert len(result[0].keys()) == 4679    assert result[0]['name'] == 'testchannel'680    assert result[0]['s3BucketName'] == 'somebucket'681    assert result[0]['snsTopicARN'] == 'sometopicarn'682    assert result[0]['configSnapshotDeliveryProperties']['deliveryFrequency'] == 'TwentyFour_Hours'683    # Can only have 1:684    with assert_raises(ClientError) as ce:685        client.put_delivery_channel(DeliveryChannel={'name': 'testchannel2', 's3BucketName': 'somebucket'})686    assert ce.exception.response['Error']['Code'] == 'MaxNumberOfDeliveryChannelsExceededException'687    assert 'because the maximum number of delivery channels: 1 is reached.' in ce.exception.response['Error']['Message']688@mock_config689def test_describe_delivery_channels():690    client = boto3.client('config', region_name='us-west-2')691    client.put_configuration_recorder(ConfigurationRecorder={692        'name': 'testrecorder',693        'roleARN': 'somearn',694        'recordingGroup': {695            'allSupported': False,696            'includeGlobalResourceTypes': False,697            'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']698        }699    })700    # Without any channels:701    result = client.describe_delivery_channels()702    assert not result['DeliveryChannels']703    client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})704    result = client.describe_delivery_channels()['DeliveryChannels']705    assert len(result) == 1706    assert len(result[0].keys()) == 2707    assert result[0]['name'] == 'testchannel'708    assert result[0]['s3BucketName'] == 'somebucket'709    # Overwrite it with another proper configuration:710    client.put_delivery_channel(DeliveryChannel={711        'name': 'testchannel',712        's3BucketName': 'somebucket',713        'snsTopicARN': 'sometopicarn',714        'configSnapshotDeliveryProperties': {'deliveryFrequency': 'TwentyFour_Hours'}715    })716    result = client.describe_delivery_channels()['DeliveryChannels']717    assert len(result) == 1718    assert len(result[0].keys()) == 4719    assert result[0]['name'] == 'testchannel'720    assert result[0]['s3BucketName'] == 'somebucket'721    assert result[0]['snsTopicARN'] == 'sometopicarn'722    assert result[0]['configSnapshotDeliveryProperties']['deliveryFrequency'] == 'TwentyFour_Hours'723    # Specify an incorrect name:724    with assert_raises(ClientError) as ce:725        client.describe_delivery_channels(DeliveryChannelNames=['wrong'])726    assert ce.exception.response['Error']['Code'] == 'NoSuchDeliveryChannelException'727    assert 'wrong' in ce.exception.response['Error']['Message']728    # And with both a good and wrong name:729    with assert_raises(ClientError) as ce:730        client.describe_delivery_channels(DeliveryChannelNames=['testchannel', 'wrong'])731    assert ce.exception.response['Error']['Code'] == 'NoSuchDeliveryChannelException'732    assert 'wrong' in ce.exception.response['Error']['Message']733@mock_config734def test_start_configuration_recorder():735    client = boto3.client('config', region_name='us-west-2')736    # Without a config recorder:737    with assert_raises(ClientError) as ce:738        client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')739    assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'740    # Make the config recorder;741    client.put_configuration_recorder(ConfigurationRecorder={742        'name': 'testrecorder',743        'roleARN': 'somearn',744        'recordingGroup': {745            'allSupported': False,746            'includeGlobalResourceTypes': False,747            'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']748        }749    })750    # Without a delivery channel:751    with assert_raises(ClientError) as ce:752        client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')753    assert ce.exception.response['Error']['Code'] == 'NoAvailableDeliveryChannelException'754    # Make the delivery channel:755    client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})756    # Start it:757    client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')758    # Verify it's enabled:759    result = client.describe_configuration_recorder_status()['ConfigurationRecordersStatus']760    lower_bound = (datetime.utcnow() - timedelta(minutes=5))761    assert result[0]['recording']762    assert result[0]['lastStatus'] == 'PENDING'763    assert lower_bound < result[0]['lastStartTime'].replace(tzinfo=None) <= datetime.utcnow()764    assert lower_bound < result[0]['lastStatusChangeTime'].replace(tzinfo=None) <= datetime.utcnow()765@mock_config766def test_stop_configuration_recorder():767    client = boto3.client('config', region_name='us-west-2')768    # Without a config recorder:769    with assert_raises(ClientError) as ce:770        client.stop_configuration_recorder(ConfigurationRecorderName='testrecorder')771    assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'772    # Make the config recorder;773    client.put_configuration_recorder(ConfigurationRecorder={774        'name': 'testrecorder',775        'roleARN': 'somearn',776        'recordingGroup': {777            'allSupported': False,778            'includeGlobalResourceTypes': False,779            'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']780        }781    })782    # Make the delivery channel for creation:783    client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})784    # Start it:785    client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')786    client.stop_configuration_recorder(ConfigurationRecorderName='testrecorder')787    # Verify it's disabled:788    result = client.describe_configuration_recorder_status()['ConfigurationRecordersStatus']789    lower_bound = (datetime.utcnow() - timedelta(minutes=5))790    assert not result[0]['recording']791    assert result[0]['lastStatus'] == 'PENDING'792    assert lower_bound < result[0]['lastStartTime'].replace(tzinfo=None) <= datetime.utcnow()793    assert lower_bound < result[0]['lastStopTime'].replace(tzinfo=None) <= datetime.utcnow()794    assert lower_bound < result[0]['lastStatusChangeTime'].replace(tzinfo=None) <= datetime.utcnow()795@mock_config796def test_describe_configuration_recorder_status():797    client = boto3.client('config', region_name='us-west-2')798    # Without any:799    result = client.describe_configuration_recorder_status()800    assert not result['ConfigurationRecordersStatus']801    # Make the config recorder;802    client.put_configuration_recorder(ConfigurationRecorder={803        'name': 'testrecorder',804        'roleARN': 'somearn',805        'recordingGroup': {806            'allSupported': False,807            'includeGlobalResourceTypes': False,808            'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']809        }810    })811    # Without specifying a config recorder:812    result = client.describe_configuration_recorder_status()['ConfigurationRecordersStatus']813    assert len(result) == 1814    assert result[0]['name'] == 'testrecorder'815    assert not result[0]['recording']816    # With a proper name:817    result = client.describe_configuration_recorder_status(818        ConfigurationRecorderNames=['testrecorder'])['ConfigurationRecordersStatus']819    assert len(result) == 1820    assert result[0]['name'] == 'testrecorder'821    assert not result[0]['recording']822    # Invalid name:823    with assert_raises(ClientError) as ce:824        client.describe_configuration_recorder_status(ConfigurationRecorderNames=['testrecorder', 'wrong'])825    assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'826    assert 'wrong' in ce.exception.response['Error']['Message']827@mock_config828def test_delete_configuration_recorder():829    client = boto3.client('config', region_name='us-west-2')830    # Make the config recorder;831    client.put_configuration_recorder(ConfigurationRecorder={832        'name': 'testrecorder',833        'roleARN': 'somearn',834        'recordingGroup': {835            'allSupported': False,836            'includeGlobalResourceTypes': False,837            'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']838        }839    })840    # Delete it:841    client.delete_configuration_recorder(ConfigurationRecorderName='testrecorder')842    # Try again -- it should be deleted:843    with assert_raises(ClientError) as ce:844        client.delete_configuration_recorder(ConfigurationRecorderName='testrecorder')845    assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationRecorderException'846@mock_config847def test_delete_delivery_channel():848    client = boto3.client('config', region_name='us-west-2')849    # Need a recorder to test the constraint on recording being enabled:850    client.put_configuration_recorder(ConfigurationRecorder={851        'name': 'testrecorder',852        'roleARN': 'somearn',853        'recordingGroup': {854            'allSupported': False,855            'includeGlobalResourceTypes': False,856            'resourceTypes': ['AWS::EC2::Volume', 'AWS::EC2::VPC']857        }858    })859    client.put_delivery_channel(DeliveryChannel={'name': 'testchannel', 's3BucketName': 'somebucket'})860    client.start_configuration_recorder(ConfigurationRecorderName='testrecorder')861    # With the recorder enabled:862    with assert_raises(ClientError) as ce:863        client.delete_delivery_channel(DeliveryChannelName='testchannel')864    assert ce.exception.response['Error']['Code'] == 'LastDeliveryChannelDeleteFailedException'865    assert 'because there is a running configuration recorder.' in ce.exception.response['Error']['Message']866    # Stop recording:867    client.stop_configuration_recorder(ConfigurationRecorderName='testrecorder')868    # Try again:869    client.delete_delivery_channel(DeliveryChannelName='testchannel')870    # Verify:871    with assert_raises(ClientError) as ce:872        client.delete_delivery_channel(DeliveryChannelName='testchannel')873    assert ce.exception.response['Error']['Code'] == 'NoSuchDeliveryChannelException'

aws_config.py

Source:aws_config.py

...42            logger.error(e)43            return44        else:45            return46    def put_delivery_channel(47        self,48        region_name: str,49        bucket_name: str,50        name="default",51    ) -> None:52        """53        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/config.html#ConfigService.Client.put_delivery_channel54        """55        try:56            client: ConfigServiceClient = self.session.client(57                "config",58                region_name=str(region_name),59                config=Config(),60            )61            client.put_delivery_channel(62                DeliveryChannel={63                    "name": str(name),64                    "s3BucketName": str(bucket_name),65                }66            )67        except Exception as e:68            logger.error(e)69            return70        else:71            return72    def start_configuration_recorder(73        self,74        region_name: str,75        config_recorder_name: str,...

responses.py

Source:responses.py

...16        recorder_statuses = self.config_backend.describe_configuration_recorder_status(17            self._get_param('ConfigurationRecorderNames'))18        schema = {'ConfigurationRecordersStatus': recorder_statuses}19        return json.dumps(schema)20    def put_delivery_channel(self):21        self.config_backend.put_delivery_channel(self._get_param('DeliveryChannel'))22        return ""23    def describe_delivery_channels(self):24        delivery_channels = self.config_backend.describe_delivery_channels(self._get_param('DeliveryChannelNames'))25        schema = {'DeliveryChannels': delivery_channels}26        return json.dumps(schema)27    def describe_delivery_channel_status(self):28        raise NotImplementedError()29    def delete_delivery_channel(self):30        self.config_backend.delete_delivery_channel(self._get_param('DeliveryChannelName'))31        return ""32    def delete_configuration_recorder(self):33        self.config_backend.delete_configuration_recorder(self._get_param('ConfigurationRecorderName'))34        return ""35    def start_configuration_recorder(self):...

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.