KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use extract_region_from_auth_header method in localstack

Best Python code snippet using localstack_python

sqs_listener.py

Source:sqs_listener.py

...33                queue_url = req_data.get('QueueUrl', [path.partition('?')[0]])[0]34                queue_name = queue_url[queue_url.rindex('/') + 1:]35                message_body = req_data.get('MessageBody', [None])[0]36                message_attributes = self.format_message_attributes(req_data)37                region_name = extract_region_from_auth_header(headers)38                process_result = lambda_api.process_sqs_message(message_body,39                    message_attributes, queue_name, region_name=region_name)40                if process_result:41                    # If a Lambda was listening, do not add the message to the queue42                    new_response = Response()43                    new_response._content = SUCCESSFUL_SEND_MESSAGE_XML_TEMPLATE.format(44                        message_attr_hash=md5(data),45                        message_body_hash=md5(message_body),46                        message_id=str(uuid.uuid4())47                    )48                    new_response.status_code = 20049                    return new_response50            if 'QueueName' in req_data:51                encoded_data = urlencode(req_data, doseq=True) if method == 'POST' else ''52                modified_url = None53                if method == 'GET':54                    base_path = path.partition('?')[0]55                    modified_url = '%s?%s' % (base_path, urlencode(req_data, doseq=True))56                request = Request(data=encoded_data, url=modified_url, headers=headers, method=method)57                return request58        return True59    def parse_request_data(self, method, path, data):60        """ Extract request data either from query string (for GET) or request body (for POST). """61        if method == 'POST':62            return urlparse.parse_qs(to_str(data))63        elif method == 'GET':64            parsed_path = urlparse.urlparse(path)65            return urlparse.parse_qs(parsed_path.query)66        return {}67    # Format of the message Name attribute is MessageAttribute.<int id>.<field>68    # Format of the Value attributes is MessageAttribute.<int id>.Value.DataType69    # and MessageAttribute.<int id>.Value.<Type>Value70    #71    # The data schema changes on transfer between SQS and Lambda (at least)72    # JS functions in real AWS!73    # It is unknown at this time whether this data structure change affects different74    # languages in different ways.75    #76    # The MessageAttributes specified in the SQS payload (in JavaScript):77    # var params = {78    #   MessageBody: "body string",79    #   MessageAttributes: {80    #       "attr_1": {81    #           DataType: "String",82    #           StringValue: "attr_1_value"83    #       },84    #       "attr_2": {85    #           DataType: "String",86    #           StringValue: "attr_2_value"87    #       }88    #   }89    # }90    #91    # The MessageAttributes specified above are massaged into the following structure:92    # {93    #    attr_1: {94    #      stringValue: 'attr_1_value',95    #      stringListValues: [],96    #      binaryListValues: [],97    #      dataType: 'String'98    #    },99    #    attr_2: {100    #      stringValue: 'attr_2_value',101    #      stringListValues: [],102    #      binaryListValues: [],103    #      dataType: 'String'104    #    }105    # }106    def format_message_attributes(self, data):107        names = []108        for (k, name) in [(k, data[k]) for k in data if k.startswith('MessageAttribute') and k.endswith('.Name')]:109            attr_name = name[0]110            k_id = k.split('.')[1]111            names.append((attr_name, k_id))112        msg_attrs = {}113        for (key_name, key_id) in names:114            msg_attrs[key_name] = {}115            # Find vals for each key_id116            attrs = [(k, data[k]) for k in data117                if k.startswith('MessageAttribute.{}.'.format(key_id)) and not k.endswith('.Name')]118            for (attr_k, attr_v) in attrs:119                attr_name = attr_k.split('.')[3]120                msg_attrs[key_name][attr_name[0].lower() + attr_name[1:]] = attr_v[0]121            # These fields are set in the payload sent to Lambda.122            # It is extremely likely additional work will123            # be required to support these fields124            msg_attrs[key_name]['stringListValues'] = []125            msg_attrs[key_name]['binaryListValues'] = []126        return msg_attrs127    def return_response(self, method, path, data, headers, response, request_handler):128        if method == 'OPTIONS' and path == '/':129            # Allow CORS preflight requests to succeed.130            return 200131        if method == 'POST' and path == '/':132            region_name = extract_region_from_auth_header(headers)133            req_data = urlparse.parse_qs(to_str(data))134            action = req_data.get('Action', [None])[0]135            event_type = None136            queue_url = None137            if action == 'CreateQueue':138                event_type = event_publisher.EVENT_SQS_CREATE_QUEUE139                response_data = xmltodict.parse(response.content)140                if 'CreateQueueResponse' in response_data:141                    queue_url = response_data['CreateQueueResponse']['CreateQueueResult']['QueueUrl']142            elif action == 'DeleteQueue':143                event_type = event_publisher.EVENT_SQS_DELETE_QUEUE144                queue_url = req_data.get('QueueUrl', [None])[0]145            if event_type and queue_url:146                event_publisher.fire_event(event_type, payload={'u': event_publisher.get_hash(queue_url)})...

request_context.py

Source:request_context.py

...27        # swallow error: "Working outside of request context."28        if "Working outside" in str(e):29            return None30        raise31def extract_region_from_auth_header(headers):32    # TODO: use method from aws_stack directly (leaving import here for now, to avoid circular dependency)33    from localstack.utils.aws import aws_stack34    auth = headers.get("Authorization") or ""35    region = re.sub(r".*Credential=[^/]+/[^/]+/([^/]+)/.*", r"\1", auth)36    if region == auth:37        return None38    region = region or aws_stack.get_local_region()39    return region40def get_request_context():41    candidates = [get_proxy_request_for_thread(), get_flask_request_for_thread()]42    for req in candidates:43        if req is not None:44            return req45class RequestContextManager(object):46    """Context manager which sets the given request context (i.e., region) for the scope of the block."""47    def __init__(self, request_context):48        self.request_context = request_context49    def __enter__(self):50        THREAD_LOCAL.request_context = self.request_context51    def __exit__(self, type, value, traceback):52        THREAD_LOCAL.request_context = None53def get_region_from_request_context():54    """look up region from request context"""55    if config.USE_SINGLE_REGION:56        return57    request_context = get_request_context()58    if not request_context:59        return60    region = extract_region_from_auth_header(request_context.headers)61    # Fix region lookup for certain requests, e.g., API gateway invocations62    #  that do not contain region details in the Authorization header.63    region = request_context.headers.get(MARKER_APIGW_REQUEST_REGION) or region64    return region65def configure_region_for_current_request(region_name: str, service_name: str):66    """Manually configure (potentially overwrite) the region in the current request context. This may be67    used by API endpoints that are invoked directly by the user (without specifying AWS Authorization68    headers), to still enable transparent region lookup via aws_stack.get_region() ..."""69    # TODO: leaving import here for now, to avoid circular dependency70    from localstack.utils.aws import aws_stack71    request_context = get_request_context()72    if not request_context:73        LOG.info(74            "Unable to set region '%s' in undefined request context: %s",...

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.