Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _list_dead_letter_source_queues method in localstack

Best Python code snippet using localstack_python

sqs_listener.py

Source:sqs_listener.py

...122    if headers.get('Host'):123        url = '%s://%s' % (get_service_protocol(), headers['Host'])124    queue_url = '%s%s' % (url, path.partition('?')[0])125    return queue_url126def _list_dead_letter_source_queues(queues, queue_url):127    dead_letter_source_queues = []128    for k, v in queues.items():129        for i, j in v.items():130            if i == 'RedrivePolicy':131                f = json.loads(v[i])132                queue_url_split = queue_url.split('/')133                if queue_url_split[-1] in f['deadLetterTargetArn']:134                    dead_letter_source_queues.append(k)135    return format_list_dl_source_queues_response(dead_letter_source_queues)136def _process_sent_message(path, req_data, headers):137    queue_name = _queue_url(path, req_data, headers).rpartition('/')[2]138    lambda_api.process_sqs_message(queue_name)139def format_list_dl_source_queues_response(queues):140    content_str = """<ListDeadLetterSourceQueuesResponse xmlns="{}">141                        <ListDeadLetterSourceQueuesResult>142                        {}143                        </ListDeadLetterSourceQueuesResult>144                    </ListDeadLetterSourceQueuesResponse>"""145    queue_urls = ''146    for q in queues:147        queue_urls += '<QueueUrl>{}</QueueUrl>'.format(q)148    return content_str.format(XMLNS_SQS, queue_urls)149# extract the external port used by the client to make the request150def get_external_port(headers, request_handler):151    host = headers.get('Host', '')152    if not host:153        forwarded = headers.get('X-Forwarded-For', '').split(',')154        host = forwarded[-2] if len(forwarded) > 2 else forwarded[-1]155    if ':' in host:156        return int(host.split(':')[1])157    if not request_handler or not request_handler.proxy:158        return config.PORT_SQS159    # If we cannot find the Host header, then fall back to the port of the proxy.160    # (note that this could be incorrect, e.g., if running in Docker with a host port that161    # is different from the internal container port, but there is not much else we can do.)162    return request_handler.proxy.port163def validate_empty_message_batch(data, req_data):164    data = to_str(data).split('Entries=')165    if len(data) > 1 and not req_data.get('Entries'):166        return True167    return False168def is_sqs_queue_url(url):169    path = path_from_url(url).partition('?')[0]170    return re.match(r'^/(queue|%s)/[a-zA-Z0-9_-]+$' % constants.TEST_AWS_ACCOUNT_ID, path)171class ProxyListenerSQS(PersistingProxyListener):172    def api_name(self):173        return 'sqs'174    def forward_request(self, method, path, data, headers):175        if method == 'OPTIONS':176            return 200177        req_data = parse_request_data(method, path, data)178        if is_sqs_queue_url(path) and method == 'GET':179            if not headers.get('Authorization'):180                headers['Authorization'] = aws_stack.mock_aws_request_headers(service='sqs')['Authorization']181            method = 'POST'182            req_data = {'Action': 'GetQueueUrl', 'Version': API_VERSION, 'QueueName': path.split('/')[-1]}183        if req_data:184            action = req_data.get('Action')185            if action in ('SendMessage', 'SendMessageBatch') and SQS_BACKEND_IMPL == 'moto':186                # check message contents187                for key, value in req_data.items():188                    if not re.match(MSG_CONTENT_REGEX, str(value)):189                        return make_requests_error(code=400, code_string='InvalidMessageContents',190                            message='Message contains invalid characters')191            elif action == 'SetQueueAttributes':192                # TODO remove this function if we stop using ElasticMQ entirely193                queue_url = _queue_url(path, req_data, headers)194                if SQS_BACKEND_IMPL == 'elasticmq':195                    forward_attrs = _set_queue_attributes(queue_url, req_data)196                    if len(req_data) != len(forward_attrs):197                        # make sure we only forward the supported attributes to the backend198                        return _get_attributes_forward_request(method, path, headers, req_data, forward_attrs)199            elif action == 'DeleteQueue':200                queue_url = _queue_url(path, req_data, headers)201                QUEUE_ATTRIBUTES.pop(queue_url, None)202                sns_listener.unsubscribe_sqs_queue(queue_url)203            elif action == 'ListDeadLetterSourceQueues':204                # TODO remove this function if we stop using ElasticMQ entirely205                queue_url = _queue_url(path, req_data, headers)206                if SQS_BACKEND_IMPL == 'elasticmq':207                    headers = {'content-type': 'application/xhtml+xml'}208                    content_str = _list_dead_letter_source_queues(QUEUE_ATTRIBUTES, queue_url)209                    return requests_response(content_str, headers=headers)210            if 'QueueName' in req_data:211                encoded_data = urlencode(req_data, doseq=True) if method == 'POST' else ''212                modified_url = None213                if method == 'GET':214                    base_path = path.partition('?')[0]215                    modified_url = '%s?%s' % (base_path, urlencode(req_data, doseq=True))216                return Request(data=encoded_data, url=modified_url, headers=headers, method=method)217        return True218    def return_response(self, method, path, data, headers, response, request_handler):219        # persist requests to disk220        super(ProxyListenerSQS, self).return_response(221            method, path, data, headers, response, request_handler222        )...

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.