KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use describe_delivery_stream method in localstack

Best Python code snippet using localstack_python

test_kinesis.py

Source:test_kinesis.py

...78        )79        resources = p.run()80        self.assertEqual(len(resources), 1)81        self.assertEqual(82            factory().client("firehose").describe_delivery_stream(83                DeliveryStreamName="sock-index-hose"84            )[85                "DeliveryStreamDescription"86            ][87                "DeliveryStreamStatus"88            ],89            "DELETING",90        )91    def test_firehose_extended_s3_encrypt_s3_destination(self):92        factory = self.replay_flight_data("test_firehose_ext_s3_encrypt_s3_destination")93        p = self.load_policy(94            {95                "name": "khole",96                "resource": "firehose",97                "filters": [{"type": "value",98                    "key": "Destinations[0].S3DestinationDescription.EncryptionConfiguration.NoEncryptionConfig",  # noqa: E50199                             "value": "present"}],100                "actions": [{"type": "encrypt-s3-destination",101                             "key_arn": "arn:aws:kms:us-east-1:123456789:alias/aws/s3"}],102            },103            session_factory=factory,104        )105        resources = p.run()106        self.assertEqual(len(resources), 1)107        s = factory().client("firehose").describe_delivery_stream(108            DeliveryStreamName="firehose-s3"109        )['DeliveryStreamDescription']['Destinations'][0]110        assert 'KMSEncryptionConfig' in s['S3DestinationDescription']['EncryptionConfiguration'].keys()  # noqa: E501111    def test_firehose_splunk_encrypt_s3_destination(self):112        factory = self.replay_flight_data("test_firehose_splunk_encrypt_s3_destination")113        p = self.load_policy(114            {115                "name": "khole",116                "resource": "firehose",117                "filters": [{"type": "value",118                    "key": "Destinations[0].SplunkDestinationDescription.S3DestinationDescription.EncryptionConfiguration.NoEncryptionConfig",  # noqa: E501119                             "value": "present"}],120                "actions": [{"type": "encrypt-s3-destination",121                             "key_arn": "arn:aws:kms:us-east-1:123456789:alias/aws/s3"}],122            },123            session_factory=factory,124        )125        resources = p.run()126        self.assertEqual(len(resources), 1)127        s = factory().client("firehose").describe_delivery_stream(128            DeliveryStreamName="firehose-splunk"129        )['DeliveryStreamDescription']['Destinations'][0]['SplunkDestinationDescription']130        assert 'KMSEncryptionConfig' in \131            s['S3DestinationDescription']['EncryptionConfiguration'].keys()132    def test_firehose_elasticsearch_encrypt_s3_destination(self):133        factory = self.replay_flight_data("test_firehose_elasticsearch_encrypt_s3_destination")134        p = self.load_policy(135            {136                "name": "khole",137                "resource": "firehose",138                "filters": [{"type": "value",139                    "key": "Destinations[0].ElasticsearchDestinationDescription.S3DestinationDescription.EncryptionConfiguration.NoEncryptionConfig",  # noqa: E501140                             "value": "present"}],141                "actions": [{"type": "encrypt-s3-destination",142                             "key_arn": "arn:aws:kms:us-east-1:123456789:alias/aws/s3"}],143            },144            session_factory=factory,145        )146        resources = p.run()147        self.assertEqual(len(resources), 1)148        s = factory().client("firehose").describe_delivery_stream(149            DeliveryStreamName="firehose-splunk"150        )['DeliveryStreamDescription']['Destinations'][0]['ElasticsearchDestinationDescription']151        assert 'KMSEncryptionConfig' in \152            s['S3DestinationDescription']['EncryptionConfiguration'].keys()153    def test_firehose_redshift_encrypt_s3_destination(self):154        factory = self.replay_flight_data("test_firehose_redshift_encrypt_s3_destination")155        p = self.load_policy(156            {157                "name": "khole",158                "resource": "firehose",159                "filters": [{"type": "value",160                    "key": "Destinations[0].RedshiftDestinationDescription.S3DestinationDescription.EncryptionConfiguration.NoEncryptionConfig",  # noqa: E501161                             "value": "present"}],162                "actions": [{"type": "encrypt-s3-destination",163                             "key_arn": "arn:aws:kms:us-east-1:123456789:alias/aws/s3"}],164            },165            session_factory=factory,166        )167        resources = p.run()168        self.assertEqual(len(resources), 1)169        s = factory().client("firehose").describe_delivery_stream(170            DeliveryStreamName="firehose-redshift"171        )['DeliveryStreamDescription']['Destinations'][0]['RedshiftDestinationDescription']172        assert 'KMSEncryptionConfig' in \173            s['S3DestinationDescription']['EncryptionConfiguration'].keys()174    def test_app_query(self):175        factory = self.replay_flight_data("test_kinesis_analytics_query")176        p = self.load_policy(177            {178                "name": "kapp",179                "resource": "kinesis-analytics",180                "filters": [{"ApplicationStatus": "RUNNING"}],181            },182            session_factory=factory,183        )...

aws_firehose_info.py

Source:aws_firehose_info.py

...58from ansible.module_utils.common.dict_transformations import camel_dict_to_snake_dict59def _firehose(client, module):60    try:61        if module.params['describe_delivery_stream']:62            return client.describe_delivery_stream(63                DeliveryStreamName=module.params['name'],64            ), False65        else:66            if client.can_paginate('list_delivery_streams'):67                paginator = client.get_paginator('list_delivery_streams')68                return paginator.paginate(), True69            else:70                return client.list_delivery_streams(), False71    except (BotoCoreError, ClientError) as e:72        module.fail_json_aws(e, msg='Failed to fetch Amazon Firehose Details')73def main():74    argument_spec = dict(75        name=dict(required=False),76        describe_delivery_stream=dict(required=False, type=bool),...

test_client_aws_kinesisfirehose.py

Source:test_client_aws_kinesisfirehose.py

1import unittest2import keydra.clients.aws.kinesisfirehose as firehose3from keydra.clients.aws.kinesisfirehose import UpdateSecretException4from unittest.mock import MagicMock5SPLUNK_STREAM_DESC = {6    'DeliveryStreamDescription': {7        'VersionId': '11',8        'Destinations': [9            {10                'DestinationId': 'id1',11                'SplunkDestinationDescription': {}12            }13        ]14    }15}16HTTP_STREAM_DESC = {17    'DeliveryStreamDescription': {18        'VersionId': '22',19        'Destinations': [20            {21                'DestinationId': 'id11',22                'HttpEndpointDestinationDescription': {23                    'EndpointConfiguration': {'url': 'test'}24                }25            }26        ]27    }28}29class TestClientFirehose(unittest.TestCase):30    def test__stream_check(self):31        cli = firehose.FirehoseClient(session=MagicMock())32        cli._client.describe_delivery_stream = MagicMock()33        self.assertEqual(cli.stream_exists('woot'), True)34        cli._client.describe_delivery_stream.side_effect = Exception('boom!')35        self.assertEqual(cli.stream_exists('woot'), False)36    def test__get_stream_ids(self):37        cli = firehose.FirehoseClient(session=MagicMock())38        with self.assertRaises(Exception):39            cli._get_stream_ids('woot', 'limewire')40        cli._client.describe_delivery_stream.return_value = SPLUNK_STREAM_DESC41        self.assertEqual(cli._get_stream_ids('woot', 'Splunk'), ('11', 'id1'))42        self.assertEqual(cli._get_stream_ids('woot', 'HttpEndpoint'), ('11', None))43        cli._client.describe_delivery_stream.return_value = HTTP_STREAM_DESC44        self.assertEqual(cli._get_stream_ids('woot', 'HttpEndpoint'), ('22', 'id11'))45    def test__get_HttpEndpointConfiguration(self):46        cli = firehose.FirehoseClient(session=MagicMock())47        cli._client.describe_delivery_stream.return_value = HTTP_STREAM_DESC48        with self.assertRaises(Exception):49            cli._get_HttpEndpointConfiguration('woot', 'id88')50        self.assertEqual(cli._get_HttpEndpointConfiguration('woot', 'id11'), {'url': 'test'})51    def test__update_splunk_hectoken(self):52        cli = firehose.FirehoseClient(session=MagicMock())53        cli._client.describe_delivery_stream.return_value = HTTP_STREAM_DESC54        with self.assertRaises(UpdateSecretException):55            cli.update_splunk_hectoken('woot', 'key')56        cli._client.describe_delivery_stream.return_value = SPLUNK_STREAM_DESC57        cli._client.update_destination.return_value = {'hep'}58        self.assertEqual(cli.update_splunk_hectoken('woot', 'key'), {'hep'})59    def test__update_http_accesskey(self):60        cli = firehose.FirehoseClient(session=MagicMock())61        cli._client.describe_delivery_stream.return_value = SPLUNK_STREAM_DESC62        with self.assertRaises(UpdateSecretException):63            cli.update_http_accesskey('woot', 'key')64        cli._client.describe_delivery_stream.return_value = HTTP_STREAM_DESC65        cli._client.update_destination.return_value = {'hep'}...

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.