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

How to use _get_shard_iterator method in localstack

Best Python code snippet using localstack_python

test_kinesis.py

Source:test_kinesis.py

...169            StreamName=stream_name,170            Records=[{"Data": "SGVsbG8gd29ybGQ=", "PartitionKey": "1"}],171        )172        # get records with JSON encoding173        iterator = self._get_shard_iterator(stream_name, kinesis_client)174        response = kinesis_client.get_records(ShardIterator=iterator)175        json_records = response.get("Records")176        assert 1 == len(json_records)177        assert "Data" in json_records[0]178        # get records with CBOR encoding179        iterator = self._get_shard_iterator(stream_name, kinesis_client)180        url = config.get_edge_url()181        headers = aws_stack.mock_aws_request_headers("kinesis")182        headers["Content-Type"] = constants.APPLICATION_AMZ_CBOR_1_1183        headers["X-Amz-Target"] = "Kinesis_20131202.GetRecords"184        data = cbor2.dumps({"ShardIterator": iterator})185        result = requests.post(url, data, headers=headers)186        assert 200 == result.status_code187        result = cbor2.loads(result.content)188        attrs = ("Data", "EncryptionType", "PartitionKey", "SequenceNumber")189        assert select_attributes(json_records[0], attrs) == select_attributes(190            result["Records"][0], attrs191        )192    def test_record_lifecycle_data_integrity(193        self, kinesis_client, kinesis_create_stream, wait_for_stream_ready194    ):195        """196        kinesis records should contain the same data from when they are sent to when they are received197        """198        stream_name = "test-%s" % short_uid()199        records_data = {"test", "Ã¼nicÃ¶dÃ© ç»ä¸ç  ð£ð»ð¥", "a" * 1000, ""}200        kinesis_create_stream(StreamName=stream_name, ShardCount=1)201        wait_for_stream_ready(stream_name)202        iterator = self._get_shard_iterator(stream_name, kinesis_client)203        for record_data in records_data:204            kinesis_client.put_record(205                StreamName=stream_name,206                Data=record_data,207                PartitionKey="1",208            )209        response = kinesis_client.get_records(ShardIterator=iterator)210        response_records = response.get("Records")211        assert len(records_data) == len(response_records)212        for response_record in response_records:213            assert response_record.get("Data").decode("utf-8") in records_data214    def _get_shard_iterator(self, stream_name, kinesis_client):215        response = kinesis_client.describe_stream(StreamName=stream_name)216        sequence_number = (217            response.get("StreamDescription")218            .get("Shards")[0]219            .get("SequenceNumberRange")220            .get("StartingSequenceNumber")221        )222        shard_id = response.get("StreamDescription").get("Shards")[0].get("ShardId")223        response = kinesis_client.get_shard_iterator(224            StreamName=stream_name,225            ShardId=shard_id,226            ShardIteratorType="AT_SEQUENCE_NUMBER",227            StartingSequenceNumber=sequence_number,228        )...

test_kinesis_logs_reader.py

Source:test_kinesis_logs_reader.py

