0e75e353-4494-4b65-a715-a78cf84b0b94

Committed 02 Apr 2025 01:40PM UTC coverage: 86.857% (+0.05%) from 86.807%

Build # 0e75e353-4494-4b65-a715-a78cf84b0b94

Build Type

push

circleci

Committed by

web-flow

Commit Message

CFn: WIP POC v2 executor (#12396)

Co-authored-by: Marco Edoardo Palma <64580864+MEPalma@users.noreply.github.com>

Coverage Stats

36 of 112 new or added lines in 6 files covered. (32.14%)

144 existing lines in 9 files now uncovered.

63556 of 73173 relevant lines covered (86.86%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

19.53

/localstack-core/localstack/services/cloudformation/v2/provider.py

import logging
from copy import deepcopy

from localstack.aws.api import RequestContext, handler
from localstack.aws.api.cloudformation import (
    ChangeSetNameOrId,
    ChangeSetNotFoundException,
    ChangeSetType,
    ClientRequestToken,
    CreateChangeSetInput,
    CreateChangeSetOutput,
    DescribeChangeSetOutput,
    DisableRollback,
    ExecuteChangeSetOutput,
    ExecutionStatus,
    IncludePropertyValues,
    InvalidChangeSetStatusException,
    NextToken,
    Parameter,
    RetainExceptOnCreate,
    StackNameOrId,
    StackStatus,
)
from localstack.services.cloudformation import api_utils
from localstack.services.cloudformation.engine import parameters as param_resolver
from localstack.services.cloudformation.engine import template_deployer, template_preparer
from localstack.services.cloudformation.engine.entities import Stack, StackChangeSet
from localstack.services.cloudformation.engine.parameters import mask_no_echo, strip_parameter_type
from localstack.services.cloudformation.engine.resource_ordering import (
    NoResourceInStack,
    order_resources,
)
from localstack.services.cloudformation.engine.template_utils import resolve_stack_conditions
from localstack.services.cloudformation.engine.v2.change_set_model_describer import (
    ChangeSetModelDescriber,
)
from localstack.services.cloudformation.engine.v2.change_set_model_executor import (
    ChangeSetModelExecutor,
)
from localstack.services.cloudformation.engine.validations import ValidationError
from localstack.services.cloudformation.provider import (
    ARN_CHANGESET_REGEX,
    ARN_STACK_REGEX,
    CloudformationProvider,
    clone_stack_params,
)
from localstack.services.cloudformation.stores import (
    find_change_set,
    find_stack,
    get_cloudformation_store,
)
from localstack.utils.collections import remove_attributes

LOG = logging.getLogger(__name__)


class CloudformationProviderV2(CloudformationProvider):
    @handler("CreateChangeSet", expand=False)
    def create_change_set(
        self, context: RequestContext, request: CreateChangeSetInput
    ) -> CreateChangeSetOutput:
        state = get_cloudformation_store(context.account_id, context.region)

        req_params = request
        change_set_type = req_params.get("ChangeSetType", "UPDATE")
        stack_name = req_params.get("StackName")
        change_set_name = req_params.get("ChangeSetName")
        template_body = req_params.get("TemplateBody")
        # s3 or secretsmanager url
        template_url = req_params.get("TemplateURL")

        # validate and resolve template
        if template_body and template_url:
            raise ValidationError(
                "Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
            )  # TODO: check proper message

        if not template_body and not template_url:
            raise ValidationError(
                "Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
            )  # TODO: check proper message

        api_utils.prepare_template_body(
            req_params
        )  # TODO: function has too many unclear responsibilities
        if not template_body:
            template_body = req_params[
                "TemplateBody"
            ]  # should then have been set by prepare_template_body
        template = template_preparer.parse_template(req_params["TemplateBody"])

        del req_params["TemplateBody"]  # TODO: stop mutating req_params
        template["StackName"] = stack_name
        # TODO: validate with AWS what this is actually doing?
        template["ChangeSetName"] = change_set_name

        # this is intentionally not in a util yet. Let's first see how the different operations deal with these before generalizing
        # handle ARN stack_name here (not valid for initial CREATE, since stack doesn't exist yet)
        if ARN_STACK_REGEX.match(stack_name):
            if not (stack := state.stacks.get(stack_name)):
                raise ValidationError(f"Stack '{stack_name}' does not exist.")
        else:
            # stack name specified, so fetch the stack by name
            stack_candidates: list[Stack] = [
                s for stack_arn, s in state.stacks.items() if s.stack_name == stack_name
            ]
            active_stack_candidates = [
                s for s in stack_candidates if self._stack_status_is_active(s.status)
            ]

            # on a CREATE an empty Stack should be generated if we didn't find an active one
            if not active_stack_candidates and change_set_type == ChangeSetType.CREATE:
                empty_stack_template = dict(template)
                empty_stack_template["Resources"] = {}
                req_params_copy = clone_stack_params(req_params)
                stack = Stack(
                    context.account_id,
                    context.region,
                    req_params_copy,
                    empty_stack_template,
                    template_body=template_body,
                )
                state.stacks[stack.stack_id] = stack
                stack.set_stack_status("REVIEW_IN_PROGRESS")
            else:
                if not active_stack_candidates:
                    raise ValidationError(f"Stack '{stack_name}' does not exist.")
                stack = active_stack_candidates[0]

        # TODO: test if rollback status is allowed as well
        if (
            change_set_type == ChangeSetType.CREATE
            and stack.status != StackStatus.REVIEW_IN_PROGRESS
        ):
            raise ValidationError(
                f"Stack [{stack_name}] already exists and cannot be created again with the changeSet [{change_set_name}]."
            )

        old_parameters: dict[str, Parameter] = {}
        match change_set_type:
            case ChangeSetType.UPDATE:
                # add changeset to existing stack
                old_parameters = {
                    k: mask_no_echo(strip_parameter_type(v))
                    for k, v in stack.resolved_parameters.items()
                }
            case ChangeSetType.IMPORT:
                raise NotImplementedError()  # TODO: implement importing resources
            case ChangeSetType.CREATE:
                pass
            case _:
                msg = (
                    f"1 validation error detected: Value '{change_set_type}' at 'changeSetType' failed to satisfy "
                    f"constraint: Member must satisfy enum value set: [IMPORT, UPDATE, CREATE] "
                )
                raise ValidationError(msg)

        # resolve parameters
        new_parameters: dict[str, Parameter] = param_resolver.convert_stack_parameters_to_dict(
            request.get("Parameters")
        )
        parameter_declarations = param_resolver.extract_stack_parameter_declarations(template)
        resolved_parameters = param_resolver.resolve_parameters(
            account_id=context.account_id,
            region_name=context.region,
            parameter_declarations=parameter_declarations,
            new_parameters=new_parameters,
            old_parameters=old_parameters,
        )

        # TODO: remove this when fixing Stack.resources and transformation order
        #   currently we need to create a stack with existing resources + parameters so that resolve refs recursively in here will work.
        #   The correct way to do it would be at a later stage anyway just like a normal intrinsic function
        req_params_copy = clone_stack_params(req_params)
        temp_stack = Stack(context.account_id, context.region, req_params_copy, template)
        temp_stack.set_resolved_parameters(resolved_parameters)

        # TODO: everything below should be async
        # apply template transformations
        transformed_template = template_preparer.transform_template(
            context.account_id,
            context.region,
            template,
            stack_name=temp_stack.stack_name,
            resources=temp_stack.resources,
            mappings=temp_stack.mappings,
            conditions={},  # TODO: we don't have any resolved conditions yet at this point but we need the conditions because of the samtranslator...
            resolved_parameters=resolved_parameters,
        )

        # create change set for the stack and apply changes
        change_set = StackChangeSet(
            context.account_id,
            context.region,
            stack,
            req_params,
            transformed_template,
            change_set_type=change_set_type,
        )
        # only set parameters for the changeset, then switch to stack on execute_change_set
        change_set.template_body = template_body
        change_set.populate_update_graph(stack.template, transformed_template)

        # TODO: evaluate conditions
        raw_conditions = transformed_template.get("Conditions", {})
        resolved_stack_conditions = resolve_stack_conditions(
            account_id=context.account_id,
            region_name=context.region,
            conditions=raw_conditions,
            parameters=resolved_parameters,
            mappings=temp_stack.mappings,
            stack_name=stack_name,
        )
        change_set.set_resolved_stack_conditions(resolved_stack_conditions)

        # a bit gross but use the template ordering to validate missing resources
        try:
            order_resources(
                transformed_template["Resources"],
                resolved_parameters=resolved_parameters,
                resolved_conditions=resolved_stack_conditions,
            )
        except NoResourceInStack as e:
            raise ValidationError(str(e)) from e

        deployer = template_deployer.TemplateDeployer(
            context.account_id, context.region, change_set
        )
        changes = deployer.construct_changes(
            stack,
            change_set,
            change_set_id=change_set.change_set_id,
            append_to_changeset=True,
            filter_unchanged_resources=True,
        )
        stack.change_sets.append(change_set)
        if not changes:
            change_set.metadata["Status"] = "FAILED"
            change_set.metadata["ExecutionStatus"] = "UNAVAILABLE"
            change_set.metadata["StatusReason"] = (
                "The submitted information didn't contain changes. Submit different information to create a change set."
            )
        else:
            change_set.metadata["Status"] = (
                "CREATE_COMPLETE"  # technically for some time this should first be CREATE_PENDING
            )
            change_set.metadata["ExecutionStatus"] = (
                "AVAILABLE"  # technically for some time this should first be UNAVAILABLE
            )

        return CreateChangeSetOutput(StackId=change_set.stack_id, Id=change_set.change_set_id)

    @handler("ExecuteChangeSet")
    def execute_change_set(
        self,
        context: RequestContext,
        change_set_name: ChangeSetNameOrId,
        stack_name: StackNameOrId | None = None,
        client_request_token: ClientRequestToken | None = None,
        disable_rollback: DisableRollback | None = None,
        retain_except_on_create: RetainExceptOnCreate | None = None,
        **kwargs,
    ) -> ExecuteChangeSetOutput:
        change_set = find_change_set(
            context.account_id,
            context.region,
            change_set_name,
            stack_name=stack_name,
            active_only=True,
        )
        if not change_set:
            raise ChangeSetNotFoundException(f"ChangeSet [{change_set_name}] does not exist")
        if change_set.metadata.get("ExecutionStatus") != ExecutionStatus.AVAILABLE:
            LOG.debug("Change set %s not in execution status 'AVAILABLE'", change_set_name)
            raise InvalidChangeSetStatusException(
                f"ChangeSet [{change_set.metadata['ChangeSetId']}] cannot be executed in its current status of [{change_set.metadata.get('Status')}]"
            )
        stack_name = change_set.stack.stack_name
        LOG.debug(
            'Executing change set "%s" for stack "%s" with %s resources ...',
            change_set_name,
            stack_name,
            len(change_set.template_resources),
        )
        if not change_set.update_graph:
            raise RuntimeError("Programming error: no update graph found for change set")

        change_set_executor = ChangeSetModelExecutor(
            change_set.update_graph,
            account_id=context.account_id,
            region=context.region,
            stack_name=change_set.stack.stack_name,
            stack_id=change_set.stack.stack_id,
        )
        new_resources = change_set_executor.execute()
        change_set.stack.set_stack_status(f"{change_set.change_set_type or 'UPDATE'}_COMPLETE")
        change_set.stack.resources = new_resources
        return ExecuteChangeSetOutput()

    @handler("DescribeChangeSet")
    def describe_change_set(
        self,
        context: RequestContext,
        change_set_name: ChangeSetNameOrId,
        stack_name: StackNameOrId | None = None,
        next_token: NextToken | None = None,
        include_property_values: IncludePropertyValues | None = None,
        **kwargs,
    ) -> DescribeChangeSetOutput:
        # TODO add support for include_property_values
        # only relevant if change_set_name isn't an ARN
        if not ARN_CHANGESET_REGEX.match(change_set_name):
            if not stack_name:
                raise ValidationError(
                    "StackName must be specified if ChangeSetName is not specified as an ARN."
                )

            stack = find_stack(context.account_id, context.region, stack_name)
            if not stack:
                raise ValidationError(f"Stack [{stack_name}] does not exist")

        change_set = find_change_set(
            context.account_id, context.region, change_set_name, stack_name=stack_name
        )
        if not change_set:
            raise ChangeSetNotFoundException(f"ChangeSet [{change_set_name}] does not exist")

        change_set_describer = ChangeSetModelDescriber(
            node_template=change_set.update_graph, include_property_values=include_property_values
        )
        resource_changes = change_set_describer.get_changes()

        attrs = [
            "ChangeSetType",
            "StackStatus",
            "LastUpdatedTime",
            "DisableRollback",
            "EnableTerminationProtection",
            "Transform",
        ]
        result = remove_attributes(deepcopy(change_set.metadata), attrs)
        # TODO: replace this patch with a better solution
        result["Parameters"] = [
            mask_no_echo(strip_parameter_type(p)) for p in result.get("Parameters", [])
        ]
        result["Changes"] = resource_changes
        return result

1	import logging	1✔
2	from copy import deepcopy	1✔
3
4	from localstack.aws.api import RequestContext, handler	1✔
5	from localstack.aws.api.cloudformation import (	1✔
6	ChangeSetNameOrId,
7	ChangeSetNotFoundException,
8	ChangeSetType,
9	ClientRequestToken,
10	CreateChangeSetInput,
11	CreateChangeSetOutput,
12	DescribeChangeSetOutput,
13	DisableRollback,
14	ExecuteChangeSetOutput,
15	ExecutionStatus,
16	IncludePropertyValues,
17	InvalidChangeSetStatusException,
18	NextToken,
19	Parameter,
20	RetainExceptOnCreate,
21	StackNameOrId,
22	StackStatus,
23	)
24	from localstack.services.cloudformation import api_utils	1✔
25	from localstack.services.cloudformation.engine import parameters as param_resolver	1✔
26	from localstack.services.cloudformation.engine import template_deployer, template_preparer	1✔
27	from localstack.services.cloudformation.engine.entities import Stack, StackChangeSet	1✔
28	from localstack.services.cloudformation.engine.parameters import mask_no_echo, strip_parameter_type	1✔
29	from localstack.services.cloudformation.engine.resource_ordering import (	1✔
30	NoResourceInStack,
31	order_resources,
32	)
33	from localstack.services.cloudformation.engine.template_utils import resolve_stack_conditions	1✔
34	from localstack.services.cloudformation.engine.v2.change_set_model_describer import (	1✔
35	ChangeSetModelDescriber,
36	)
37	from localstack.services.cloudformation.engine.v2.change_set_model_executor import (	1✔
38	ChangeSetModelExecutor,
39	)
40	from localstack.services.cloudformation.engine.validations import ValidationError	1✔
41	from localstack.services.cloudformation.provider import (	1✔
42	ARN_CHANGESET_REGEX,
43	ARN_STACK_REGEX,
44	CloudformationProvider,
45	clone_stack_params,
46	)
47	from localstack.services.cloudformation.stores import (	1✔
48	find_change_set,
49	find_stack,
50	get_cloudformation_store,
51	)
52	from localstack.utils.collections import remove_attributes	1✔
53
54	LOG = logging.getLogger(__name__)	1✔
55
56
57	class CloudformationProviderV2(CloudformationProvider):	1✔
58	@handler("CreateChangeSet", expand=False)	1✔
59	def create_change_set(	1✔
60	self, context: RequestContext, request: CreateChangeSetInput
61	) -> CreateChangeSetOutput:
UNCOV 62	state = get_cloudformation_store(context.account_id, context.region)	×
63
UNCOV 64	req_params = request	×
UNCOV 65	change_set_type = req_params.get("ChangeSetType", "UPDATE")	×
UNCOV 66	stack_name = req_params.get("StackName")	×
UNCOV 67	change_set_name = req_params.get("ChangeSetName")	×
UNCOV 68	template_body = req_params.get("TemplateBody")	×
69	# s3 or secretsmanager url
UNCOV 70	template_url = req_params.get("TemplateURL")	×
71
72	# validate and resolve template
UNCOV 73	if template_body and template_url:	×
74	raise ValidationError(	×
75	"Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
76	) # TODO: check proper message
77
78	if not template_body and not template_url:	×
79	raise ValidationError(	×
80	"Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
81	) # TODO: check proper message
82
UNCOV 83	api_utils.prepare_template_body(	×
84	req_params
85	) # TODO: function has too many unclear responsibilities
86	if not template_body:	×
UNCOV 87	template_body = req_params[	×
88	"TemplateBody"
89	] # should then have been set by prepare_template_body
90	template = template_preparer.parse_template(req_params["TemplateBody"])	×
91
UNCOV 92	del req_params["TemplateBody"] # TODO: stop mutating req_params	×
UNCOV 93	template["StackName"] = stack_name	×
94	# TODO: validate with AWS what this is actually doing?
95	template["ChangeSetName"] = change_set_name	×
96
97	# this is intentionally not in a util yet. Let's first see how the different operations deal with these before generalizing
98	# handle ARN stack_name here (not valid for initial CREATE, since stack doesn't exist yet)
99	if ARN_STACK_REGEX.match(stack_name):	×
UNCOV 100	if not (stack := state.stacks.get(stack_name)):	×
UNCOV 101	raise ValidationError(f"Stack '{stack_name}' does not exist.")	×
102	else:
103	# stack name specified, so fetch the stack by name
104	stack_candidates: list[Stack] = [	×
105	s for stack_arn, s in state.stacks.items() if s.stack_name == stack_name
106	]
107	active_stack_candidates = [	×
108	s for s in stack_candidates if self._stack_status_is_active(s.status)
109	]
110
111	# on a CREATE an empty Stack should be generated if we didn't find an active one
112	if not active_stack_candidates and change_set_type == ChangeSetType.CREATE:	×
113	empty_stack_template = dict(template)	×
UNCOV 114	empty_stack_template["Resources"] = {}	×
UNCOV 115	req_params_copy = clone_stack_params(req_params)	×
116	stack = Stack(	×
117	context.account_id,
118	context.region,
119	req_params_copy,
120	empty_stack_template,
121	template_body=template_body,
122	)
UNCOV 123	state.stacks[stack.stack_id] = stack	×
124	stack.set_stack_status("REVIEW_IN_PROGRESS")	×
125	else:
126	if not active_stack_candidates:	×
127	raise ValidationError(f"Stack '{stack_name}' does not exist.")	×
128	stack = active_stack_candidates[0]	×
129
130	# TODO: test if rollback status is allowed as well
UNCOV 131	if (	×
132	change_set_type == ChangeSetType.CREATE
133	and stack.status != StackStatus.REVIEW_IN_PROGRESS
134	):
135	raise ValidationError(	×
136	f"Stack [{stack_name}] already exists and cannot be created again with the changeSet [{change_set_name}]."
137	)
138
139	old_parameters: dict[str, Parameter] = {}	×
140	match change_set_type:	×
UNCOV 141	case ChangeSetType.UPDATE:	×
142	# add changeset to existing stack
143	old_parameters = {	×
144	k: mask_no_echo(strip_parameter_type(v))
145	for k, v in stack.resolved_parameters.items()
146	}
147	case ChangeSetType.IMPORT:	×
148	raise NotImplementedError() # TODO: implement importing resources
UNCOV 149	case ChangeSetType.CREATE:	×
UNCOV 150	pass	×
151	case _:	×
152	msg = (	×
153	f"1 validation error detected: Value '{change_set_type}' at 'changeSetType' failed to satisfy "
154	f"constraint: Member must satisfy enum value set: [IMPORT, UPDATE, CREATE] "
155	)
UNCOV 156	raise ValidationError(msg)	×
157
158	# resolve parameters
159	new_parameters: dict[str, Parameter] = param_resolver.convert_stack_parameters_to_dict(	×
160	request.get("Parameters")
161	)
162	parameter_declarations = param_resolver.extract_stack_parameter_declarations(template)	×
163	resolved_parameters = param_resolver.resolve_parameters(	×
164	account_id=context.account_id,
165	region_name=context.region,
166	parameter_declarations=parameter_declarations,
167	new_parameters=new_parameters,
168	old_parameters=old_parameters,
169	)
170
171	# TODO: remove this when fixing Stack.resources and transformation order
172	# currently we need to create a stack with existing resources + parameters so that resolve refs recursively in here will work.
173	# The correct way to do it would be at a later stage anyway just like a normal intrinsic function
174	req_params_copy = clone_stack_params(req_params)	×
175	temp_stack = Stack(context.account_id, context.region, req_params_copy, template)	×
UNCOV 176	temp_stack.set_resolved_parameters(resolved_parameters)	×
177
178	# TODO: everything below should be async
179	# apply template transformations
UNCOV 180	transformed_template = template_preparer.transform_template(	×
181	context.account_id,
182	context.region,
183	template,
184	stack_name=temp_stack.stack_name,
185	resources=temp_stack.resources,
186	mappings=temp_stack.mappings,
187	conditions={}, # TODO: we don't have any resolved conditions yet at this point but we need the conditions because of the samtranslator...
188	resolved_parameters=resolved_parameters,
189	)
190
191	# create change set for the stack and apply changes
192	change_set = StackChangeSet(	×
193	context.account_id,
194	context.region,
195	stack,
196	req_params,
197	transformed_template,
198	change_set_type=change_set_type,
199	)
200	# only set parameters for the changeset, then switch to stack on execute_change_set
UNCOV 201	change_set.template_body = template_body	×
UNCOV 202	change_set.populate_update_graph(stack.template, transformed_template)	×
203
204	# TODO: evaluate conditions
UNCOV 205	raw_conditions = transformed_template.get("Conditions", {})	×
UNCOV 206	resolved_stack_conditions = resolve_stack_conditions(	×
207	account_id=context.account_id,
208	region_name=context.region,
209	conditions=raw_conditions,
210	parameters=resolved_parameters,
211	mappings=temp_stack.mappings,
212	stack_name=stack_name,
213	)
UNCOV 214	change_set.set_resolved_stack_conditions(resolved_stack_conditions)	×
215
216	# a bit gross but use the template ordering to validate missing resources
UNCOV 217	try:	×
218	order_resources(	×
219	transformed_template["Resources"],
220	resolved_parameters=resolved_parameters,
221	resolved_conditions=resolved_stack_conditions,
222	)
223	except NoResourceInStack as e:	×
UNCOV 224	raise ValidationError(str(e)) from e	×
225
UNCOV 226	deployer = template_deployer.TemplateDeployer(	×
227	context.account_id, context.region, change_set
228	)
UNCOV 229	changes = deployer.construct_changes(	×
230	stack,
231	change_set,
232	change_set_id=change_set.change_set_id,
233	append_to_changeset=True,
234	filter_unchanged_resources=True,
235	)
UNCOV 236	stack.change_sets.append(change_set)	×
UNCOV 237	if not changes:	×
UNCOV 238	change_set.metadata["Status"] = "FAILED"	×
UNCOV 239	change_set.metadata["ExecutionStatus"] = "UNAVAILABLE"	×
240	change_set.metadata["StatusReason"] = (	×
241	"The submitted information didn't contain changes. Submit different information to create a change set."
242	)
243	else:
UNCOV 244	change_set.metadata["Status"] = (	×
245	"CREATE_COMPLETE" # technically for some time this should first be CREATE_PENDING
246	)
UNCOV 247	change_set.metadata["ExecutionStatus"] = (	×
248	"AVAILABLE" # technically for some time this should first be UNAVAILABLE
249	)
250
UNCOV 251	return CreateChangeSetOutput(StackId=change_set.stack_id, Id=change_set.change_set_id)	×
252
253	@handler("ExecuteChangeSet")	1✔
254	def execute_change_set(	1✔
255	self,
256	context: RequestContext,
257	change_set_name: ChangeSetNameOrId,
258	stack_name: StackNameOrId \| None = None,
259	client_request_token: ClientRequestToken \| None = None,
260	disable_rollback: DisableRollback \| None = None,
261	retain_except_on_create: RetainExceptOnCreate \| None = None,
262	**kwargs,
263	) -> ExecuteChangeSetOutput:
NEW 264	change_set = find_change_set(	×
265	context.account_id,
266	context.region,
267	change_set_name,
268	stack_name=stack_name,
269	active_only=True,
270	)
NEW 271	if not change_set:	×
NEW 272	raise ChangeSetNotFoundException(f"ChangeSet [{change_set_name}] does not exist")	×
NEW 273	if change_set.metadata.get("ExecutionStatus") != ExecutionStatus.AVAILABLE:	×
NEW 274	LOG.debug("Change set %s not in execution status 'AVAILABLE'", change_set_name)	×
NEW 275	raise InvalidChangeSetStatusException(	×
276	f"ChangeSet [{change_set.metadata['ChangeSetId']}] cannot be executed in its current status of [{change_set.metadata.get('Status')}]"
277	)
NEW 278	stack_name = change_set.stack.stack_name	×
NEW 279	LOG.debug(	×
280	'Executing change set "%s" for stack "%s" with %s resources ...',
281	change_set_name,
282	stack_name,
283	len(change_set.template_resources),
284	)
NEW 285	if not change_set.update_graph:	×
NEW 286	raise RuntimeError("Programming error: no update graph found for change set")	×
287
NEW 288	change_set_executor = ChangeSetModelExecutor(	×
289	change_set.update_graph,
290	account_id=context.account_id,
291	region=context.region,
292	stack_name=change_set.stack.stack_name,
293	stack_id=change_set.stack.stack_id,
294	)
NEW 295	new_resources = change_set_executor.execute()	×
NEW 296	change_set.stack.set_stack_status(f"{change_set.change_set_type or 'UPDATE'}_COMPLETE")	×
NEW 297	change_set.stack.resources = new_resources	×
NEW 298	return ExecuteChangeSetOutput()	×
299
300	@handler("DescribeChangeSet")	1✔
301	def describe_change_set(	1✔
302	self,
303	context: RequestContext,
304	change_set_name: ChangeSetNameOrId,
305	stack_name: StackNameOrId \| None = None,
306	next_token: NextToken \| None = None,
307	include_property_values: IncludePropertyValues \| None = None,
308	**kwargs,
309	) -> DescribeChangeSetOutput:
310	# TODO add support for include_property_values
311	# only relevant if change_set_name isn't an ARN
UNCOV 312	if not ARN_CHANGESET_REGEX.match(change_set_name):	×
UNCOV 313	if not stack_name:	×
UNCOV 314	raise ValidationError(	×
315	"StackName must be specified if ChangeSetName is not specified as an ARN."
316	)
317
UNCOV 318	stack = find_stack(context.account_id, context.region, stack_name)	×
UNCOV 319	if not stack:	×
UNCOV 320	raise ValidationError(f"Stack [{stack_name}] does not exist")	×
321
UNCOV 322	change_set = find_change_set(	×
323	context.account_id, context.region, change_set_name, stack_name=stack_name
324	)
UNCOV 325	if not change_set:	×
UNCOV 326	raise ChangeSetNotFoundException(f"ChangeSet [{change_set_name}] does not exist")	×
327
NEW 328	change_set_describer = ChangeSetModelDescriber(	×
329	node_template=change_set.update_graph, include_property_values=include_property_values
330	)
NEW 331	resource_changes = change_set_describer.get_changes()	×
332
UNCOV 333	attrs = [	×
334	"ChangeSetType",
335	"StackStatus",
336	"LastUpdatedTime",
337	"DisableRollback",
338	"EnableTerminationProtection",
339	"Transform",
340	]
341	result = remove_attributes(deepcopy(change_set.metadata), attrs)	×
342	# TODO: replace this patch with a better solution
UNCOV 343	result["Parameters"] = [	×
344	mask_no_echo(strip_parameter_type(p)) for p in result.get("Parameters", [])
345	]
UNCOV 346	result["Changes"] = resource_changes	×
UNCOV 347	return result	×

localstack / localstack / 0e75e353-4494-4b65-a715-a78cf84b0b94

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous