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

How to use convert_s3_to_local_url method in localstack

Best Python code snippet using localstack_python

template_preparer.py

Source:template_preparer.py

...46                os.environ["AWS_DEFAULT_REGION"] = region_before47def prepare_template_body(req_data) -> Optional[str]:  # TODO: mutating and returning48    template_url = req_data.get("TemplateURL")49    if template_url:50        req_data["TemplateURL"] = convert_s3_to_local_url(template_url)51    url = req_data.get("TemplateURL", "")52    if is_local_service_url(url):53        modified_template_body = get_template_body(req_data)54        if modified_template_body:55            req_data.pop("TemplateURL", None)56            req_data["TemplateBody"] = modified_template_body57    modified_template_body = transform_template(req_data)58    if modified_template_body:59        req_data["TemplateBody"] = modified_template_body60    return modified_template_body61def validate_template(req_data):62    # TODO implement actual validation logic63    # Note: if we enable this via moto, ensure that we have cfnlint module available (adds ~58MB in size :/)64    response_content = """65        <Capabilities></Capabilities>66        <CapabilitiesReason></CapabilitiesReason>67        <DeclaredTransforms></DeclaredTransforms>68        <Description>{description}</Description>69        <Parameters>70            {parameters}71        </Parameters>72    """73    template_body = get_template_body(req_data)74    valid_template = json.loads(template_to_json(template_body))75    parameters = "".join(76        [77            """78        <member>79            <ParameterKey>{pk}</ParameterKey>80            <DefaultValue>{dv}</DefaultValue>81            <NoEcho>{echo}</NoEcho>82            <Description>{desc}</Description>83        </member>84        """.format(85                pk=k, dv=v.get("Default", ""), echo=False, desc=v.get("Description", "")86            )87            for k, v in valid_template.get("Parameters", {}).items()88        ]89    )90    resp = response_content.format(91        parameters=parameters, description=valid_template.get("Description", "")92    )93    return resp94def get_template_body(req_data):95    body = req_data.get("TemplateBody")96    if body:97        return body98    url = req_data.get("TemplateURL")99    if url:100        response = run_safe(lambda: safe_requests.get(url, verify=False))101        # check error codes, and code 301 - fixes https://github.com/localstack/localstack/issues/1884102        status_code = 0 if response is None else response.status_code103        if response is None or status_code == 301 or status_code >= 400:104            # check if this is an S3 URL, then get the file directly from there105            url = convert_s3_to_local_url(url)106            if is_local_service_url(url):107                parsed_path = urlparse(url).path.lstrip("/")108                parts = parsed_path.partition("/")109                client = aws_stack.connect_to_service("s3")110                LOG.debug(111                    "Download CloudFormation template content from local S3: %s - %s",112                    parts[0],113                    parts[2],114                )115                result = client.get_object(Bucket=parts[0], Key=parts[2])116                body = to_str(result["Body"].read())117                return body118            raise Exception(119                "Unable to fetch template body (code %s) from URL %s" % (status_code, url)120            )121        return response.content122    raise Exception("Unable to get template body from input: %s" % req_data)123def parse_template(template):124    try:125        return json.loads(template)126    except Exception:127        yaml.add_multi_constructor(128            "", moto.cloudformation.utils.yaml_tag_constructor, Loader=NoDatesSafeLoader129        )  # TODO: remove moto dependency here130        try:131            return clone_safe(yaml.safe_load(template))132        except Exception:133            try:134                return clone_safe(yaml.load(template, Loader=NoDatesSafeLoader))135            except Exception as e:136                LOG.debug("Unable to parse CloudFormation template (%s): %s", e, template)137                raise138def template_to_json(template):139    template = parse_template(template)140    return json.dumps(template)141def is_local_service_url(url):142    if not url:143        return False144    candidates = (145        constants.LOCALHOST,146        constants.LOCALHOST_HOSTNAME,147        config.LOCALSTACK_HOSTNAME,148        config.HOSTNAME_EXTERNAL,149    )150    if any(re.match(r"^[^:]+://[^:/]*%s([:/]|$)" % host, url) for host in candidates):151        return True152    host = url.split("://")[-1].split("/")[0]153    return "localhost" in host154def convert_s3_to_local_url(url):155    url_parsed = urlparse(url)156    path = url_parsed.path157    headers = CaseInsensitiveDict({"Host": url_parsed.netloc})158    bucket_name = s3_utils.extract_bucket_name(headers, path)159    key_name = s3_utils.extract_key_name(headers, path)160    # note: make sure to normalize the bucket name here!161    bucket_name = s3_listener.normalize_bucket_name(bucket_name)162    local_url = f"{config.service_url('s3')}/{bucket_name}/{key_name}"...

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.