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

How to use get_queue_arn method in localstack

Best Python code snippet using localstack_python

sqs.py

Source:sqs.py

...21    )22    return SQSQueue(23        name=QueueName,24        url=resp["QueueUrl"],25        arn=get_queue_arn(resp["QueueUrl"]),26        response=resp,27    )28def create_queues(queues: List[dict]):29    return {q.name: q for q in [create_queue(**config) for config in queues]}30@backoff.on_exception(backoff.expo, ClientError, max_tries=3)31def destroy_queue(QueueUrl, **kwargs):32    return utils.call(33        boto3_fixtures.contrib.boto3.client("sqs").delete_queue, QueueUrl=QueueUrl34    )35def destroy_queues(queues: Dict[str, SQSQueue]):36    return [destroy_queue(QueueUrl=q.url) for _, q in queues.items()]37def setup(queues: Union[List[str], List[dict]], **kwargs):38    queues = [39        {"QueueName": queue, **kwargs} if isinstance(queue, str) else queue...

subscriptions.py

Source:subscriptions.py

...5import aws.kinesis as kinesis6import aws.sns as sns7def subscribe_queue_to_topic(topic_arn: str, queue_url: str):8    client = common.get_client('sns')9    queue_arn = sqs.get_queue_arn(queue_url)10    response = client.list_subscriptions_by_topic(11        TopicArn=topic_arn,12    )13    subscription_exists = any([14        queue_arn == item['Endpoint']15        for item in response['Subscriptions']16    ])17    if subscription_exists:18        common.print_resource(19            'Queue to topic subscription already exists',20            meta_list=[f'topic: {topic_arn}', f'qeue: {queue_arn}']21        )22    else:23        client.subscribe(24            TopicArn=topic_arn,25            Protocol='sqs',26            Endpoint=queue_arn,27        )28        common.print_resource(29            'Added queue subscription on topic',30            meta_list=[f'topic: {topic_arn}', f'qeue: {queue_arn}']31        )32def subscribe_queue_to_bucket(bucket_name: str, queue_url: str, events: list):33    client = common.get_client('s3')34    queue_arn = sqs.get_queue_arn(queue_url)35    client = common.get_client('s3')36    response = client.get_bucket_notification_configuration(37        Bucket=bucket_name38    )39    subscription_exists = any([40        queue_arn == item.get('QueueArn') and item.get('Events') == events41        for item in response.get('QueueConfigurations', [])42    ])43    if subscription_exists:44        common.print_resource(45            'Queue to bucket subscription already exists',46            meta_list=[47                f'bucket: {bucket_name}',48                f'queue: {queue_arn}',...

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.