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

How to use receive_dlq method in localstack

Best Python code snippet using localstack_python

test_sns.py

Source:test_sns.py

...351        payload = {352            lambda_integration.MSG_BODY_RAISE_ERROR_FLAG: 1,353        }354        self.sns_client.publish(TopicArn=topic_arn, Message=json.dumps(payload))355        def receive_dlq():356            result = self.sqs_client.receive_message(QueueUrl=queue_url, MessageAttributeNames=['All'])357            msg_attrs = result['Messages'][0]['MessageAttributes']358            self.assertGreater(len(result['Messages']), 0)359            self.assertIn('RequestID', msg_attrs)360            self.assertIn('ErrorCode', msg_attrs)361            self.assertIn('ErrorMessage', msg_attrs)362        retry(receive_dlq, retries=8, sleep=2)363    def unsubscribe_all_from_sns(self):364        for subscription_arn in self.sns_client.list_subscriptions()['Subscriptions']:365            self.sns_client.unsubscribe(SubscriptionArn=subscription_arn['SubscriptionArn'])366    def test_redrive_policy_http_subscription(self):367        self.unsubscribe_all_from_sns()368        # create HTTP endpoint and connect it to SNS topic369        class MyUpdateListener(ProxyListener):370            def forward_request(self, method, path, data, headers):371                records.append((json.loads(to_str(data)), headers))372                return 200373        records = []374        local_port = get_free_tcp_port()375        proxy = start_proxy(local_port, backend_url=None, update_listener=MyUpdateListener())376        wait_for_port_open(local_port)377        http_endpoint = '%s://localhost:%s' % (get_service_protocol(), local_port)378        subscription = self.sns_client.subscribe(TopicArn=self.topic_arn,379            Protocol='http', Endpoint=http_endpoint)380        self.sns_client.set_subscription_attributes(381            SubscriptionArn=subscription['SubscriptionArn'],382            AttributeName='RedrivePolicy',383            AttributeValue=json.dumps({'deadLetterTargetArn': aws_stack.sqs_queue_arn(TEST_QUEUE_DLQ_NAME)})384        )385        proxy.stop()386        # for some reason, it takes a long time to stop the proxy thread -> TODO investigate387        time.sleep(5)388        self.sns_client.publish(TopicArn=self.topic_arn, Message=json.dumps({'message': 'test_redrive_policy'}))389        def receive_dlq():390            result = self.sqs_client.receive_message(QueueUrl=self.dlq_url, MessageAttributeNames=['All'])391            self.assertGreater(len(result['Messages']), 0)392            self.assertEqual(393                json.loads(json.loads(result['Messages'][0]['Body'])['Message'][0])['message'],394                'test_redrive_policy'395            )396        retry(receive_dlq, retries=7, sleep=2.5)397    def test_redrive_policy_lambda_subscription(self):398        self.unsubscribe_all_from_sns()399        lambda_name = 'test-%s' % short_uid()400        lambda_arn = aws_stack.lambda_function_arn(lambda_name)401        testutil.create_lambda_function(func_name=lambda_name, libs=TEST_LAMBDA_LIBS,402            handler_file=TEST_LAMBDA_PYTHON, runtime=LAMBDA_RUNTIME_PYTHON36)403        subscription = self.sns_client.subscribe(TopicArn=self.topic_arn, Protocol='lambda', Endpoint=lambda_arn)404        self.sns_client.set_subscription_attributes(405            SubscriptionArn=subscription['SubscriptionArn'],406            AttributeName='RedrivePolicy',407            AttributeValue=json.dumps({'deadLetterTargetArn': aws_stack.sqs_queue_arn(TEST_QUEUE_DLQ_NAME)})408        )409        testutil.delete_lambda_function(lambda_name)410        self.sns_client.publish(TopicArn=self.topic_arn, Message=json.dumps({'message': 'test_redrive_policy'}))411        def receive_dlq():412            result = self.sqs_client.receive_message(QueueUrl=self.dlq_url, MessageAttributeNames=['All'])413            self.assertGreater(len(result['Messages']), 0)414            self.assertEqual(415                json.loads(json.loads(result['Messages'][0]['Body'])['Message'][0])['message'],416                'test_redrive_policy'417            )418        retry(receive_dlq, retries=10, sleep=2)419    def test_redrive_policy_queue_subscription(self):420        self.unsubscribe_all_from_sns()421        topic_arn = self.sns_client.create_topic(Name='topic-%s' % short_uid())['TopicArn']422        invalid_queue_arn = aws_stack.sqs_queue_arn('invalid_queue')423        # subscribe with an invalid queue ARN, to trigger event on DLQ below424        subscription = self.sns_client.subscribe(TopicArn=topic_arn, Protocol='sqs', Endpoint=invalid_queue_arn)425        self.sns_client.set_subscription_attributes(426            SubscriptionArn=subscription['SubscriptionArn'],427            AttributeName='RedrivePolicy',428            AttributeValue=json.dumps({'deadLetterTargetArn': aws_stack.sqs_queue_arn(TEST_QUEUE_DLQ_NAME)})429        )430        self.sns_client.publish(TopicArn=topic_arn, Message=json.dumps({'message': 'test_redrive_policy'}))431        def receive_dlq():432            result = self.sqs_client.receive_message(QueueUrl=self.dlq_url, MessageAttributeNames=['All'])433            self.assertGreater(len(result['Messages']), 0)434            self.assertEqual(435                json.loads(json.loads(result['Messages'][0]['Body'])['Message'][0])['message'],436                'test_redrive_policy'437            )438        retry(receive_dlq, retries=10, sleep=2)439    def test_publish_with_empty_subject(self):440        topic_arn = self.sns_client.create_topic(Name=TEST_TOPIC_NAME_2)['TopicArn']441        # Publish without subject442        rs = self.sns_client.publish(443            TopicArn=topic_arn,444            Message=json.dumps({'message': 'test_publish'})445        )...

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.