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

How to use list_stream_consumers method in localstack

Best Python code snippet using localstack_python

test_kinesis_stream_consumers.py

Source:test_kinesis_stream_consumers.py

...8    client.create_stream(StreamName=stream_name, ShardCount=4)9    stream = client.describe_stream(StreamName=stream_name)["StreamDescription"]10    return stream["StreamARN"]11@mock_kinesis12def test_list_stream_consumers():13    client = boto3.client("kinesis", region_name="eu-west-1")14    stream_arn = create_stream(client)15    resp = client.list_stream_consumers(StreamARN=stream_arn)16    resp.should.have.key("Consumers").equals([])17@mock_kinesis18def test_register_stream_consumer():19    client = boto3.client("kinesis", region_name="eu-west-1")20    stream_arn = create_stream(client)21    resp = client.register_stream_consumer(22        StreamARN=stream_arn, ConsumerName="newconsumer"23    )24    resp.should.have.key("Consumer")25    consumer = resp["Consumer"]26    consumer.should.have.key("ConsumerName").equals("newconsumer")27    consumer.should.have.key("ConsumerARN").equals(28        f"arn:aws:kinesis:eu-west-1:{ACCOUNT_ID}:stream/my-stream/consumer/newconsumer"29    )30    consumer.should.have.key("ConsumerStatus").equals("ACTIVE")31    consumer.should.have.key("ConsumerCreationTimestamp")32    resp = client.list_stream_consumers(StreamARN=stream_arn)33    resp.should.have.key("Consumers").length_of(1)34    consumer = resp["Consumers"][0]35    consumer.should.have.key("ConsumerName").equals("newconsumer")36    consumer.should.have.key("ConsumerARN").equals(37        f"arn:aws:kinesis:eu-west-1:{ACCOUNT_ID}:stream/my-stream/consumer/newconsumer"38    )39    consumer.should.have.key("ConsumerStatus").equals("ACTIVE")40    consumer.should.have.key("ConsumerCreationTimestamp")41@mock_kinesis42def test_describe_stream_consumer_by_name():43    client = boto3.client("kinesis", region_name="us-east-2")44    stream_arn = create_stream(client)45    client.register_stream_consumer(StreamARN=stream_arn, ConsumerName="newconsumer")46    resp = client.describe_stream_consumer(47        StreamARN=stream_arn, ConsumerName="newconsumer"48    )49    resp.should.have.key("ConsumerDescription")50    consumer = resp["ConsumerDescription"]51    consumer.should.have.key("ConsumerName").equals("newconsumer")52    consumer.should.have.key("ConsumerARN")53    consumer.should.have.key("ConsumerStatus").equals("ACTIVE")54    consumer.should.have.key("ConsumerCreationTimestamp")55    consumer.should.have.key("StreamARN").equals(stream_arn)56@mock_kinesis57def test_describe_stream_consumer_by_arn():58    client = boto3.client("kinesis", region_name="us-east-2")59    stream_arn = create_stream(client)60    resp = client.register_stream_consumer(61        StreamARN=stream_arn, ConsumerName="newconsumer"62    )63    consumer_arn = resp["Consumer"]["ConsumerARN"]64    resp = client.describe_stream_consumer(ConsumerARN=consumer_arn)65    resp.should.have.key("ConsumerDescription")66    consumer = resp["ConsumerDescription"]67    consumer.should.have.key("ConsumerName").equals("newconsumer")68    consumer.should.have.key("ConsumerARN")69    consumer.should.have.key("ConsumerStatus").equals("ACTIVE")70    consumer.should.have.key("ConsumerCreationTimestamp")71    consumer.should.have.key("StreamARN").equals(stream_arn)72@mock_kinesis73def test_describe_stream_consumer_unknown():74    client = boto3.client("kinesis", region_name="us-east-2")75    create_stream(client)76    with pytest.raises(ClientError) as exc:77        client.describe_stream_consumer(ConsumerARN="unknown")78    err = exc.value.response["Error"]79    err["Code"].should.equal("ResourceNotFoundException")80    err["Message"].should.equal(f"Consumer unknown, account {ACCOUNT_ID} not found.")81@mock_kinesis82def test_deregister_stream_consumer_by_name():83    client = boto3.client("kinesis", region_name="ap-southeast-1")84    stream_arn = create_stream(client)85    client.register_stream_consumer(StreamARN=stream_arn, ConsumerName="consumer1")86    client.register_stream_consumer(StreamARN=stream_arn, ConsumerName="consumer2")87    client.list_stream_consumers(StreamARN=stream_arn)[88        "Consumers"89    ].should.have.length_of(2)90    client.deregister_stream_consumer(StreamARN=stream_arn, ConsumerName="consumer1")91    client.list_stream_consumers(StreamARN=stream_arn)[92        "Consumers"93    ].should.have.length_of(1)94@mock_kinesis95def test_deregister_stream_consumer_by_arn():96    client = boto3.client("kinesis", region_name="ap-southeast-1")97    stream_arn = create_stream(client)98    resp = client.register_stream_consumer(99        StreamARN=stream_arn, ConsumerName="consumer1"100    )101    consumer1_arn = resp["Consumer"]["ConsumerARN"]102    client.register_stream_consumer(StreamARN=stream_arn, ConsumerName="consumer2")103    client.list_stream_consumers(StreamARN=stream_arn)[104        "Consumers"105    ].should.have.length_of(2)106    client.deregister_stream_consumer(ConsumerARN=consumer1_arn)107    client.list_stream_consumers(StreamARN=stream_arn)[108        "Consumers"...

aws_kinesis_info.py

Source:aws_kinesis_info.py

...99                return paginator.paginate(100                    StreamARN=module.params['arn'],101                ), True102            else:103                return client.list_stream_consumers(104                    StreamARN=module.params['arn'],105                ), False106        elif module.params['list_streams']:107            if client.can_paginate('list_streams'):108                paginator = client.get_paginator('list_streams')109                return paginator.paginate(), True110            else:111                return client.list_streams(), False112        else:113            return None, False114    except (BotoCoreError, ClientError) as e:115        module.fail_json_aws(e, msg='Failed to fetch Amazon Kinesis details')116def main():117    argument_spec = dict(...

helper_functions.py

Source:helper_functions.py

...27    :param conn: Connection we'll use to get the list28    :param name: The name of the stream29    :return: A list of consumers attached to the provided stream30    """31    return dumps(conn.list_stream_consumers(StreamARN=name), sort_keys=True, indent=2)32def renew_shard_iterator(conn, name, id, type):33    return conn.get_shard_iterator(StreamName=name, ShardId=id, ShardIteratorType=type)[...

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.