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.