KaneAI - World’s First E2E Software Testing Agent.Signup for Private Beta Join TestMu Conference

How to use get_subscription_by_arn method in localstack

Best Python code snippet using localstack_python

sns_listener.py

Source:sns_listener.py

...21                topic_arn = topic_arn[0]22                if topic_arn not in SNS_SUBSCRIPTIONS:23                    SNS_SUBSCRIPTIONS[topic_arn] = []24            if req_action == 'SetSubscriptionAttributes':25                sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])26                if not sub:27                    return make_error(message='Unable to find subscription for given ARN', code=400)28                attr_name = req_data['AttributeName'][0]29                attr_value = req_data['AttributeValue'][0]30                sub[attr_name] = attr_value31                return make_response(req_action)32            elif req_action == 'GetSubscriptionAttributes':33                sub = get_subscription_by_arn(req_data['SubscriptionArn'][0])34                if not sub:35                    return make_error(message='Unable to find subscription for given ARN', code=400)36                content = '<Attributes>'37                for key, value in sub.items():38                    content += '<entry><key>%s</key><value>%s</value></entry>\n' % (key, value)39                content += '</Attributes>'40                return make_response(req_action, content=content)41            elif req_action == 'Subscribe':42                if 'Endpoint' not in req_data:43                    return make_error(message='Endpoint not specified in subscription', code=400)44            elif req_action == 'Publish':45                message = req_data['Message'][0]46                sqs_client = aws_stack.connect_to_service('sqs')47                for subscriber in SNS_SUBSCRIPTIONS[topic_arn]:48                    if subscriber['Protocol'] == 'sqs':49                        queue_name = subscriber['Endpoint'].split(':')[5]50                        queue_url = subscriber.get('sqs_queue_url')51                        if not queue_url:52                            queue_url = aws_stack.get_sqs_queue_url(queue_name)53                            subscriber['sqs_queue_url'] = queue_url54                        sqs_client.send_message(QueueUrl=queue_url,55                            MessageBody=create_sns_message_body(subscriber, req_data))56                    elif subscriber['Protocol'] == 'lambda':57                        lambda_api.process_sns_notification(58                            subscriber['Endpoint'],59                            topic_arn, message, subject=req_data.get('Subject')60                        )61                    elif subscriber['Protocol'] == 'http':62                        requests.post(63                            subscriber['Endpoint'],64                            headers={65                                'Content-Type': 'text/plain',66                                'x-amz-sns-message-type': 'Notification'67                            },68                            data=json.dumps({69                                'Type': 'Notification',70                                'Message': message,71                            })72                        )73                    else:74                        LOGGER.warning('Unexpected protocol "%s" for SNS subscription' % subscriber['Protocol'])75                # return response here because we do not want the request to be forwarded to SNS76                return make_response(req_action)77        return True78    else:79        # This branch is executed by the proxy after we've already received a80        # response from the backend, hence we can utilize the "reponse" variable here81        if method == 'POST' and path == '/':82            req_data = urlparse.parse_qs(data)83            req_action = req_data['Action'][0]84            if req_action == 'Subscribe' and response.status_code < 400:85                response_data = xmltodict.parse(response.content)86                topic_arn = (req_data.get('TargetArn') or req_data.get('TopicArn'))[0]87                sub_arn = response_data['SubscribeResponse']['SubscribeResult']['SubscriptionArn']88                subscription = {89                    # http://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html90                    'TopicArn': topic_arn,91                    'Endpoint': req_data['Endpoint'][0],92                    'Protocol': req_data['Protocol'][0],93                    'SubscriptionArn': sub_arn,94                    'RawMessageDelivery': 'false'95                }96                SNS_SUBSCRIPTIONS[topic_arn].append(subscription)97# ---------------98# HELPER METHODS99# ---------------100def get_subscription_by_arn(sub_arn):101    # TODO maintain separate map instead of traversing all items102    for key, subscriptions in SNS_SUBSCRIPTIONS.items():103        for sub in subscriptions:104            if sub['SubscriptionArn'] == sub_arn:105                return sub106def make_response(op_name, content=''):107    response = Response()108    if not content:109        content = '<MessageId>%s</MessageId>' % short_uid()110    response._content = """<{op_name}Response xmlns="http://sns.amazonaws.com/doc/2010-03-31/">111        <{op_name}Result>112            {content}113        </{op_name}Result>114        <ResponseMetadata><RequestId>{req_id}</RequestId></ResponseMetadata>...

test_sns_listener.py

Source:test_sns_listener.py

...15    sns_listener.do_subscribe(topic_arn,16                     'http://localhost:1234/listen',17                     'http',18                     sub_arn)19    assert(sns_listener.get_subscription_by_arn(sub_arn) is not None)20    sns_listener.do_unsubscribe(sub_arn)21    assert(sns_listener.get_subscription_by_arn(sub_arn) is None)22def test_create_sns_message_body_raw_message_delivery():23    subscriber = {24        'RawMessageDelivery': 'true'25    }26    action = {27        'Message': ['msg']28    }29    result = sns_listener.create_sns_message_body(subscriber, action)30    assert (result == 'msg')31def test_create_sns_message_body():32    subscriber = {33        'TopicArn': 'arn',34        'RawMessageDelivery': 'false',35    }...

34136_test_sns_listener.py

Source:34136_test_sns_listener.py

...15    sns_listener.do_subscribe(topic_arn,16                     'http://localhost:1234/listen',17                     'http',18                     sub_arn)19    assert(sns_listener.get_subscription_by_arn(sub_arn) is not None)20    sns_listener.do_unsubscribe(sub_arn)21    assert(sns_listener.get_subscription_by_arn(sub_arn) is None)22def test_create_sns_message_body_raw_message_delivery():23    subscriber = {24        'RawMessageDelivery': 'true'25    }26    action = {27        'Message': ['msg']28    }29    result = sns_listener.create_sns_message_body(subscriber, action)30    assert (result == 'msg')31def test_create_sns_message_body():32    subscriber = {33        'TopicArn': 'arn',34        'RawMessageDelivery': 'false',35    }...

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.