How to use get_template_body method in localstack

Best Python code snippet using localstack_python

test_renderers.py

Source:test_renderers.py

...54}55parameters = {"p1": "p11", "p2": "p22"}56def test_get_template_body_local_file():57    url = "examples/consumables/aws/network/vpc.mako"58    parsed_url, body = get_template_body(url)59    assert parsed_url.scheme == ""60    assert isinstance(body, str)61def test_get_template_body_http(mocker):62    url = "https://my-site.com/vpc.mako"63    mock_req = mocker.patch("requests.get")64    mock_resp = mocker.MagicMock()65    mock_req.return_value = mock_resp66    mock_resp.text = "somedata"67    parsed_url, body = get_template_body(url)68    assert parsed_url.scheme == "https"69    assert body == "somedata"70def test_get_template_body_s3(mocker):71    bucket = "my-bucket"72    filename = "vpc.mako"73    url = f"s3://{bucket}/{filename}"74    content = "somedata"75    mock_boto = mocker.patch("gpwm.renderers.boto3")76    mock_boto.resource.return_value.Object.return_value.get.return_value.__getitem__.return_value.read.return_value = content # noqa77    parsed_url, body = get_template_body(url)78    mock_boto.resource.assert_called_with("s3")79    mock_boto.resource().Object.assert_called_with(bucket, filename)80    assert parsed_url.scheme == "s3"81    assert body == content82def test_get_template_use_prefix(mocker):83    url = "vpc.mako"84    prefix = "https://my-site.com"85    mocker.patch.dict('os.environ', {"GPWM_TEMPLATE_URL_PREFIX": prefix})86    mock_req = mocker.patch("requests.get")87    mock_resp = mocker.MagicMock()88    mock_req.return_value = mock_resp89    mock_resp.text = "somedata"90    parsed_url, body = get_template_body(url)91    assert parsed_url.scheme == "https"92    assert body == "somedata"93    mock_req.assert_called_with(f"{prefix}/{url}")94def test_get_template_body_http_exception(mocker):95    url = "https://my-site.com/vpc.mako"96    mock_req = mocker.patch("requests.get")97    mock_req.side_effect = requests.exceptions.RequestException()98    with pytest.raises(SystemExit):99        parsed_url, body = get_template_body(url)100def test_get_template_body_s3_exception(mocker):101    bucket = "my-bucket"102    filename = "vpc.mako"103    url = f"s3://{bucket}/{filename}"104    content = "somedata"105    mock_boto = mocker.patch("gpwm.renderers.boto3")106#    mock_boto.meta.return_value.client.return_value.exceptions.return_value.NoSuchBucket.return_value = Exception()  # noqa107    mock_boto.resource.return_value.Object.return_value.get.side_effect = mock_boto.meta.client.exceptions.NoSuchBucket  # noqa108    print(mock_boto.meta.client.exceptions.NoSuchBucket)109    print(mock_boto.resource("s3").Object(bucket, filename).get())110#    get_template_body(url)111#    mock_boto.resource.return_value.Object.return_value.get.return_value.__getitem__.return_value.read.side_effect = "s3.meta.client.exceptions.NoSuchBucket  # noqa112#    mock_bucket_exc = mocker.patch(113#        "gpwm.renderers.boto3.resource.meta.client.exceptions.NoSuchBucket",114#    )115#    mock_bucket_exc.return_value = Exception()116#117#    mock_get = mocker.patch("gpwm.renderers.boto3.resource().Object.get",118#    #        side_effect=mock_bucket_exc)119#    )120#    mock_get.side_effect = mock_bucket_exc121#    with pytest.raises(SystemExit):122#        parsed_url, body = get_template_body(url)123def test_parse_mako(mocker):124    stack_name = "my-stack"125    parsed_template = parse_mako(stack_name, mako_template, parameters)126    # dumping to yaml to make the assert done on a string, instead of127    # handling yaml node objects in a dict128    assert yaml.dump(parsed_template) == yaml.dump(expected_parsed_dict)129    mock_engine = mocker.patch("gpwm.renderers.mako.template.Template")130    mock_engine.render.return_value = rendered_template131    mock_yaml = mocker.patch("gpwm.renderers.yaml")132    mock_yaml.load.return_value = expected_resources_dict133    parse_mako(stack_name, mako_template, parameters)134    mock_engine.assert_called_once_with(mako_template, strict_undefined=False)135    mock_engine().render.assert_called_once_with(**parameters)136    mock_yaml.load.assert_called_once_with(mock_engine().render())...

template_preparer.py

Source:template_preparer.py

...21    for k, v in NoDatesSafeLoader.yaml_implicit_resolvers.items()22}23def transform_template(req_data) -> Optional[str]:24    """only returns string when parsing SAM template, otherwise None"""25    template_body = get_template_body(req_data)26    parsed = parse_template(template_body)27    if parsed.get("Transform") == "AWS::Serverless-2016-10-31":28        policy_map = {29            # SAM Transformer expects this map to be non-empty, but apparently the content doesn't matter (?)30            "dummy": "entry"31            # 'AWSLambdaBasicExecutionRole': 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole',32        }33        class MockPolicyLoader(object):34            def load(self):35                return policy_map36        # Note: we need to fix boto3 region, otherwise AWS SAM transformer fails37        region_before = os.environ.get("AWS_DEFAULT_REGION")38        if boto3.session.Session().region_name is None:39            os.environ["AWS_DEFAULT_REGION"] = aws_stack.get_region()40        try:41            transformed = transform_sam(parsed, {}, MockPolicyLoader())42            return json.dumps(transformed)43        finally:44            os.environ.pop("AWS_DEFAULT_REGION", None)45            if region_before is not None: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("/")...

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.