KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use describe_stream_consumer method in localstack
Best Python code snippet using localstack_python
test_kinesis.py
Source:test_kinesis.py 
...41 assert consumer_name == consumers[0]["ConsumerName"]42 assert "/%s" % consumer_name in consumer_arn43 assert isinstance(consumers[0]["ConsumerCreationTimestamp"], datetime)44 # lookup stream consumer by describe calls, assert response45 consumer_description_by_arn = kinesis_client.describe_stream_consumer(46 StreamARN=stream_arn, ConsumerARN=consumer_arn47 )["ConsumerDescription"]48 assert consumer_name == consumer_description_by_arn["ConsumerName"]49 assert consumer_arn == consumer_description_by_arn["ConsumerARN"]50 assert stream_arn == consumer_description_by_arn["StreamARN"]51 assert "ACTIVE", consumer_description_by_arn["ConsumerStatus"]52 assert isinstance(consumer_description_by_arn["ConsumerCreationTimestamp"], datetime)53 consumer_description_by_name = kinesis_client.describe_stream_consumer(54 StreamARN=stream_arn, ConsumerName=consumer_name55 )["ConsumerDescription"]56 assert consumer_description_by_arn == consumer_description_by_name57 # delete existing consumer and assert 0 remaining consumers58 kinesis_client.deregister_stream_consumer(StreamARN=stream_arn, ConsumerName=consumer_name)59 retry(assert_consumers, count=0, retries=6, sleep=3.0)60 def test_subscribe_to_shard(61 self, kinesis_client, kinesis_create_stream, wait_for_stream_ready, wait_for_consumer_ready62 ):63 stream_name = "test-%s" % short_uid()64 # create stream and consumer65 kinesis_create_stream(StreamName=stream_name, ShardCount=1)66 stream_arn = kinesis_client.describe_stream(StreamName=stream_name)["StreamDescription"][67 "StreamARN"68 ]69 wait_for_stream_ready(stream_name)70 result = kinesis_client.register_stream_consumer(StreamARN=stream_arn, ConsumerName="c1")[71 "Consumer"72 ]73 consumer_arn = result["ConsumerARN"]74 wait_for_consumer_ready(consumer_arn=consumer_arn)75 # subscribe to shard76 response = kinesis_client.describe_stream(StreamName=stream_name)77 shard_id = response.get("StreamDescription").get("Shards")[0].get("ShardId")78 result = kinesis_client.subscribe_to_shard(79 ConsumerARN=result["ConsumerARN"],80 ShardId=shard_id,81 StartingPosition={"Type": "TRIM_HORIZON"},82 )83 stream = result["EventStream"]84 # put records85 num_records = 586 msg = b"Hello world"87 for i in range(num_records):88 kinesis_client.put_records(89 StreamName=stream_name, Records=[{"Data": msg, "PartitionKey": "1"}]90 )91 # assert results92 results = []93 for entry in stream:94 records = entry["SubscribeToShardEvent"]["Records"]95 continuation_sequence_number = entry["SubscribeToShardEvent"][96 "ContinuationSequenceNumber"97 ]98 # https://docs.aws.amazon.com/kinesis/latest/APIReference/API_SubscribeToShardEvent.html99 assert re.fullmatch("^0|([1-9][0-9]{0,128})$", continuation_sequence_number)100 results.extend(records)101 if len(results) >= num_records:102 break103 # assert results104 assert num_records == len(results)105 for record in results:106 assert msg == record["Data"]107 # clean up108 kinesis_client.deregister_stream_consumer(StreamARN=stream_arn, ConsumerName="c1")109 def test_subscribe_to_shard_with_sequence_number_as_iterator(110 self, kinesis_client, kinesis_create_stream, wait_for_stream_ready, wait_for_consumer_ready111 ):112 stream_name = "test-%s" % short_uid()113 record_data = "Hello world"114 # create stream and consumer115 kinesis_create_stream(StreamName=stream_name, ShardCount=1)116 stream_arn = kinesis_client.describe_stream(StreamName=stream_name)["StreamDescription"][117 "StreamARN"118 ]119 wait_for_stream_ready(stream_name)120 result = kinesis_client.register_stream_consumer(StreamARN=stream_arn, ConsumerName="c1")[121 "Consumer"122 ]123 consumer_arn = result["ConsumerARN"]124 wait_for_consumer_ready(consumer_arn=consumer_arn)125 # get starting sequence number126 response = kinesis_client.describe_stream(StreamName=stream_name)127 sequence_number = (128 response.get("StreamDescription")129 .get("Shards")[0]130 .get("SequenceNumberRange")131 .get("StartingSequenceNumber")132 )133 # subscribe to shard with iterator type as AT_SEQUENCE_NUMBER134 response = kinesis_client.describe_stream(StreamName=stream_name)135 shard_id = response.get("StreamDescription").get("Shards")[0].get("ShardId")136 result = kinesis_client.subscribe_to_shard(137 ConsumerARN=result["ConsumerARN"],138 ShardId=shard_id,139 StartingPosition={140 "Type": "AT_SEQUENCE_NUMBER",141 "SequenceNumber": sequence_number,142 },143 )144 stream = result["EventStream"]145 # put records146 num_records = 5147 for i in range(num_records):148 kinesis_client.put_records(149 StreamName=stream_name,150 Records=[{"Data": record_data, "PartitionKey": "1"}],151 )152 results = []153 for entry in stream:154 records = entry["SubscribeToShardEvent"]["Records"]155 results.extend(records)156 if len(results) >= num_records:157 break158 # assert results159 assert num_records == len(results)160 for record in results:161 assert str.encode(record_data) == record["Data"]162 # clean up163 kinesis_client.deregister_stream_consumer(StreamARN=stream_arn, ConsumerName="c1")164 def test_get_records(self, kinesis_client, kinesis_create_stream, wait_for_stream_ready):165 stream_name = "test-%s" % short_uid()166 kinesis_create_stream(StreamName=stream_name, ShardCount=1)167 wait_for_stream_ready(stream_name)168 kinesis_client.put_records(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 )229 return response.get("ShardIterator")230@pytest.fixture231def wait_for_consumer_ready(kinesis_client):232 def _wait_for_consumer_ready(consumer_arn: str):233 def is_consumer_ready():234 describe_response = kinesis_client.describe_stream_consumer(ConsumerARN=consumer_arn)235 return describe_response["ConsumerDescription"]["ConsumerStatus"] == "ACTIVE"236 poll_condition(is_consumer_ready)237 return _wait_for_consumer_ready238def test_get_records_next_shard_iterator(239 kinesis_client, kinesis_create_stream, wait_for_stream_ready240):241 stream_name = kinesis_create_stream()242 wait_for_stream_ready(stream_name)243 first_stream_shard_data = kinesis_client.describe_stream(StreamName=stream_name)[244 "StreamDescription"245 ]["Shards"][0]246 shard_id = first_stream_shard_data["ShardId"]247 shard_iterator = kinesis_client.get_shard_iterator(248 StreamName=stream_name, ShardIteratorType="LATEST", ShardId=shard_id...
kinesis_source.py
Source:kinesis_source.py
...53 )54 status = "DELETING"55 while status == "DELETING": 56 try:57 response = self.client.describe_stream_consumer(58 StreamARN=self.stream_arn,59 ConsumerName=consumer_name,60 ConsumerARN=consumer_arn61 )62 status = response["ConsumerDescription"]["ConsumerStatus"]63 except Exception as e:64 print(e)65 break66 print(f"Deregistered {consumer_name}")67 # create stream consumer68 response = self.client.register_stream_consumer(69 StreamARN=self.stream_arn,70 ConsumerName=self.name71 )72 consumer_name = response["Consumer"]["ConsumerName"] 73 consumer_arn = response["Consumer"]["ConsumerARN"]74 status = None75 while status != "ACTIVE": 76 response = self.client.describe_stream_consumer(77 StreamARN=self.stream_arn,78 ConsumerName=consumer_name,79 ConsumerARN=consumer_arn80 )81 status = response["ConsumerDescription"]["ConsumerStatus"]82 time.sleep(2)83 print("Finished creating stream consumer")84 # mark shards85 shards = sorted(shards, key=lambda d: d["ShardId"])86 for i in range(len(shards)):87 if i % self.num_replicas == self.shard_idx: 88 self.shards.append({89 "shard_id": shards[i]["ShardId"],90 "seq_no": shards[i]['SequenceNumberRange']['StartingSequenceNumber']...
consume_stream_data.py
Source:consume_stream_data.py
...16 remove_empty_string(entry)17 return dic18def consumer_exists(stream_arn, consumer_name):19 try:20 response = client.describe_stream_consumer(21 StreamARN=stream_arn,22 ConsumerName=consumer_name23 )24 return True25 except client.exceptions.ResourceNotFoundException:26 return False27def register_consumer(stream_arn, consumer_name, force=False):28 # check if exists29 if consumer_exists(stream_arn, consumer_name):30 print('Consumer exists')31 if force:32 deregister_consumer(stream_arn, consumer_name)33 else:34 return35 # register consumer36 response = client.register_stream_consumer(37 StreamARN=stream_arn,38 ConsumerName=consumer_name39 )40 consumer_arn = response['Consumer']['ConsumerARN']41 consumer_status = response['Consumer']['ConsumerStatus']42 # wait until consumer active43 while consumer_status != 'ACTIVE':44 time.sleep(5)45 response = client.describe_stream_consumer(46 StreamARN=stream_arn,47 ConsumerName=consumer_name48 )49 consumer_status = response['ConsumerDescription']['ConsumerStatus']50 return consumer_arn51def deregister_consumer(stream_arn, consumer_name):52 response = client.deregister_stream_consumer(53 StreamARN=stream_arn,54 ConsumerName=consumer_name55 )56 while consumer_exists(stream_arn, consumer_name):57 time.sleep(5)58def fetch_records(stream_name, stream_arn, shard_id, consumer_name, consumer_arn):59 # get dynamodb table to be inserted60 table = get_dynamodb_table('SensorNetworkData')61 response = client.subscribe_to_shard(62 ConsumerARN=consumer_arn,63 ShardId=shard_id,64 StartingPosition={'Type': 'TRIM_HORIZON'}65 )66 shard_iterator = client.get_shard_iterator(StreamName=stream_name,67 ShardId=shard_id,68 ShardIteratorType='TRIM_HORIZON')['ShardIterator']69 counter = 070 while(True and counter < 10000):71 response = client.get_records(ShardIterator=shard_iterator)72 shard_iterator = response['NextShardIterator']73 if len(response['Records']) > 0:74 data = json.loads(response['Records'][0]['Data'], parse_float=Decimal)75 print('Processing sensor: %s', data['sensor_uuid'])76 data_fix = remove_empty_string(data)77 try:78 '''79 table.put_item(80 Item=data_fix81 )82 '''83 print(data_fix)84 counter += 185 except:86 print('ERROR : ')87 else:88 print('Waiting..')89 time.sleep(5)90if __name__ == '__main__':91 stream_name = 'DataStream'92 consumer_name = 'StreamConsumer3'93 # setup dynamodb94 dynamodb = boto3.resource('dynamodb')95 table = dynamodb.Table('DataStream')96 # setup kinesis97 client = boto3.client('kinesis')98 # get stream details99 response = client.describe_stream(StreamName=stream_name)100 stream_arn = response['StreamDescription']['StreamARN']101 shard_id = response['StreamDescription']['Shards'][2]['ShardId']102 # get consumer details103 response = client.describe_stream_consumer(104 StreamARN=stream_arn,105 ConsumerName=consumer_name106 )107 consumer_arn = response['ConsumerDescription']['ConsumerARN']108 # subscribe to shard109 response = client.subscribe_to_shard(110 ConsumerARN=consumer_arn,111 ShardId=shard_id,112 StartingPosition={'Type': 'TRIM_HORIZON'}113 )114 # iterate event stream115 for event in response['EventStream']:116 print(event)117 if len(event['SubscribeToShardEvent']['Records']) > 0:...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
