How to use test_redrive_policy_http_subscription method in localstack

Best Python code snippet using localstack_python

test_sns.py

Source:test_sns.py

...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(...

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.