...42        'dstaddr': '198.51.100.1',43        'dstport': 443,44        'protocol': 6,45    }46def _get_shard_iterator(**kwargs):47    return {'ShardIterator': '{}_iterator-0001'.format(kwargs['ShardId'])}48DESCRIBE_STREAM = {49    'StreamDescription': {50        'Shards': [{'ShardId': 'shard-0001'}, {'ShardId': 'shard-0002'}],51    }52}53GET_RECORDS = {54    'shard-0001_iterator-0001': {55        'Records': [56            _control_message(),57            _data_message([_create_event(0), _create_event(1)]),58        ],59        'NextShardIterator': 'shard-0001_iterator-0002',60        'MillisBehindLatest': 100,61    },62    'shard-0001_iterator-0002': {63        'Records': [_data_message([_create_event(2), _create_event(3)])],64        'NextShardIterator': 'shard-0001_iterator-0003',65        'MillisBehindLatest': 0,66    },67    'shard-0002_iterator-0001': {68        'Records': [_data_message([_create_event(4), _create_event(5)])],69        'NextShardIterator': 'shard-0002_iterator-0002',70        'MillisBehindLatest': 0,71    },72    'shard-0002_iterator-0002': {73        'Records': [],74        'NextShardIterator': 'shard-0002_iterator-0003',75        'MillisBehindLatest': 0,76    },77}78def _get_client(*args, **kwargs):79    # Mock the boto3 client so tests don't hit the AWS API80    mock_client = MagicMock()81    mock_client.get_paginator.return_value.paginate.return_value = (82        [DESCRIBE_STREAM]83    )84    mock_client.get_shard_iterator.side_effect = _get_shard_iterator85    mock_client.get_records.side_effect = (86        lambda **kwargs: GET_RECORDS[kwargs['ShardIterator']]87    )88    return mock_client89class UtilsTestCase(TestCase):90    def setUp(self):91        self.data = b'Test data'92        self.gz_data = (93            b'\x1f\x8b\x08\x00M\x986W\x02\xff\x0bI-.QHI,I\x04\x00\x11,\xf9Q\t'94            b'\x00\x00\x00'95        )96    def test_gunzip_bytes(self):97        self.assertEqual(gunzip_bytes(self.gz_data), self.data)98    def test_gzip_bytes(self):99        gz_data = gzip_bytes(self.data)100        self.assertEqual(gunzip_bytes(gz_data), self.data)101class KinesisLogsReaderTestCase(TestCase):102    def __init__(self, *args, **kwargs):103        # Python 2 compatibility for tests104        if not hasattr(self, 'assertCountEqual'):105            self.assertCountEqual = self.assertItemsEqual106        return super(KinesisLogsReaderTestCase, self).__init__(*args, **kwargs)107    def setUp(self):108        self.stream_name = 'test-stream'109        self.start_time = datetime(2016, 5, 13, 22, 55, 0)110        self.reader = KinesisLogsReader(111            self.stream_name, kinesis_client=_get_client()112        )113    def test_init(self):114        kwargs = {'region_name': 'test-region', 'profile_name': 'test_profile'}115        patch_path = 'kinesis_logs_reader.kinesis_logs_reader.Session'116        with patch(patch_path) as mock_Session:117            KinesisLogsReader(self.stream_name, **kwargs)118            mock_Session.assert_called_once_with(**kwargs)119            mock_Session.return_value.client.assert_called_once_with('kinesis')120    def test_get_shard_ids(self):121        actual = list(self.reader._get_shard_ids())122        expected = ['shard-0001', 'shard-0002']123        self.assertCountEqual(actual, expected)124    def test_get_shard_iterators(self):125        shard_id = 'shard-0001'126        actual = self.reader._get_shard_iterator(shard_id)127        self.assertEqual(actual, 'shard-0001_iterator-0001')128        self.reader.kinesis_client.get_shard_iterator.assert_called_with(129            StreamName=self.stream_name,130            ShardId=shard_id,131            ShardIteratorType='LATEST',132        )133        shard_id = 'shard-0002'134        actual = self.reader._get_shard_iterator(shard_id, self.start_time)135        self.assertEqual(actual, 'shard-0002_iterator-0001')136        self.reader.kinesis_client.get_shard_iterator.assert_called_with(137            StreamName=self.stream_name,138            ShardId=shard_id,139            ShardIteratorType='AT_TIMESTAMP',140            Timestamp=self.start_time,141        )142    def test_read_shard(self):143        shard_id = 'shard-0001'144        self.reader.shard_iterators[shard_id] = 'shard-0001_iterator-0001'145        # Data messages should have been extracted and returned146        actual_results = list(self.reader._read_shard(shard_id))147        expected_results = [_create_event(0), _create_event(1)]148        self.assertEqual(actual_results, expected_results)...

reader.py

Source:reader.py

...17        :return: A generator that yields messages18        :rtype: generator19        """20        shard_id = self._get_shard_id(stream_name)21        iterator = self._get_shard_iterator(22            stream_name,23            shard_id,24            seq_number25        )26        while True:27            response = self.client.get_records(28                ShardIterator=iterator,29                Limit=10030            )31            records = response['Records']32            for record in records:33                yield record34            time.sleep(self.read_interval)35            iterator = response['NextShardIterator']36    def _get_shard_id(self, stream_name):37        """Get the shard ID for a given stream name.38        :param stream_name:39        :return: Shard id of Kinesis stream40        :rtype: str41        """42        response = self.client.describe_stream(43            StreamName=stream_name44        )45        return response['StreamDescription']['Shards'][0]['ShardId']46    def _get_shard_iterator(self, stream_name, shard_id, seq_number):47        """Get the initial shard iterator.48        :param stream_name: Name of Kinesis stream49        :param shard_id:  Shard id50        :param seq_number: Sequence number of message51        :return: Shard iterator52        :rtype: str53        """54        if not seq_number:55            response = self.client.get_shard_iterator(56                StreamName=stream_name,57                ShardId=shard_id,58                ShardIteratorType='TRIM_HORIZON'59            )60        else:...

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.