Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use sns_error_to_dead_letter_queue method in localstack

Best Python code snippet using localstack_python

sns_listener.py

Source:sns_listener.py

...245                    MessageAttributes=create_sqs_message_attributes(subscriber, message_attributes)246                )247            except Exception as exc:248                LOG.warning('Unable to forward SNS message to SQS: %s %s' % (exc, traceback.format_exc()))249                sns_error_to_dead_letter_queue(subscriber['SubscriptionArn'], req_data, str(exc))250                if 'NonExistentQueue' in str(exc):251                    LOG.info('Removing non-existent queue "%s" subscribed to topic "%s"' % (queue_url, topic_arn))252                    subscriptions.remove(subscriber)253        elif subscriber['Protocol'] == 'lambda':254            try:255                external_url = external_service_url('sns')256                unsubscribe_url = '%s/?Action=Unsubscribe&SubscriptionArn=%s' % (external_url,257                                    subscriber['SubscriptionArn'])258                response = lambda_api.process_sns_notification(259                    subscriber['Endpoint'],260                    topic_arn,261                    subscriber['SubscriptionArn'],262                    message,263                    message_id,264                    message_attributes,265                    unsubscribe_url,266                    subject=req_data.get('Subject', [None])[0]267                )268                if isinstance(response, Response):269                    response.raise_for_status()270                elif isinstance(response, FlaskResponse):271                    if response.status_code >= 400:272                        raise Exception('Error response (code %s): %s' % (response.status_code, response.data))273            except Exception as exc:274                LOG.warning('Unable to run Lambda function on SNS message: %s %s' % (exc, traceback.format_exc()))275                sns_error_to_dead_letter_queue(subscriber['SubscriptionArn'], req_data, str(exc))276        elif subscriber['Protocol'] in ['http', 'https']:277            msg_type = (req_data.get('Type') or ['Notification'])[0]278            try:279                message_body = create_sns_message_body(subscriber, req_data, message_id)280            except Exception:281                continue282            try:283                response = requests.post(284                    subscriber['Endpoint'],285                    headers={286                        'Content-Type': 'text/plain',287                        # AWS headers according to288                        # https://docs.aws.amazon.com/sns/latest/dg/sns-message-and-json-formats.html#http-header289                        'x-amz-sns-message-type': msg_type,290                        'x-amz-sns-topic-arn': subscriber['TopicArn'],291                        'x-amz-sns-subscription-arn': subscriber['SubscriptionArn'],292                        'User-Agent': 'Amazon Simple Notification Service Agent'293                    },294                    data=message_body,295                    verify=False296                )297                response.raise_for_status()298            except Exception as exc:299                LOG.info('Received error on sending SNS message, putting to DLQ (if configured): %s' % exc)300                sns_error_to_dead_letter_queue(subscriber['SubscriptionArn'], req_data, str(exc))301        elif subscriber['Protocol'] == 'application':302            try:303                sns_client = aws_stack.connect_to_service('sns')304                sns_client.publish(TargetArn=subscriber['Endpoint'], Message=message)305            except Exception as exc:306                LOG.warning('Unable to forward SNS message to SNS platform app: %s %s' % (exc, traceback.format_exc()))307                sns_error_to_dead_letter_queue(subscriber['SubscriptionArn'], req_data, str(exc))308        else:309            LOG.warning('Unexpected protocol "%s" for SNS subscription' % subscriber['Protocol'])310def publish_message(topic_arn, req_data, subscription_arn=None, skip_checks=False):311    message = req_data['Message'][0]312    message_id = str(uuid.uuid4())313    if topic_arn and ':endpoint/' in topic_arn:314        # cache messages published to platform endpoints315        cache = PLATFORM_ENDPOINT_MESSAGES[topic_arn] = PLATFORM_ENDPOINT_MESSAGES.get(topic_arn) or []316        cache.append(req_data)317    LOG.debug('Publishing message to TopicArn: %s | Message: %s' % (topic_arn, message))318    start_thread(319        lambda _: message_to_subscribers(message_id, message, topic_arn, req_data, subscription_arn, skip_checks))320    return message_id321def do_delete_topic(topic_arn):...

dead_letter_queue.py

Source:dead_letter_queue.py

...17    target_arn = policy.get('deadLetterTargetArn')18    if not target_arn:19        return20    return _send_to_dead_letter_queue('SQS', queue_arn, target_arn, event, error)21def sns_error_to_dead_letter_queue(sns_subscriber_arn, event, error):22    client = aws_stack.connect_to_service('sns')23    attrs = client.get_subscription_attributes(SubscriptionArn=sns_subscriber_arn)24    attrs = attrs.get('Attributes', {})25    policy = json.loads(attrs.get('RedrivePolicy') or '{}')26    target_arn = policy.get('deadLetterTargetArn')27    if not target_arn:28        return29    return _send_to_dead_letter_queue('SQS', sns_subscriber_arn, target_arn, event, error)30def lambda_error_to_dead_letter_queue(func_details, event, error):31    dlq_arn = (func_details.dead_letter_config or {}).get('TargetArn')32    source_arn = func_details.id33    return _send_to_dead_letter_queue('Lambda', source_arn, dlq_arn, event, error)34def _send_to_dead_letter_queue(source_type, source_arn, dlq_arn, event, error):35    if not dlq_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.