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

How to use is_real_s3_url method in localstack

Best Python code snippet using localstack_python

s3_utils.py

Source:s3_utils.py

...136def get_forwarded_for_host(headers):137    x_forwarded_header = re.split(r",\s?", headers.get("X-Forwarded-For", ""))138    host = x_forwarded_header[len(x_forwarded_header) - 1]139    return host140def is_real_s3_url(url):141    return re.match(r".*s3(\-website)?\.([^\.]+\.)?amazonaws.com.*", url or "")142def is_expired(expiry_datetime):143    now_datetime = datetime.datetime.now(tz=expiry_datetime.tzinfo)144    return now_datetime > expiry_datetime145def authenticate_presign_url(method, path, headers, data=None):146    url = "{}{}".format(config.get_edge_url(), path)147    parsed = urlparse.urlparse(url)148    query_params = parse_qs(parsed.query)149    forwarded_for = get_forwarded_for_host(headers)150    if forwarded_for:151        url = re.sub("://[^/]+", "://%s" % forwarded_for, url)152    LOGGER.debug("Received presign S3 URL: %s" % url)153    sign_headers = {}154    query_string = {}...

cloudformation_listener.py

Source:cloudformation_listener.py

...127    raise Exception('Unable to get template body from input: %s' % req_data)128def is_local_service_url(url):129    candidates = ('localhost', config.LOCALSTACK_HOSTNAME, config.HOSTNAME_EXTERNAL, config.HOSTNAME)130    return url and any('://%s:' % host in url for host in candidates)131def is_real_s3_url(url):132    return re.match(r'.*s3(\-website)?\.([^\.]+\.)?amazonaws.com.*', url or '')133def convert_s3_to_local_url(url):134    if not is_real_s3_url(url):135        return url136    url_parsed = urlparse.urlparse(url)137    path = url_parsed.path138    bucket_name, _, key = path.lstrip('/').replace('//', '/').partition('/')139    # note: make sure to normalize the bucket name here!140    bucket_name = s3_listener.normalize_bucket_name(bucket_name)141    local_url = '%s/%s/%s' % (config.TEST_S3_URL, bucket_name, key)142    return local_url143def fix_hardcoded_creation_date(response):144    # TODO: remove once this is fixed upstream145    search = r'<CreationTime>\s*(2011-05-23T15:47:44Z)?\s*</CreationTime>'146    replace = r'<CreationTime>%s</CreationTime>' % timestamp_millis()147    fix_in_response(search, replace, response)148def fix_region_in_arns(response):149    search = r'arn:aws:cloudformation:[^:]+:'150    replace = r'arn:aws:cloudformation:%s:' % aws_stack.get_region()151    fix_in_response(search, replace, response)152def fix_in_response(search, replace, response):153    response._content = re.sub(search, replace, to_str(response._content or ''))154    response.headers['Content-Length'] = str(len(response._content))155class ProxyListenerCloudFormation(ProxyListener):156    def forward_request(self, method, path, data, headers):157        if method == 'OPTIONS':158            return 200159        data = data or ''160        data_orig = data161        data = aws_stack.fix_account_id_in_arns(data, existing='%3A{}%3Astack/'.format(TEST_AWS_ACCOUNT_ID),162            replace='%3A{}%3Astack/'.format(MOTO_CLOUDFORMATION_ACCOUNT_ID), colon_delimiter='')163        data = aws_stack.fix_account_id_in_arns(data, existing='%3A{}%3AchangeSet/'.format(TEST_AWS_ACCOUNT_ID),164            replace='%3A{}%3AchangeSet/'.format(MOTO_CLOUDFORMATION_ACCOUNT_ID), colon_delimiter='')165        data = aws_stack.fix_account_id_in_arns(data, existing=TEST_AWS_ACCOUNT_ID,166            replace=MOTO_ACCOUNT_ID, colon_delimiter='%3A')167        req_data = None168        if method == 'POST' and path == '/':169            req_data = urlparse.parse_qs(to_str(data))170            req_data = dict([(k, v[0]) for k, v in req_data.items()])171            action = req_data.get('Action')172            stack_name = req_data.get('StackName')173            if action == 'CreateStack':174                event_publisher.fire_event(175                    event_publisher.EVENT_CLOUDFORMATION_CREATE_STACK,176                    payload={'n': event_publisher.get_hash(stack_name)}177                )178            if action == 'DeleteStack':179                client = aws_stack.connect_to_service(180                    'cloudformation',181                    region_name=aws_stack.extract_region_from_auth_header(headers)182                )183                stack_resources = client.list_stack_resources(StackName=stack_name)['StackResourceSummaries']184                template_deployer.delete_stack(stack_name, stack_resources)185            if action == 'DescribeStackEvents':186                # fix an issue where moto cannot handle ARNs as stack names (or missing names)187                run_fix = not stack_name188                if stack_name:189                    if stack_name.startswith('arn:aws:cloudformation'):190                        run_fix = True191                        pattern = r'arn:aws:cloudformation:[^:]+:[^:]+:stack/([^/]+)(/.+)?'192                        stack_name = re.sub(pattern, r'\1', stack_name)193                if run_fix:194                    stack_names = [stack_name] if stack_name else self._list_stack_names()195                    client = aws_stack.connect_to_service('cloudformation')196                    events = []197                    for stack_name in stack_names:198                        tmp = client.describe_stack_events(StackName=stack_name)['StackEvents'][:1]199                        events.extend(tmp)200                    events = [{'member': e} for e in events]201                    response_content = '<StackEvents>%s</StackEvents>' % obj_to_xml(events)202                    return make_response('DescribeStackEvents', response_content)203        if req_data:204            if action == 'ValidateTemplate':205                return validate_template(req_data)206            if action in ['CreateStack', 'UpdateStack', 'CreateChangeSet']:207                do_replace_url = is_real_s3_url(req_data.get('TemplateURL'))208                if do_replace_url:209                    req_data['TemplateURL'] = convert_s3_to_local_url(req_data['TemplateURL'])210                url = req_data.get('TemplateURL', '')211                is_custom_local_endpoint = is_local_service_url(url) and '://localhost:' not in url212                modified_template_body = transform_template(req_data)213                if not modified_template_body and is_custom_local_endpoint:214                    modified_template_body = get_template_body(req_data)215                if modified_template_body:216                    req_data.pop('TemplateURL', None)217                    req_data['TemplateBody'] = modified_template_body218                if modified_template_body or do_replace_url:219                    data = urlparse.urlencode(req_data, doseq=True)220                    return Request(data=data, headers=headers, method=method)221            if data != data_orig or action in ['DescribeChangeSet', 'ExecuteChangeSet']:...

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.