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

How to use describe_stream_summary method in localstack

Best Python code snippet using localstack_python

test_kinesis_client.py

Source:test_kinesis_client.py

1import unittest2import json3from unittest.mock import patch, Mock4from libs.kinesis_client import KinesisClient5from botocore.exceptions import ClientError6def get_mocked_boto_kinesis_put_record_error():7    mocked_client = Mock()8    mocked_client.put_record.side_effect = Mock(side_effect=ClientError({}, 'test'))9    return mocked_client10def get_mocked_boto_kinesis_active():11    mocked_client = Mock()12    mocked_client.describe_stream_summary.return_value = {13        'StreamDescriptionSummary': {14            'StreamStatus': 'ACTIVE'15        }16    }17    return mocked_client18def get_mocked_boto_kinesis_deleting():19    mocked_client = Mock()20    mocked_client.describe_stream_summary.return_value = {21        'StreamDescriptionSummary': {22            'StreamStatus': 'DELETING'23        }24    }25    return mocked_client26def get_mocked_boto_kinesis_error():27    mocked_client = Mock()28    mocked_client.describe_stream_summary.side_effect = Mock(side_effect=ClientError({}, 'test'))29    return mocked_client30class KinesisClientTestCase(unittest.TestCase):31    @patch( 'boto3.client')32    @patch( 'logging.getLogger')33    def test_put_records_into_stream_will_log_record(self, boto3, logger):34        kinesis_client = KinesisClient(boto3, 'stream_name', logger)35        kinesis_client.put_records_into_stream([{'key': 'value'}], 'key')36        logger.info.assert_called_with('Put record in stream stream_name.')37        boto3.put_record.assert_called_with(38            StreamName='stream_name',39            Data=json.dumps({'key': 'value'}),40            PartitionKey='key'41        )42    43    @patch( 'logging.getLogger')44    def test_put_records_into_stream_will_log_error(self, logger):45        boto3 = get_mocked_boto_kinesis_put_record_error()46        kinesis_client = KinesisClient(boto3, 'stream_name', logger)47        kinesis_client.put_records_into_stream([{'key': 'value'}], 'key')48        self.assertTrue(logger.error.called)49    @patch( 'logging.getLogger')50    def test_wait_for_active_stream_will_return_true(self, logger):51        boto3 = get_mocked_boto_kinesis_active()52        kinesis_client = KinesisClient(boto3, 'stream_name', logger)53        self.assertTrue(kinesis_client.wait_for_active_stream())54    @patch( 'logging.getLogger')55    def test_wait_for_active_stream_will_return_false_on_deleted_stream(self, logger):56        boto3 = get_mocked_boto_kinesis_deleting()57        kinesis_client = KinesisClient(boto3, 'stream_name', logger)58        self.assertFalse(kinesis_client.wait_for_active_stream())59    @patch( 'logging.getLogger')60    def test_wait_for_active_stream_will_return_false_on_error(self, logger):61        boto3 = get_mocked_boto_kinesis_error()62        kinesis_client = KinesisClient(boto3, 'stream_name', logger)63        self.assertFalse(kinesis_client.wait_for_active_stream())...

stream_util.py

Source:stream_util.py

...28    if full_form:29        headers = LIST_STREAM_FULL_HEADERS30    table_data = [headers]31    for stream_name in stream_names:32        describe_stream_summary_response = kinesis_client.describe_stream_summary(33            StreamName=stream_name34        )35        describe_stream_summary = describe_stream_summary_response[36            "StreamDescriptionSummary"37        ]38        stream_status = describe_stream_summary["StreamStatus"]39        table_data.append(40            [41                constants.STREAM_STATUS_ICON_MAP[stream_status] + " " + stream_name,42                describe_stream_summary["OpenShardCount"],43                describe_stream_summary["ConsumerCount"],44                # describe_stream_summary['StreamARN'],45                describe_stream_summary["RetentionPeriodHours"],46                # describe_stream_summary['StreamCreationTimestamp'],...

list_streams.py

Source:list_streams.py

...11    streams = []12    if 'StreamNames' in result:13        for st in result['StreamNames']:14            record = {}15            details = client.describe_stream_summary(16                StreamName=st)['StreamDescriptionSummary']17            record['StreamName'] = st18            record['StreamARN'] = details['StreamARN']19            record['OpenShardCount'] = details['OpenShardCount']20            shards_info = client.list_shards(StreamName=st)21            record['shards'] = shards_info['Shards']22            streams.append(record)23    return streams24def parse_known_args():25    # AWS credentials should be provided as environ variables26    if 'AWS_ACCESS_KEY_ID' not in os.environ:27        print('Error. Please setup AWS_ACCESS_KEY_ID')28        exit(1)29    elif 'AWS_SECRET_ACCESS_KEY' not in os.environ:...

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.