Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use load_template_raw method in localstack
Best Python code snippet using localstack_python
test_cloudformation_changesets.py
Source:test_cloudformation_changesets.py 
...7from localstack.utils.common import short_uid8from localstack.utils.generic.wait_utils import wait_until9from localstack.utils.sync import poll_condition10# TODO: refactor file and remove this compatibility fn11def load_template_raw(file_name: str):12 return load_template_file(os.path.join(os.path.dirname(__file__), "../templates", file_name))13def test_create_change_set_without_parameters(14 cfn_client, sns_client, cleanup_stacks, cleanup_changesets, is_change_set_created_and_available15):16 stack_name = f"stack-{short_uid()}"17 change_set_name = f"change-set-{short_uid()}"18 response = cfn_client.create_change_set(19 StackName=stack_name,20 ChangeSetName=change_set_name,21 TemplateBody=load_template_raw("sns_topic_simple.yaml"),22 ChangeSetType="CREATE",23 )24 change_set_id = response["Id"]25 stack_id = response["StackId"]26 assert change_set_id27 assert stack_id28 try:29 # make sure the change set wasn't executed (which would create a topic)30 topics = sns_client.list_topics()31 topic_arns = list(map(lambda x: x["TopicArn"], topics["Topics"]))32 assert not any("sns-topic-simple" in arn for arn in topic_arns)33 # stack is initially in REVIEW_IN_PROGRESS state. only after executing the change_set will it change its status34 stack_response = cfn_client.describe_stacks(StackName=stack_id)35 assert stack_response["Stacks"][0]["StackStatus"] == "REVIEW_IN_PROGRESS"36 # Change set can now either be already created/available or it is pending/unavailable37 wait_until(38 is_change_set_created_and_available(change_set_id), 2, 10, strategy="exponential"39 )40 describe_response = cfn_client.describe_change_set(ChangeSetName=change_set_id)41 assert describe_response["ChangeSetName"] == change_set_name42 assert describe_response["ChangeSetId"] == change_set_id43 assert describe_response["StackId"] == stack_id44 assert describe_response["StackName"] == stack_name45 assert describe_response["ExecutionStatus"] == "AVAILABLE"46 assert describe_response["Status"] == "CREATE_COMPLETE"47 changes = describe_response["Changes"]48 assert len(changes) == 149 assert changes[0]["Type"] == "Resource"50 assert changes[0]["ResourceChange"]["Action"] == "Add"51 assert changes[0]["ResourceChange"]["ResourceType"] == "AWS::SNS::Topic"52 assert changes[0]["ResourceChange"]["LogicalResourceId"] == "topic123"53 finally:54 cleanup_stacks([stack_id])55 cleanup_changesets([change_set_id])56# TODO: implement57@pytest.mark.xfail(condition=not is_aws_cloud(), reason="Not properly implemented")58def test_create_change_set_update_without_parameters(59 cfn_client,60 sns_client,61 cleanup_stacks,62 cleanup_changesets,63 is_change_set_created_and_available,64 is_change_set_finished,65 snapshot,66):67 snapshot.add_transformer(snapshot.transform.cloudformation_api())68 """after creating a stack via a CREATE change set we send an UPDATE change set changing the SNS topic name"""69 stack_name = f"stack-{short_uid()}"70 change_set_name = f"change-set-{short_uid()}"71 change_set_name2 = f"change-set-{short_uid()}"72 response = cfn_client.create_change_set(73 StackName=stack_name,74 ChangeSetName=change_set_name,75 TemplateBody=load_template_raw("sns_topic_simple.yaml"),76 ChangeSetType="CREATE",77 )78 snapshot.match("create_change_set", response)79 change_set_id = response["Id"]80 stack_id = response["StackId"]81 assert change_set_id82 assert stack_id83 try:84 # Change set can now either be already created/available or it is pending/unavailable85 wait_until(is_change_set_created_and_available(change_set_id))86 cfn_client.execute_change_set(ChangeSetName=change_set_id)87 wait_until(is_change_set_finished(change_set_id))88 template = load_template_raw("sns_topic_simple.yaml")89 update_response = cfn_client.create_change_set(90 StackName=stack_name,91 ChangeSetName=change_set_name2,92 TemplateBody=template.replace("sns-topic-simple", "sns-topic-simple-2"),93 ChangeSetType="UPDATE",94 )95 assert wait_until(is_change_set_created_and_available(update_response["Id"]))96 snapshot.match(97 "describe_change_set",98 cfn_client.describe_change_set(ChangeSetName=update_response["Id"]),99 )100 snapshot.match("list_change_set", cfn_client.list_change_sets(StackName=stack_name))101 describe_response = cfn_client.describe_change_set(ChangeSetName=update_response["Id"])102 changes = describe_response["Changes"]103 assert len(changes) == 1104 assert changes[0]["Type"] == "Resource"105 change = changes[0]["ResourceChange"]106 assert change["Action"] == "Modify"107 assert change["ResourceType"] == "AWS::SNS::Topic"108 assert change["LogicalResourceId"] == "topic123"109 assert "sns-topic-simple" in change["PhysicalResourceId"]110 assert change["Replacement"] == "True"111 assert "Properties" in change["Scope"]112 assert len(change["Details"]) == 1113 assert change["Details"][0]["Target"]["Name"] == "TopicName"114 assert change["Details"][0]["Target"]["RequiresRecreation"] == "Always"115 finally:116 cleanup_changesets(changesets=[change_set_id])117 cleanup_stacks(stacks=[stack_id])118@pytest.mark.skip(reason="TODO")119def test_create_change_set_with_template_url(cfn_client):120 pass121@pytest.mark.xfail(reason="change set type not implemented")122def test_create_change_set_create_existing(123 cfn_client, is_stack_created, cleanup_changesets, cleanup_stacks124):125 """tries to create an already existing stack"""126 stack_name = f"stack-{short_uid()}"127 change_set_name = f"change-set-{short_uid()}"128 response = cfn_client.create_change_set(129 StackName=stack_name,130 ChangeSetName=change_set_name,131 TemplateBody=load_template_raw("sns_topic_simple.yaml"),132 ChangeSetType="CREATE",133 )134 change_set_id = response["Id"]135 stack_id = response["StackId"]136 assert change_set_id137 assert stack_id138 try:139 cfn_client.execute_change_set(ChangeSetName=change_set_id)140 wait_until(is_stack_created(stack_id))141 with pytest.raises(Exception) as ex:142 change_set_name2 = f"change-set-{short_uid()}"143 cfn_client.create_change_set(144 StackName=stack_name,145 ChangeSetName=change_set_name2,146 TemplateBody=load_template_raw("sns_topic_simple.yaml"),147 ChangeSetType="CREATE",148 )149 assert ex is not None150 finally:151 cleanup_changesets([change_set_id])152 cleanup_stacks([stack_id])153def test_create_change_set_update_nonexisting(cfn_client):154 stack_name = f"stack-{short_uid()}"155 change_set_name = f"change-set-{short_uid()}"156 with pytest.raises(Exception) as ex:157 response = cfn_client.create_change_set(158 StackName=stack_name,159 ChangeSetName=change_set_name,160 TemplateBody=load_template_raw("sns_topic_simple.yaml"),161 ChangeSetType="UPDATE",162 )163 change_set_id = response["Id"]164 stack_id = response["StackId"]165 assert change_set_id166 assert stack_id167 err = ex.value.response["Error"]168 assert err["Code"] == "ValidationError"169 assert "does not exist" in err["Message"]170@pytest.mark.skip(reason="TODO")171def test_create_change_set_import(cfn_client):172 """test importing existing resources into a stack via the change set"""173 pass # TODO174def test_create_change_set_invalid_params(cfn_client):175 stack_name = f"stack-{short_uid()}"176 change_set_name = f"change-set-{short_uid()}"177 with pytest.raises(ClientError) as ex:178 cfn_client.create_change_set(179 StackName=stack_name,180 ChangeSetName=change_set_name,181 TemplateBody=load_template_raw("sns_topic_simple.yaml"),182 ChangeSetType="INVALID",183 )184 err = ex.value.response["Error"]185 assert err["Code"] == "ValidationError"186def test_create_change_set_missing_stackname(cfn_client):187 """in this case boto doesn't even let us send the request"""188 change_set_name = f"change-set-{short_uid()}"189 with pytest.raises(Exception):190 cfn_client.create_change_set(191 StackName="",192 ChangeSetName=change_set_name,193 TemplateBody=load_template_raw("sns_topic_simple.yaml"),194 ChangeSetType="CREATE",195 )196def test_create_change_set_with_ssm_parameter(197 cfn_client,198 sns_client,199 ssm_client,200 cleanup_changesets,201 cleanup_stacks,202 is_change_set_created_and_available,203 is_stack_created,204):205 """References a simple stack parameter"""206 stack_name = f"stack-{short_uid()}"207 change_set_name = f"change-set-{short_uid()}"208 parameter_name = f"ls-param-{short_uid()}"209 parameter_value = f"ls-param-value-{short_uid()}"210 sns_topic_logical_id = "topic123"211 parameter_logical_id = "parameter123"212 ssm_client.put_parameter(Name=parameter_name, Value=parameter_value, Type="String")213 template = load_template_raw("dynamicparameter_ssm_string.yaml")214 template_rendered = jinja2.Template(template).render(parameter_name=parameter_name)215 response = cfn_client.create_change_set(216 StackName=stack_name,217 ChangeSetName=change_set_name,218 TemplateBody=template_rendered,219 ChangeSetType="CREATE",220 )221 change_set_id = response["Id"]222 stack_id = response["StackId"]223 assert change_set_id224 assert stack_id225 try:226 # make sure the change set wasn't executed (which would create a new topic)227 list_topics_response = sns_client.list_topics()228 matching_topics = [229 t for t in list_topics_response["Topics"] if parameter_value in t["TopicArn"]230 ]231 assert matching_topics == []232 # stack is initially in REVIEW_IN_PROGRESS state. only after executing the change_set will it change its status233 stack_response = cfn_client.describe_stacks(StackName=stack_id)234 assert stack_response["Stacks"][0]["StackStatus"] == "REVIEW_IN_PROGRESS"235 # Change set can now either be already created/available or it is pending/unavailable236 wait_until(is_change_set_created_and_available(change_set_id))237 describe_response = cfn_client.describe_change_set(ChangeSetName=change_set_id)238 assert describe_response["ChangeSetName"] == change_set_name239 assert describe_response["ChangeSetId"] == change_set_id240 assert describe_response["StackId"] == stack_id241 assert describe_response["StackName"] == stack_name242 assert describe_response["ExecutionStatus"] == "AVAILABLE"243 assert describe_response["Status"] == "CREATE_COMPLETE"244 changes = describe_response["Changes"]245 assert len(changes) == 1246 assert changes[0]["Type"] == "Resource"247 assert changes[0]["ResourceChange"]["Action"] == "Add"248 assert changes[0]["ResourceChange"]["ResourceType"] == "AWS::SNS::Topic"249 assert changes[0]["ResourceChange"]["LogicalResourceId"] == sns_topic_logical_id250 parameters = describe_response["Parameters"]251 assert len(parameters) == 1252 assert parameters[0]["ParameterKey"] == parameter_logical_id253 assert parameters[0]["ParameterValue"] == parameter_name254 assert parameters[0]["ResolvedValue"] == parameter_value # the important part255 cfn_client.execute_change_set(ChangeSetName=change_set_id)256 wait_until(is_stack_created(stack_id))257 topics = sns_client.list_topics()258 topic_arns = list(map(lambda x: x["TopicArn"], topics["Topics"]))259 assert any((parameter_value in t) for t in topic_arns)260 finally:261 cleanup_changesets([change_set_id])262 cleanup_stacks([stack_id])263def test_describe_change_set_nonexisting(cfn_client):264 with pytest.raises(Exception) as ex:265 cfn_client.describe_change_set(ChangeSetName="DoesNotExist")266 assert ex.value.response["Error"]["Code"] == "ResourceNotFoundException"267def test_execute_change_set(268 cfn_client,269 sns_client,270 is_change_set_finished,271 is_change_set_created_and_available,272 is_change_set_failed_and_unavailable,273 cleanup_changesets,274 cleanup_stacks,275):276 """check if executing a change set succeeds in creating/modifying the resources in changed"""277 stack_name = f"stack-{short_uid()}"278 change_set_name = f"change-set-{short_uid()}"279 response = cfn_client.create_change_set(280 StackName=stack_name,281 ChangeSetName=change_set_name,282 TemplateBody=load_template_raw("sns_topic_simple.yaml"),283 ChangeSetType="CREATE",284 )285 change_set_id = response["Id"]286 stack_id = response["StackId"]287 assert change_set_id288 assert stack_id289 try:290 assert wait_until(is_change_set_created_and_available(change_set_id=change_set_id))291 cfn_client.execute_change_set(ChangeSetName=change_set_id)292 assert wait_until(is_change_set_finished(change_set_id))293 # check if stack resource was created294 topics = sns_client.list_topics()295 topic_arns = list(map(lambda x: x["TopicArn"], topics["Topics"]))296 assert any(("sns-topic-simple" in t) for t in topic_arns)297 # new change set name298 change_set_name = f"change-set-{short_uid()}"299 # check if update with identical stack leads to correct behavior300 response = cfn_client.create_change_set(301 StackName=stack_name,302 ChangeSetName=change_set_name,303 TemplateBody=load_template_raw("sns_topic_simple.yaml"),304 ChangeSetType="UPDATE",305 )306 change_set_id = response["Id"]307 stack_id = response["StackId"]308 assert wait_until(is_change_set_failed_and_unavailable(change_set_id=change_set_id))309 describe_failed_change_set_result = cfn_client.describe_change_set(310 ChangeSetName=change_set_id311 )312 assert describe_failed_change_set_result["ChangeSetName"] == change_set_name313 assert (314 describe_failed_change_set_result["StatusReason"]315 == "The submitted information didn't contain changes. Submit different information to create a change set."316 )317 with pytest.raises(ClientError) as e:318 cfn_client.execute_change_set(ChangeSetName=change_set_id)319 e.match("InvalidChangeSetStatus")320 e.match(321 rf"ChangeSet \[{change_set_id}\] cannot be executed in its current status of \[FAILED\]"322 )323 finally:324 cleanup_changesets([change_set_id])325 cleanup_stacks([stack_id])326def test_delete_change_set_nonexisting(cfn_client):327 with pytest.raises(Exception) as ex:328 cfn_client.delete_change_set(ChangeSetName="DoesNotExist")329 assert ex.value.response["Error"]["Code"] == "ResourceNotFoundException"330@pytest.mark.aws_validated331def test_create_and_then_remove_non_supported_resource_change_set(deploy_cfn_template):332 # first deploy cfn with a CodeArtifact resource that is not actually supported333 stack = deploy_cfn_template(334 template=load_template_raw("code_artifact_template.yaml"),335 parameters={"CADomainName": f"domainname-{short_uid()}"},336 )337 # removal of CodeArtifact should not throw exception338 deploy_cfn_template(339 is_update=True,340 template=load_template_raw("code_artifact_remove_template.yaml"),341 stack_name=stack.stack_name,342 )343@pytest.mark.aws_validated344def test_create_and_then_remove_supported_resource_change_set(deploy_cfn_template, s3_client):345 first_bucket_name = f"test-bucket-1-{short_uid()}"346 second_bucket_name = f"test-bucket-2-{short_uid()}"347 stack = deploy_cfn_template(348 template=load_template_raw("for_removal_setup.yaml"),349 template_mapping={350 "first_bucket_name": first_bucket_name,351 "second_bucket_name": second_bucket_name,352 },353 )354 available_buckets = s3_client.list_buckets()355 bucket_names = [bucket["Name"] for bucket in available_buckets["Buckets"]]356 assert first_bucket_name in bucket_names357 assert second_bucket_name in bucket_names358 deploy_cfn_template(359 is_update=True,360 template=load_template_raw("for_removal_remove.yaml"),361 template_mapping={"first_bucket_name": first_bucket_name},362 stack_name=stack.stack_name,363 )364 def assert_bucket_gone():365 available_buckets = s3_client.list_buckets()366 bucket_names = [bucket["Name"] for bucket in available_buckets["Buckets"]]367 return first_bucket_name in bucket_names and second_bucket_name not in bucket_names...
test_cloudformation_stacks.py
Source:test_cloudformation_stacks.py
...5from localstack.testing.aws.cloudformation_utils import load_template_file6from localstack.utils.common import short_uid7from localstack.utils.generic.wait_utils import wait_until8# TODO: refactor file and remove this compatibility fn9def load_template_raw(file_name: str):10 return load_template_file(os.path.join(os.path.dirname(__file__), "../templates", file_name))11def test_create_stack_with_ssm_parameters(12 cfn_client, ssm_client, sns_client, cleanup_stacks, is_stack_created13):14 stack_name = f"stack-{short_uid()}"15 parameter_name = f"ls-param-{short_uid()}"16 parameter_value = f"ls-param-value-{short_uid()}"17 parameter_logical_id = "parameter123"18 ssm_client.put_parameter(Name=parameter_name, Value=parameter_value, Type="String")19 template = load_template_raw("dynamicparameter_ssm_string.yaml")20 template_rendered = jinja2.Template(template).render(parameter_name=parameter_name)21 response = cfn_client.create_stack(22 StackName=stack_name,23 TemplateBody=template_rendered,24 )25 stack_id = response["StackId"]26 assert stack_id27 try:28 wait_until(is_stack_created(stack_id))29 created_stack = cfn_client.describe_stacks(StackName=stack_name)["Stacks"][0]30 assert created_stack is not None31 assert created_stack["Parameters"][0]["ParameterKey"] == parameter_logical_id32 assert created_stack["Parameters"][0]["ParameterValue"] == parameter_name33 assert created_stack["Parameters"][0]["ResolvedValue"] == parameter_value34 topics = sns_client.list_topics()35 topic_arns = [t["TopicArn"] for t in topics["Topics"]]36 assert any(parameter_value in t for t in topic_arns)37 finally:38 cleanup_stacks([stack_id])39 # TODO: cleanup parameter40def test_list_stack_resources_for_removed_resource(41 cfn_client, is_stack_created, is_change_set_finished42):43 event_bus_name = f"bus-{short_uid()}"44 template = jinja2.Template(load_template_raw("eventbridge_policy.yaml")).render(45 event_bus_name=event_bus_name46 )47 stack_name = f"stack-{short_uid()}"48 response = cfn_client.create_stack(StackName=stack_name, TemplateBody=template)49 stack_id = response["StackId"]50 assert stack_id51 wait_until(is_stack_created(stack_id))52 # get list of stack resources53 resources = cfn_client.list_stack_resources(StackName=stack_name)["StackResourceSummaries"]54 resources_before = len(resources)55 assert resources_before == 356 statuses = set([res["ResourceStatus"] for res in resources])57 assert statuses == {"CREATE_COMPLETE", "UPDATE_COMPLETE"}58 # remove one resource from the template, then update stack (via change set)59 template_dict = yaml.load(template)60 template_dict["Resources"].pop("eventPolicy2")61 template2 = yaml.dump(template_dict)62 response = cfn_client.create_change_set(63 StackName=stack_name, ChangeSetName="cs1", TemplateBody=template264 )65 change_set_id = response["Id"]66 cfn_client.execute_change_set(ChangeSetName=change_set_id)67 wait_until(is_change_set_finished(change_set_id))68 # get list of stack resources, again - make sure that deleted resource is not contained in result69 resources = cfn_client.list_stack_resources(StackName=stack_name)["StackResourceSummaries"]70 assert len(resources) == resources_before - 171 statuses = set([res["ResourceStatus"] for res in resources])72 assert statuses == {"CREATE_COMPLETE", "UPDATE_COMPLETE"}73@pytest.mark.xfail(reason="outputs don't behave well in combination with conditions")74@pytest.mark.aws_validated75def test_parameter_usepreviousvalue_behavior(cfn_client, cleanups):76 stack_name = f"stack-{short_uid()}"77 cleanups.append(lambda _: cfn_client.delete_stack(StackName=stack_name))78 # 1. create with overridden default value. Due to the condition this should neither create the optional topic, nor the corresponding output79 create_response = cfn_client.create_stack(80 StackName=stack_name,81 TemplateBody=load_template_raw("cfn_reuse_param.yaml"),82 Parameters=[{"ParameterKey": "DeployParam", "ParameterValue": "no"}],83 )84 stack_id = create_response["StackId"]85 def wait_stack_done():86 return cfn_client.describe_stacks(StackName=stack_id)["Stacks"][0]["StackStatus"] in [87 "CREATE_COMPLETE",88 "UPDATE_COMPLETE",89 ]90 assert wait_until(wait_stack_done)91 stack_describe_response = cfn_client.describe_stacks(StackName=stack_id)["Stacks"][0]92 assert len(stack_describe_response["Outputs"]) == 193 # 2. update using UsePreviousValue. DeployParam should still be "no", still overriding the default and the only change should be the changed tag on the required topic94 cfn_client.update_stack(95 StackName=stack_name,96 TemplateBody=load_template_raw("cfn_reuse_param.yaml"),97 Parameters=[98 {"ParameterKey": "CustomTag", "ParameterValue": "trigger-change"},99 {"ParameterKey": "DeployParam", "UsePreviousValue": True},100 ],101 )102 assert wait_until(wait_stack_done)103 stack_describe_response = cfn_client.describe_stacks(StackName=stack_id)["Stacks"][0]104 assert len(stack_describe_response["Outputs"]) == 1105 # 3. update with setting the deployparam to "yes" not. The condition will evaluate to true and thus create the topic + output106 # note: for an even trickier challenge for the cloudformation engine, remove the second parameter key. Behavior should stay the same.107 cfn_client.update_stack(108 StackName=stack_name,109 TemplateBody=load_template_raw("cfn_reuse_param.yaml"),110 Parameters=[111 {"ParameterKey": "CustomTag", "ParameterValue": "trigger-change-2"},112 {"ParameterKey": "DeployParam", "ParameterValue": "yes"},113 ],114 )115 assert wait_until(wait_stack_done)116 stack_describe_response = cfn_client.describe_stacks(StackName=stack_id)["Stacks"][0]...
entity.py
Source:entity.py
...4 def __init__(self, color):5 self.__color = options.color(color)6 def color(self):7 return self.__color8 def load_template_raw(self, name):9 with open('templates/{0}'.format(name), 'r') as file:10 return file.read()11 def load_template(self, name):12 return self.load_template_raw(name).replace('\n', '')13 def normalize_name(self, name):14 return re.sub('[^0-9a-z]+', '_', name.lower())15 def compact_string(self, 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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
