How to use _format_attributes_names method in localstack

Best Python code snippet using localstack_python

sqs_listener.py

Source:sqs_listener.py

...45        key_name = req_data[key1]46        key_value = req_data.get(key2) or ''47        result[key_name] = key_value48    return result49def _format_attributes_names(req_data):50    result = set()51    for i in range(1, 500):52        key = 'AttributeName.%s' % i53        if key not in req_data:54            break55        result.add(req_data[key])56    return result57def _get_attributes_forward_request(method, path, headers, req_data, forward_attrs):58    req_data_new = dict([(k, v) for k, v in req_data.items() if not k.startswith('Attribute.')])59    i = 160    for k, v in forward_attrs.items():61        req_data_new['Attribute.%s.Name' % i] = [k]62        req_data_new['Attribute.%s.Value' % i] = [v]63        i += 164    data = urlencode(req_data_new, doseq=True)65    return Request(data=data, headers=headers, method=method)66def _set_queue_attributes(queue_url, req_data):67    # TODO remove this function if we stop using ElasticMQ entirely68    if SQS_BACKEND_IMPL != 'elasticmq':69        return70    attrs = _format_attributes(req_data)71    # select only the attributes in UNSUPPORTED_ATTRIBUTE_NAMES72    local_attrs = {}73    for k, v in attrs.items():74        if k in UNSUPPORTED_ATTRIBUTE_NAMES:75            try:76                _v = json.loads(v)77                if isinstance(_v, dict):78                    if 'maxReceiveCount' in _v:79                        _v['maxReceiveCount'] = int(_v['maxReceiveCount'])80                local_attrs.update(dict({k: json.dumps(_v)}))81            except Exception:82                local_attrs.update(dict({k: v}))83    QUEUE_ATTRIBUTES[queue_url] = QUEUE_ATTRIBUTES.get(queue_url) or {}84    QUEUE_ATTRIBUTES[queue_url].update(local_attrs)85    forward_attrs = dict([(k, v) for k, v in attrs.items() if k not in UNSUPPORTED_ATTRIBUTE_NAMES])86    return forward_attrs87def _add_queue_attributes(path, req_data, content_str, headers):88    # TODO remove this function if we stop using ElasticMQ entirely89    if SQS_BACKEND_IMPL != 'elasticmq':90        return content_str91    flags = re.MULTILINE | re.DOTALL92    queue_url = _queue_url(path, req_data, headers)93    requested_attributes = _format_attributes_names(req_data)94    regex = r'(.*<GetQueueAttributesResult>)(.*)(</GetQueueAttributesResult>.*)'95    attrs = re.sub(regex, r'\2', content_str, flags=flags)96    for key, value in QUEUE_ATTRIBUTES.get(queue_url, {}).items():97        if (not requested_attributes or requested_attributes.intersection({'All', key})) and \98                not re.match(r'<Name>\s*%s\s*</Name>' % key, attrs, flags=flags):99            attrs += '<Attribute><Name>%s</Name><Value>%s</Value></Attribute>' % (key, value)100    content_str = (re.sub(regex, r'\1', content_str, flags=flags) +101                   attrs + re.sub(regex, r'\3', content_str, flags=flags))102    return content_str103def _fire_event(req_data, response):104    action = req_data.get('Action')105    event_type = None106    queue_url = None107    if action == 'CreateQueue':...

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.