Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use prepare_template_body method in localstack
Best Python code snippet using localstack_python
cloudformation_api.py
Source:cloudformation_api.py 
...393# API ENDPOINTS394# --------------395def create_stack(req_params):396 state = CloudFormationRegion.get()397 template_deployer.prepare_template_body(req_params) # TODO: avoid mutating req_params directly398 template = template_preparer.parse_template(req_params["TemplateBody"])399 stack_name = template["StackName"] = req_params.get("StackName")400 stack = Stack(req_params, template)401 # find existing stack with same name, and remove it if this stack is in DELETED state402 existing = ([s for s in state.stacks.values() if s.stack_name == stack_name] or [None])[0]403 if existing:404 if "DELETE" not in existing.status:405 return error_response(406 'Stack named "%s" already exists with status "%s"' % (stack_name, existing.status),407 code=400,408 code_string="ValidationError",409 )410 state.stacks.pop(existing.stack_id)411 state.stacks[stack.stack_id] = stack412 LOG.debug(413 'Creating stack "%s" with %s resources ...'414 % (stack.stack_name, len(stack.template_resources))415 )416 deployer = template_deployer.TemplateDeployer(stack)417 try:418 # TODO: create separate step to first resolve parameters419 deployer.deploy_stack()420 except Exception as e:421 stack.set_stack_status("CREATE_FAILED")422 msg = 'Unable to create stack "%s": %s' % (stack.stack_name, e)423 LOG.debug("%s %s" % (msg, traceback.format_exc()))424 return error_response(msg, code=400, code_string="ValidationError")425 result = {"StackId": stack.stack_id}426 return result427def create_stack_set(req_params):428 state = CloudFormationRegion.get()429 stack_set = StackSet(req_params)430 stack_set_id = short_uid()431 stack_set.metadata["StackSetId"] = stack_set_id432 state.stack_sets[stack_set_id] = stack_set433 result = {"StackSetId": stack_set_id}434 return result435def create_stack_instances(req_params):436 state = CloudFormationRegion.get()437 set_name = req_params.get("StackSetName")438 stack_set = [sset for sset in state.stack_sets.values() if sset.stack_set_name == set_name]439 if not stack_set:440 return not_found_error('Stack set named "%s" does not exist' % set_name)441 stack_set = stack_set[0]442 op_id = req_params.get("OperationId") or short_uid()443 sset_meta = stack_set.metadata444 accounts = extract_url_encoded_param_list(req_params, "Accounts.member.%s")445 accounts = accounts or extract_url_encoded_param_list(446 req_params, "DeploymentTargets.Accounts.member.%s"447 )448 regions = extract_url_encoded_param_list(req_params, "Regions.member.%s")449 stacks_to_await = []450 for account in accounts:451 for region in regions:452 # deploy new stack453 LOG.debug('Deploying instance for stack set "%s" in region "%s"' % (set_name, region))454 cf_client = aws_stack.connect_to_service("cloudformation", region_name=region)455 kwargs = select_attributes(sset_meta, "TemplateBody") or select_attributes(456 sset_meta, "TemplateURL"457 )458 stack_name = "sset-%s-%s" % (set_name, account)459 result = cf_client.create_stack(StackName=stack_name, **kwargs)460 stacks_to_await.append((stack_name, region))461 # store stack instance462 instance = {463 "StackSetId": sset_meta["StackSetId"],464 "OperationId": op_id,465 "Account": account,466 "Region": region,467 "StackId": result["StackId"],468 "Status": "CURRENT",469 "StackInstanceStatus": {"DetailedStatus": "SUCCEEDED"},470 }471 instance = StackInstance(instance)472 stack_set.stack_instances.append(instance)473 # wait for completion of stack474 for stack in stacks_to_await:475 aws_stack.await_stack_completion(stack[0], region_name=stack[1])476 # record operation477 operation = {478 "OperationId": op_id,479 "StackSetId": stack_set.metadata["StackSetId"],480 "Action": "CREATE",481 "Status": "SUCCEEDED",482 }483 stack_set.operations[op_id] = operation484 result = {"OperationId": op_id}485 return result486def delete_stack(req_params):487 stack_name = req_params.get("StackName")488 stack = find_stack(stack_name)489 deployer = template_deployer.TemplateDeployer(stack)490 deployer.delete_stack()491 return {}492def delete_stack_set(req_params):493 state = CloudFormationRegion.get()494 set_name = req_params.get("StackSetName")495 stack_set = [sset for sset in state.stack_sets.values() if sset.stack_set_name == set_name]496 if not stack_set:497 return not_found_error('Stack set named "%s" does not exist' % set_name)498 for instance in stack_set[0].stack_instances:499 deployer = template_deployer.TemplateDeployer(instance.stack)500 deployer.delete_stack()501 return {}502def update_stack(req_params):503 stack_name = req_params.get("StackName")504 stack = find_stack(stack_name)505 if not stack:506 return not_found_error('Unable to update non-existing stack "%s"' % stack_name)507 template_preparer.prepare_template_body(req_params)508 template = template_preparer.parse_template(req_params["TemplateBody"])509 new_stack = Stack(req_params, template)510 deployer = template_deployer.TemplateDeployer(stack)511 try:512 deployer.update_stack(new_stack)513 except Exception as e:514 stack.set_stack_status("UPDATE_FAILED")515 msg = 'Unable to update stack "%s": %s' % (stack_name, e)516 LOG.debug("%s %s" % (msg, traceback.format_exc()))517 return error_response(msg, code=400, code_string="ValidationError")518 result = {"StackId": stack.stack_id}519 return result520def update_stack_set(req_params):521 state = CloudFormationRegion.get()522 set_name = req_params.get("StackSetName")523 stack_set = [sset for sset in state.stack_sets.values() if sset.stack_set_name == set_name]524 if not stack_set:525 return not_found_error('Stack set named "%s" does not exist' % set_name)526 stack_set = stack_set[0]527 stack_set.metadata.update(req_params)528 op_id = req_params.get("OperationId") or short_uid()529 operation = {530 "OperationId": op_id,531 "StackSetId": stack_set.metadata["StackSetId"],532 "Action": "UPDATE",533 "Status": "SUCCEEDED",534 }535 stack_set.operations[op_id] = operation536 return {"OperationId": op_id}537def describe_stacks(req_params):538 state = CloudFormationRegion.get()539 stack_name = req_params.get("StackName")540 stack_list = list(state.stacks.values())541 stacks = [542 s.describe_details() for s in stack_list if stack_name in [None, s.stack_name, s.stack_id]543 ]544 if stack_name and not stacks:545 return error_response(546 "Stack with id %s does not exist" % stack_name,547 code=400,548 code_string="ValidationError",549 )550 result = {"Stacks": stacks}551 return result552def list_stacks(req_params):553 state = CloudFormationRegion.get()554 stack_status_filters = _get_status_filter_members(req_params)555 stacks = [556 s.describe_details()557 for s in state.stacks.values()558 if not stack_status_filters or s.status in stack_status_filters559 ]560 attrs = [561 "StackId",562 "StackName",563 "TemplateDescription",564 "CreationTime",565 "LastUpdatedTime",566 "DeletionTime",567 "StackStatus",568 "StackStatusReason",569 "ParentId",570 "RootId",571 "DriftInformation",572 ]573 stacks = [select_attributes(stack, attrs) for stack in stacks]574 result = {"StackSummaries": stacks}575 return result576def describe_stack_resource(req_params):577 stack_name = req_params.get("StackName")578 resource_id = req_params.get("LogicalResourceId")579 stack = find_stack(stack_name)580 if not stack:581 return stack_not_found_error(stack_name)582 details = stack.resource_status(resource_id)583 result = {"StackResourceDetail": details}584 return result585def describe_stack_resources(req_params):586 stack_name = req_params.get("StackName")587 resource_id = req_params.get("LogicalResourceId")588 phys_resource_id = req_params.get("PhysicalResourceId")589 if phys_resource_id and stack_name:590 return error_response("Cannot specify both StackName and PhysicalResourceId", code=400)591 # TODO: filter stack by PhysicalResourceId!592 stack = find_stack(stack_name)593 if not stack:594 return stack_not_found_error(stack_name)595 statuses = [596 res_status597 for res_id, res_status in stack.resource_states.items()598 if resource_id in [res_id, None]599 ]600 return {"StackResources": statuses}601def list_stack_resources(req_params):602 result = describe_stack_resources(req_params)603 if not isinstance(result, dict):604 return result605 result = {"StackResourceSummaries": result.pop("StackResources")}606 return result607def list_stack_instances(req_params):608 state = CloudFormationRegion.get()609 set_name = req_params.get("StackSetName")610 stack_set = [sset for sset in state.stack_sets.values() if sset.stack_set_name == set_name]611 if not stack_set:612 return not_found_error('Stack set named "%s" does not exist' % set_name)613 stack_set = stack_set[0]614 result = [inst.metadata for inst in stack_set.stack_instances]615 result = {"Summaries": result}616 return result617ChangeSetTypes = Literal["CREATE", "UPDATE", "IMPORT"]618def create_change_set(req_params: Dict[str, Any]):619 change_set_type: ChangeSetTypes = req_params.get("ChangeSetType", "UPDATE")620 stack_name: Optional[str] = req_params.get("StackName")621 change_set_name: Optional[str] = req_params.get("ChangeSetName")622 template_body: Optional[str] = req_params.get("TemplateBody")623 # s3 or secretsmanager url624 template_url: Optional[str] = req_params.get("TemplateUrl") or req_params.get("TemplateURL")625 if is_none_or_empty(change_set_name):626 return error_response(627 "ChangeSetName required", 400, "ValidationError"628 ) # TODO: check proper message629 if is_none_or_empty(stack_name):630 return error_response(631 "StackName required", 400, "ValidationError"632 ) # TODO: check proper message633 stack: Optional[Stack] = find_stack(stack_name)634 # validate and resolve template635 if template_body and template_url:636 return error_response(637 "Specify exactly one of 'TemplateBody' or 'TemplateUrl'", 400, "ValidationError"638 ) # TODO: check proper message639 if not template_body and not template_url:640 return error_response(641 "Specify exactly one of 'TemplateBody' or 'TemplateUrl'", 400, "ValidationError"642 ) # TODO: check proper message643 prepare_template_body(req_params) # TODO: function has too many unclear responsibilities644 template = template_preparer.parse_template(req_params["TemplateBody"])645 del req_params["TemplateBody"] # TODO: stop mutating req_params646 template["StackName"] = stack_name647 template[648 "ChangeSetName"649 ] = change_set_name # TODO: validate with AWS what this is actually doing?650 if change_set_type == "UPDATE":651 # add changeset to existing stack652 if stack is None:653 return error_response(654 f"Stack '{stack_name}' does not exist.", 400, "ValidationError"655 ) # stack should exist already656 elif change_set_type == "CREATE":657 # create new (empty) stack658 if stack is not None:659 return error_response(660 f"Stack {stack_name} already exists", 400, "ValidationError"661 ) # stack should not exist yet (TODO: check proper message)662 state = CloudFormationRegion.get()663 empty_stack_template = dict(template)664 empty_stack_template["Resources"] = {}665 req_params_copy = clone_stack_params(req_params)666 stack = Stack(req_params_copy, empty_stack_template)667 state.stacks[stack.stack_id] = stack668 stack.set_stack_status("REVIEW_IN_PROGRESS")669 elif change_set_type == "IMPORT":670 raise NotImplementedError() # TODO: implement importing resources671 else:672 msg = f"1 validation error detected: Value '{change_set_type}' at 'changeSetType' failed to satisfy constraint: Member must satisfy enum value set: [IMPORT, UPDATE, CREATE]"673 return error_response(msg, code=400, code_string="ValidationError")674 change_set = StackChangeSet(req_params, template)675 # TODO: refactor the flow here676 deployer = template_deployer.TemplateDeployer(change_set)677 deployer.construct_changes(678 stack,679 change_set,680 change_set_id=change_set.change_set_id,681 append_to_changeset=True,682 ) # TODO: ignores return value (?)683 deployer.apply_parameter_changes(change_set, change_set) # TODO: bandaid to populate metadata684 stack.change_sets.append(change_set)685 change_set.metadata[686 "Status"687 ] = "CREATE_COMPLETE" # technically for some time this should first be CREATE_PENDING688 change_set.metadata[689 "ExecutionStatus"690 ] = "AVAILABLE" # technically for some time this should first be UNAVAILABLE691 return {"StackId": change_set.stack_id, "Id": change_set.change_set_id}692def execute_change_set(req_params):693 stack_name = req_params.get("StackName")694 cs_name = req_params.get("ChangeSetName")695 change_set = find_change_set(cs_name, stack_name=stack_name)696 if not change_set:697 return not_found_error(698 'Unable to find change set "%s" for stack "%s"' % (cs_name, stack_name)699 )700 LOG.debug(701 'Executing change set "%s" for stack "%s" with %s resources ...'702 % (cs_name, stack_name, len(change_set.template_resources))703 )704 deployer = template_deployer.TemplateDeployer(change_set.stack)705 deployer.apply_change_set(change_set)706 change_set.stack.metadata["ChangeSetId"] = change_set.change_set_id707 return {}708def list_change_sets(req_params):709 stack_name = req_params.get("StackName")710 stack = find_stack(stack_name)711 if not stack:712 return not_found_error('Unable to find stack "%s"' % stack_name)713 result = [cs.metadata for cs in stack.change_sets]714 result = {"Summaries": result}715 return result716def list_stack_sets(req_params):717 state = CloudFormationRegion.get()718 result = [sset.metadata for sset in state.stack_sets.values()]719 result = {"Summaries": result}720 return result721def describe_change_set(req_params):722 stack_name = req_params.get("StackName")723 cs_name = req_params.get("ChangeSetName")724 change_set: Optional[StackChangeSet] = find_change_set(cs_name, stack_name=stack_name)725 if not change_set:726 return not_found_error(727 'Unable to find change set "%s" for stack "%s"' % (cs_name, stack_name)728 )729 return change_set.metadata730def describe_stack_set(req_params):731 state = CloudFormationRegion.get()732 set_name = req_params.get("StackSetName")733 result = [734 sset.metadata for sset in state.stack_sets.values() if sset.stack_set_name == set_name735 ]736 if not result:737 return not_found_error('Unable to find stack set "%s"' % set_name)738 result = {"StackSet": result[0]}739 return result740def describe_stack_set_operation(req_params):741 state = CloudFormationRegion.get()742 set_name = req_params.get("StackSetName")743 stack_set = [sset for sset in state.stack_sets.values() if sset.stack_set_name == set_name]744 if not stack_set:745 return not_found_error('Unable to find stack set "%s"' % set_name)746 stack_set = stack_set[0]747 op_id = req_params.get("OperationId")748 result = stack_set.operations.get(op_id)749 if not result:750 LOG.debug(751 'Unable to find operation ID "%s" for stack set "%s" in list: %s'752 % (op_id, set_name, list(stack_set.operations.keys()))753 )754 return not_found_error(755 'Unable to find operation ID "%s" for stack set "%s"' % (op_id, set_name)756 )757 result = {"StackSetOperation": result}758 return result759def list_exports(req_params):760 state = CloudFormationRegion.get()761 result = {"Exports": state.exports}762 return result763def list_imports(req_params):764 state = CloudFormationRegion.get()765 export_name = req_params.get("ExportName")766 importing_stack_names = []767 for stack in state.stacks.values():768 if export_name in stack.imports:769 importing_stack_names.append(stack.stack_name)770 result = {"Imports": importing_stack_names}771 return result772def validate_template(req_params):773 try:774 result = template_preparer.validate_template(req_params)775 result = "<tmp>%s</tmp>" % result776 result = xmltodict.parse(result)["tmp"]777 return result778 except Exception as err:779 return error_response("Template Validation Error: %s" % err)780def describe_stack_events(req_params):781 stack_name = req_params.get("StackName")782 state = CloudFormationRegion.get()783 events = []784 for stack_id, stack in state.stacks.items():785 if stack_name in [None, stack.stack_name, stack.stack_id]:786 events.extend(stack.events)787 return {"StackEvents": events}788def delete_change_set(req_params):789 stack_name = req_params.get("StackName")790 cs_name = req_params.get("ChangeSetName")791 change_set = find_change_set(cs_name, stack_name=stack_name)792 if not change_set:793 return not_found_error(794 'Unable to find change set "%s" for stack "%s"' % (cs_name, stack_name)795 )796 change_set.stack.change_sets = [797 cs for cs in change_set.stack.change_sets if cs.change_set_name != cs_name798 ]799 return {}800def get_template(req_params):801 stack_name = req_params.get("StackName")802 cs_name = req_params.get("ChangeSetName")803 stack = find_stack(stack_name)804 if cs_name:805 stack = find_change_set(stack_name, cs_name)806 if not stack:807 return stack_not_found_error(stack_name)808 result = {"TemplateBody": json.dumps(stack._template_raw)}809 return result810def get_template_summary(req_params):811 stack_name = req_params.get("StackName")812 stack = None813 if stack_name:814 stack = find_stack(stack_name)815 if not stack:816 return stack_not_found_error(stack_name)817 else:818 template_deployer.prepare_template_body(req_params)819 template = template_preparer.parse_template(req_params["TemplateBody"])820 req_params["StackName"] = "tmp-stack"821 stack = Stack(req_params, template)822 result = stack.describe_details()823 id_summaries = {}824 for resource_id, resource in stack.template_resources.items():825 res_type = resource["Type"]826 id_summaries[res_type] = id_summaries.get(res_type) or []827 id_summaries[res_type].append(resource_id)828 result["ResourceTypes"] = list(id_summaries.keys())829 result["ResourceIdentifierSummaries"] = [830 {"ResourceType": key, "LogicalResourceIds": values} for key, values in id_summaries.items()831 ]832 return result...
template.py
Source:template.py
...57 return {58 "index.lifecycle.name": lifecycle_policy_name,59 "index.lifecycle.rollover_alias": index_alias60 }61def prepare_template_body(index_alias, lifecycle_policy_name="default", meta=None):62 return {63 "index_patterns": [64 f"{index_alias}*"65 ],66 "template": {67 "settings": {68 "index": {69 "number_of_shards": "1",70 "number_of_replicas": "0",71 "refresh_interval": settings.LEEK_ES_DEFAULT_REFRESH_INTERVAL72 },73 **get_im_settings(index_alias, lifecycle_policy_name),74 },75 "mappings": {76 "_source": {77 "enabled": True78 },79 "_meta": meta,80 "dynamic": False,81 "properties": properties82 },83 }84 }85def create_index_template(index_alias, lifecycle_policy_name="default", meta=None):86 """87 This is considered as an organization project88 An organization can have multiple applications(templates)89 Each day events will be sent to a new index orgName-appName-2020-08-2490 The newly created index will be assigned the template if index name matches index_patterns91 Each day indexes older than 14 days will be deleted using curator92 :param lifecycle_policy_name: Index Lifecycle Policy Name93 :param meta: application level settings94 :param index_alias: index alias in the form of orgName-appName95 """96 connection = es.connection97 body = prepare_template_body(index_alias, lifecycle_policy_name=lifecycle_policy_name, meta=meta)98 try:99 connection.indices.put_index_template(name=index_alias, body=body, create=True)100 # Bootstrap first index101 connection.indices.create(f"{index_alias}-000001", body={102 "aliases": {103 index_alias: {104 "is_write_index": True105 }106 }107 })108 return meta, 201109 except es_exceptions.ConnectionError:110 return responses.search_backend_unavailable111 except es_exceptions.RequestError as e:...
cfn.py
Source:cfn.py
...70 def prepare_change_sets(self, target):71 for stack_name, stack in target.stacks.items():72 if stack.change_set is None:73 continue74 template_body = self.prepare_template_body(stack)75 stack.change_set = self.prepare_change_set(76 stack, stack.change_set.type, template_body)77 def analyse_single_stack(self, stack):78 deployed = self.deployed_stacks.get(stack.name)79 if (not deployed80 or deployed.status == 'REVIEW_IN_PROGRESS'):81 return ChangeSet(ChangeSetType.CREATE, stack.name)82 if deployed.status == 'ROLLBACK_COMPLETE':83 raise ValidationError(84 f'Stack {stack.name} with ROLLBACK_COMPLETE status needs to be deleted '85 f'before it can be recreated.')86 deployed.is_outdated = (87 self.get_content_hash(stack) != deployed.content_hash)88 if deployed.is_outdated:89 return ChangeSet(ChangeSetType.UPDATE, stack.name)90 def prepare_template_body(self, stack):91 return loader.dump_yaml(92 deep_merge(93 dict(Parameters={self.metadata_parameter: dict(Type='String')}),94 stack.template),95 stream=None)96 def validate_template_body(self, stack, template_body):97 v = self.cfn.validate_template(TemplateBody=template_body)98 for cap in v.get('Capabilities', []):99 if cap not in stack.capabilities:100 reason = v.get('CapabilitiesReason', '(no reason provided)')101 raise ValidationError(102 'Required capability, {}, is missing in stack {}: {}'103 .format(cap, stack.name, reason))104 def prepare_change_set(self, stack, change_set_type, template_body):105 content_hash = self.get_content_hash(stack)106 parameters = [dict(107 ParameterKey=self.metadata_parameter,108 ParameterValue=content_hash + self.metadata_suffix)]109 parameters.extend(110 dict(ParameterKey=k, ParameterValue=v)111 for k, v in stack.parameters.items())112 tags = [dict(Key=k, Value=v) for k, v in stack.tags.items()]113 change_set = self.cfn.create_change_set(114 StackName=stack.name,115 TemplateBody=template_body,116 Capabilities=stack.capabilities,117 ChangeSetType=change_set_type.value,118 ChangeSetName=content_hash,119 Parameters=parameters,120 Tags=tags)121 stack_id = change_set['StackId']122 change_set_id = change_set['Id']123 return ChangeSet(change_set_type, stack_id, change_set_id)124 def analyse_target(self, target):125 result = TargetAnalysisResults()126 for stack_name, stack in target.stacks.items():127 stack.change_set = self.analyse_single_stack(stack)128 if stack.change_set is None:129 continue130 if stack.change_set.type == ChangeSetType.CREATE:131 result.stack_summary.total += 1132 result.stack_summary.new += 1133 elif stack.change_set.type == ChangeSetType.UPDATE:134 result.stack_summary.updated += 1135 template_body = self.prepare_template_body(stack)136 self.validate_template_body(stack, template_body)137 for stack_name, stack in self.deployed_stacks.items():138 if stack.status == 'REVIEW_IN_PROGRESS':139 # Not yet a stack, counted as NEW if a CREATE change set was created.140 continue141 if stack.status == 'ROLLBACK_COMPLETE':142 # Not a stack, creation failed. Should be deleted.143 result.failed_stacks += [stack_name]144 continue145 result.stack_summary.total += 1146 if hasattr(stack, 'is_outdated'):147 # Managed (or now adopted) stack148 if not stack.content_hash:149 result.stack_summary.adopted += 1...
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?
