274ae585-9ad2-4b5f-8087-866ef08d3d6e

Committed 24 Apr 2025 05:15PM UTC coverage: 85.262% (-1.0%) from 86.266%

Build # 274ae585-9ad2-4b5f-8087-866ef08d3d6e

Build Type

push

circleci

Committed by

web-flow

Commit Message

CFn v2: support outputs (#12536)

Coverage Stats

10 of 29 new or added lines in 3 files covered. (34.48%)

1105 existing lines in 26 files now uncovered.

63256 of 74190 relevant lines covered (85.26%)

0.85 hits per line

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

25.0

/localstack-core/localstack/services/cloudformation/engine/v2/change_set_model_executor.py

import copy
import logging
import uuid
from dataclasses import dataclass
from typing import Final, Optional

from localstack.aws.api.cloudformation import ChangeAction, StackStatus
from localstack.constants import INTERNAL_AWS_SECRET_ACCESS_KEY
from localstack.services.cloudformation.engine.v2.change_set_model import (
    NodeOutput,
    NodeParameter,
    NodeResource,
)
from localstack.services.cloudformation.engine.v2.change_set_model_preproc import (
    ChangeSetModelPreproc,
    PreprocEntityDelta,
    PreprocOutput,
    PreprocProperties,
    PreprocResource,
)
from localstack.services.cloudformation.resource_provider import (
    Credentials,
    OperationStatus,
    ProgressEvent,
    ResourceProviderExecutor,
    ResourceProviderPayload,
)
from localstack.services.cloudformation.v2.entities import ChangeSet

LOG = logging.getLogger(__name__)


@dataclass
class ChangeSetModelExecutorResult:
    resources: dict
    parameters: dict
    outputs: dict


class ChangeSetModelExecutor(ChangeSetModelPreproc):
    _change_set: Final[ChangeSet]
    # TODO: add typing for resolved resources and parameters.
    resources: Final[dict]
    outputs: Final[dict]
    resolved_parameters: Final[dict]

    def __init__(self, change_set: ChangeSet):
        super().__init__(
            node_template=change_set.update_graph,
            before_resolved_resources=change_set.stack.resolved_resources,
        )
        self._change_set = change_set
        self.resources = dict()
        self.outputs = dict()
        self.resolved_parameters = dict()

    # TODO: use a structured type for the return value
    def execute(self) -> ChangeSetModelExecutorResult:
        self.process()
        return ChangeSetModelExecutorResult(
            resources=self.resources, parameters=self.resolved_parameters, outputs=self.outputs
        )

    def visit_node_parameter(self, node_parameter: NodeParameter) -> PreprocEntityDelta:
        delta = super().visit_node_parameter(node_parameter=node_parameter)
        self.resolved_parameters[node_parameter.name] = delta.after
        return delta

    def _after_resource_physical_id(self, resource_logical_id: str) -> Optional[str]:
        after_resolved_resources = self.resources
        return self._resource_physical_resource_id_from(
            logical_resource_id=resource_logical_id, resolved_resources=after_resolved_resources
        )

    def visit_node_resource(
        self, node_resource: NodeResource
    ) -> PreprocEntityDelta[PreprocResource, PreprocResource]:
        """
        Overrides the default preprocessing for NodeResource objects by annotating the
        `after` delta with the physical resource ID, if side effects resulted in an update.
        """
        delta = super().visit_node_resource(node_resource=node_resource)
        self._execute_on_resource_change(
            name=node_resource.name, before=delta.before, after=delta.after
        )
        after_resource = delta.after
        if after_resource is not None and delta.before != delta.after:
            after_logical_id = after_resource.logical_id
            after_physical_id: Optional[str] = self._after_resource_physical_id(
                resource_logical_id=after_logical_id
            )
            if after_physical_id is None:
                raise RuntimeError(
                    f"No PhysicalResourceId was found for resource '{after_physical_id}' post-update."
                )
            after_resource.physical_resource_id = after_physical_id
        return delta

    def visit_node_output(
        self, node_output: NodeOutput
    ) -> PreprocEntityDelta[PreprocOutput, PreprocOutput]:
        delta = super().visit_node_output(node_output=node_output)
        if delta.after is None:
            # handling deletion so the output does not really matter
            # TODO: are there other situations?
            return delta

        self.outputs[delta.after.name] = delta.after.value
        return delta

    def _execute_on_resource_change(
        self, name: str, before: Optional[PreprocResource], after: Optional[PreprocResource]
    ) -> None:
        if before == after:
            # unchanged: nothing to do.
            return
        # TODO: this logic is a POC and should be revised.
        if before is not None and after is not None:
            # Case: change on same type.
            if before.resource_type == after.resource_type:
                # Register a Modified if changed.
                # XXX hacky, stick the previous resources' properties into the payload
                before_properties = self._merge_before_properties(name, before)

                self._execute_resource_action(
                    action=ChangeAction.Modify,
                    logical_resource_id=name,
                    resource_type=before.resource_type,
                    before_properties=before_properties,
                    after_properties=after.properties,
                )
            # Case: type migration.
            # TODO: Add test to assert that on type change the resources are replaced.
            else:
                # XXX hacky, stick the previous resources' properties into the payload
                before_properties = self._merge_before_properties(name, before)
                # Register a Removed for the previous type.
                self._execute_resource_action(
                    action=ChangeAction.Remove,
                    logical_resource_id=name,
                    resource_type=before.resource_type,
                    before_properties=before_properties,
                    after_properties=None,
                )
                # Register a Create for the next type.
                self._execute_resource_action(
                    action=ChangeAction.Add,
                    logical_resource_id=name,
                    resource_type=after.resource_type,
                    before_properties=None,
                    after_properties=after.properties,
                )
        elif before is not None:
            # Case: removal
            # XXX hacky, stick the previous resources' properties into the payload
            # XXX hacky, stick the previous resources' properties into the payload
            before_properties = self._merge_before_properties(name, before)

            self._execute_resource_action(
                action=ChangeAction.Remove,
                logical_resource_id=name,
                resource_type=before.resource_type,
                before_properties=before_properties,
                after_properties=None,
            )
        elif after is not None:
            # Case: addition
            self._execute_resource_action(
                action=ChangeAction.Add,
                logical_resource_id=name,
                resource_type=after.resource_type,
                before_properties=None,
                after_properties=after.properties,
            )

    def _merge_before_properties(
        self, name: str, preproc_resource: PreprocResource
    ) -> PreprocProperties:
        if previous_resource_properties := self._change_set.stack.resolved_resources.get(
            name, {}
        ).get("Properties"):
            return PreprocProperties(properties=previous_resource_properties)

        # XXX fall back to returning the input value
        return copy.deepcopy(preproc_resource.properties)

    def _execute_resource_action(
        self,
        action: ChangeAction,
        logical_resource_id: str,
        resource_type: str,
        before_properties: Optional[PreprocProperties],
        after_properties: Optional[PreprocProperties],
    ) -> None:
        LOG.debug("Executing resource action: %s for resource '%s'", action, logical_resource_id)
        resource_provider_executor = ResourceProviderExecutor(
            stack_name=self._change_set.stack.stack_name, stack_id=self._change_set.stack.stack_id
        )
        payload = self.create_resource_provider_payload(
            action=action,
            logical_resource_id=logical_resource_id,
            resource_type=resource_type,
            before_properties=before_properties,
            after_properties=after_properties,
        )
        resource_provider = resource_provider_executor.try_load_resource_provider(resource_type)

        extra_resource_properties = {}
        if resource_provider is not None:
            # TODO: stack events
            try:
                event = resource_provider_executor.deploy_loop(
                    resource_provider, extra_resource_properties, payload
                )
            except Exception as e:
                reason = str(e)
                LOG.warning(
                    "Resource provider operation failed: '%s'",
                    reason,
                    exc_info=LOG.isEnabledFor(logging.DEBUG),
                )
                stack = self._change_set.stack
                stack_status = stack.status
                if stack_status == StackStatus.CREATE_IN_PROGRESS:
                    stack.set_stack_status(StackStatus.CREATE_FAILED, reason=reason)
                elif stack_status == StackStatus.UPDATE_IN_PROGRESS:
                    stack.set_stack_status(StackStatus.UPDATE_FAILED, reason=reason)
                return
        else:
            event = ProgressEvent(OperationStatus.SUCCESS, resource_model={})

        self.resources.setdefault(logical_resource_id, {"Properties": {}})
        match event.status:
            case OperationStatus.SUCCESS:
                # merge the resources state with the external state
                # TODO: this is likely a duplicate of updating from extra_resource_properties
                self.resources[logical_resource_id]["Properties"].update(event.resource_model)
                self.resources[logical_resource_id].update(extra_resource_properties)
                # XXX for legacy delete_stack compatibility
                self.resources[logical_resource_id]["LogicalResourceId"] = logical_resource_id
                self.resources[logical_resource_id]["Type"] = resource_type
            case OperationStatus.FAILED:
                reason = event.message
                LOG.warning(
                    "Resource provider operation failed: '%s'",
                    reason,
                )
                # TODO: duplication
                stack = self._change_set.stack
                stack_status = stack.status
                if stack_status == StackStatus.CREATE_IN_PROGRESS:
                    stack.set_stack_status(StackStatus.CREATE_FAILED, reason=reason)
                elif stack_status == StackStatus.UPDATE_IN_PROGRESS:
                    stack.set_stack_status(StackStatus.UPDATE_FAILED, reason=reason)
                else:
                    raise NotImplementedError(f"Unhandled stack status: '{stack.status}'")
            case any:
                raise NotImplementedError(f"Event status '{any}' not handled")

    def create_resource_provider_payload(
        self,
        action: ChangeAction,
        logical_resource_id: str,
        resource_type: str,
        before_properties: Optional[PreprocProperties],
        after_properties: Optional[PreprocProperties],
    ) -> Optional[ResourceProviderPayload]:
        # FIXME: use proper credentials
        creds: Credentials = {
            "accessKeyId": self._change_set.stack.account_id,
            "secretAccessKey": INTERNAL_AWS_SECRET_ACCESS_KEY,
            "sessionToken": "",
        }
        before_properties_value = before_properties.properties if before_properties else None
        after_properties_value = after_properties.properties if after_properties else None

        match action:
            case ChangeAction.Add:
                resource_properties = after_properties_value or {}
                previous_resource_properties = None
            case ChangeAction.Modify | ChangeAction.Dynamic:
                resource_properties = after_properties_value or {}
                previous_resource_properties = before_properties_value or {}
            case ChangeAction.Remove:
                resource_properties = before_properties_value or {}
                previous_resource_properties = None
            case _:
                raise NotImplementedError(f"Action '{action}' not handled")

        resource_provider_payload: ResourceProviderPayload = {
            "awsAccountId": self._change_set.stack.account_id,
            "callbackContext": {},
            "stackId": self._change_set.stack.stack_name,
            "resourceType": resource_type,
            "resourceTypeVersion": "000000",
            # TODO: not actually a UUID
            "bearerToken": str(uuid.uuid4()),
            "region": self._change_set.stack.region_name,
            "action": str(action),
            "requestData": {
                "logicalResourceId": logical_resource_id,
                "resourceProperties": resource_properties,
                "previousResourceProperties": previous_resource_properties,
                "callerCredentials": creds,
                "providerCredentials": creds,
                "systemTags": {},
                "previousSystemTags": {},
                "stackTags": {},
                "previousStackTags": {},
            },
        }
        return resource_provider_payload

1	import copy	1✔
2	import logging	1✔
3	import uuid	1✔
4	from dataclasses import dataclass	1✔
5	from typing import Final, Optional	1✔
6
7	from localstack.aws.api.cloudformation import ChangeAction, StackStatus	1✔
8	from localstack.constants import INTERNAL_AWS_SECRET_ACCESS_KEY	1✔
9	from localstack.services.cloudformation.engine.v2.change_set_model import (	1✔
10	NodeOutput,
11	NodeParameter,
12	NodeResource,
13	)
14	from localstack.services.cloudformation.engine.v2.change_set_model_preproc import (	1✔
15	ChangeSetModelPreproc,
16	PreprocEntityDelta,
17	PreprocOutput,
18	PreprocProperties,
19	PreprocResource,
20	)
21	from localstack.services.cloudformation.resource_provider import (	1✔
22	Credentials,
23	OperationStatus,
24	ProgressEvent,
25	ResourceProviderExecutor,
26	ResourceProviderPayload,
27	)
28	from localstack.services.cloudformation.v2.entities import ChangeSet	1✔
29
30	LOG = logging.getLogger(__name__)	1✔
31
32
33	@dataclass	1✔
34	class ChangeSetModelExecutorResult:	1✔
35	resources: dict	1✔
36	parameters: dict	1✔
37	outputs: dict	1✔
38
39
40	class ChangeSetModelExecutor(ChangeSetModelPreproc):	1✔
41	_change_set: Final[ChangeSet]	1✔
42	# TODO: add typing for resolved resources and parameters.
43	resources: Final[dict]	1✔
44	outputs: Final[dict]	1✔
45	resolved_parameters: Final[dict]	1✔
46
47	def __init__(self, change_set: ChangeSet):	1✔
48	super().__init__(	×
49	node_template=change_set.update_graph,
50	before_resolved_resources=change_set.stack.resolved_resources,
51	)
UNCOV 52	self._change_set = change_set	×
UNCOV 53	self.resources = dict()	×
NEW 54	self.outputs = dict()	×
55	self.resolved_parameters = dict()	×
56
57	# TODO: use a structured type for the return value
58	def execute(self) -> ChangeSetModelExecutorResult:	1✔
UNCOV 59	self.process()	×
NEW 60	return ChangeSetModelExecutorResult(	×
61	resources=self.resources, parameters=self.resolved_parameters, outputs=self.outputs
62	)
63
64	def visit_node_parameter(self, node_parameter: NodeParameter) -> PreprocEntityDelta:	1✔
UNCOV 65	delta = super().visit_node_parameter(node_parameter=node_parameter)	×
UNCOV 66	self.resolved_parameters[node_parameter.name] = delta.after	×
UNCOV 67	return delta	×
68
69	def _after_resource_physical_id(self, resource_logical_id: str) -> Optional[str]:	1✔
70	after_resolved_resources = self.resources	×
UNCOV 71	return self._resource_physical_resource_id_from(	×
72	logical_resource_id=resource_logical_id, resolved_resources=after_resolved_resources
73	)
74
75	def visit_node_resource(	1✔
76	self, node_resource: NodeResource
77	) -> PreprocEntityDelta[PreprocResource, PreprocResource]:
78	"""
79	Overrides the default preprocessing for NodeResource objects by annotating the
80	`after` delta with the physical resource ID, if side effects resulted in an update.
81	"""
82	delta = super().visit_node_resource(node_resource=node_resource)	×
83	self._execute_on_resource_change(	×
84	name=node_resource.name, before=delta.before, after=delta.after
85	)
UNCOV 86	after_resource = delta.after	×
87	if after_resource is not None and delta.before != delta.after:	×
UNCOV 88	after_logical_id = after_resource.logical_id	×
89	after_physical_id: Optional[str] = self._after_resource_physical_id(	×
90	resource_logical_id=after_logical_id
91	)
92	if after_physical_id is None:	×
93	raise RuntimeError(	×
94	f"No PhysicalResourceId was found for resource '{after_physical_id}' post-update."
95	)
96	after_resource.physical_resource_id = after_physical_id	×
97	return delta	×
98
99	def visit_node_output(	1✔
100	self, node_output: NodeOutput
101	) -> PreprocEntityDelta[PreprocOutput, PreprocOutput]:
NEW 102	delta = super().visit_node_output(node_output=node_output)	×
NEW 103	if delta.after is None:	×
104	# handling deletion so the output does not really matter
105	# TODO: are there other situations?
NEW 106	return delta	×
107
NEW 108	self.outputs[delta.after.name] = delta.after.value	×
NEW 109	return delta	×
110
111	def _execute_on_resource_change(	1✔
112	self, name: str, before: Optional[PreprocResource], after: Optional[PreprocResource]
113	) -> None:
114	if before == after:	×
115	# unchanged: nothing to do.
UNCOV 116	return	×
117	# TODO: this logic is a POC and should be revised.
UNCOV 118	if before is not None and after is not None:	×
119	# Case: change on same type.
120	if before.resource_type == after.resource_type:	×
121	# Register a Modified if changed.
122	# XXX hacky, stick the previous resources' properties into the payload
UNCOV 123	before_properties = self._merge_before_properties(name, before)	×
124
UNCOV 125	self._execute_resource_action(	×
126	action=ChangeAction.Modify,
127	logical_resource_id=name,
128	resource_type=before.resource_type,
129	before_properties=before_properties,
130	after_properties=after.properties,
131	)
132	# Case: type migration.
133	# TODO: Add test to assert that on type change the resources are replaced.
134	else:
135	# XXX hacky, stick the previous resources' properties into the payload
UNCOV 136	before_properties = self._merge_before_properties(name, before)	×
137	# Register a Removed for the previous type.
UNCOV 138	self._execute_resource_action(	×
139	action=ChangeAction.Remove,
140	logical_resource_id=name,
141	resource_type=before.resource_type,
142	before_properties=before_properties,
143	after_properties=None,
144	)
145	# Register a Create for the next type.
UNCOV 146	self._execute_resource_action(	×
147	action=ChangeAction.Add,
148	logical_resource_id=name,
149	resource_type=after.resource_type,
150	before_properties=None,
151	after_properties=after.properties,
152	)
UNCOV 153	elif before is not None:	×
154	# Case: removal
155	# XXX hacky, stick the previous resources' properties into the payload
156	# XXX hacky, stick the previous resources' properties into the payload
UNCOV 157	before_properties = self._merge_before_properties(name, before)	×
158
159	self._execute_resource_action(	×
160	action=ChangeAction.Remove,
161	logical_resource_id=name,
162	resource_type=before.resource_type,
163	before_properties=before_properties,
164	after_properties=None,
165	)
UNCOV 166	elif after is not None:	×
167	# Case: addition
UNCOV 168	self._execute_resource_action(	×
169	action=ChangeAction.Add,
170	logical_resource_id=name,
171	resource_type=after.resource_type,
172	before_properties=None,
173	after_properties=after.properties,
174	)
175
176	def _merge_before_properties(	1✔
177	self, name: str, preproc_resource: PreprocResource
178	) -> PreprocProperties:
UNCOV 179	if previous_resource_properties := self._change_set.stack.resolved_resources.get(	×
180	name, {}
181	).get("Properties"):
UNCOV 182	return PreprocProperties(properties=previous_resource_properties)	×
183
184	# XXX fall back to returning the input value
185	return copy.deepcopy(preproc_resource.properties)	×
186
187	def _execute_resource_action(	1✔
188	self,
189	action: ChangeAction,
190	logical_resource_id: str,
191	resource_type: str,
192	before_properties: Optional[PreprocProperties],
193	after_properties: Optional[PreprocProperties],
194	) -> None:
UNCOV 195	LOG.debug("Executing resource action: %s for resource '%s'", action, logical_resource_id)	×
UNCOV 196	resource_provider_executor = ResourceProviderExecutor(	×
197	stack_name=self._change_set.stack.stack_name, stack_id=self._change_set.stack.stack_id
198	)
UNCOV 199	payload = self.create_resource_provider_payload(	×
200	action=action,
201	logical_resource_id=logical_resource_id,
202	resource_type=resource_type,
203	before_properties=before_properties,
204	after_properties=after_properties,
205	)
UNCOV 206	resource_provider = resource_provider_executor.try_load_resource_provider(resource_type)	×
207
UNCOV 208	extra_resource_properties = {}	×
UNCOV 209	if resource_provider is not None:	×
210	# TODO: stack events
UNCOV 211	try:	×
212	event = resource_provider_executor.deploy_loop(	×
213	resource_provider, extra_resource_properties, payload
214	)
215	except Exception as e:	×
UNCOV 216	reason = str(e)	×
217	LOG.warning(	×
218	"Resource provider operation failed: '%s'",
219	reason,
220	exc_info=LOG.isEnabledFor(logging.DEBUG),
221	)
222	stack = self._change_set.stack	×
223	stack_status = stack.status	×
UNCOV 224	if stack_status == StackStatus.CREATE_IN_PROGRESS:	×
UNCOV 225	stack.set_stack_status(StackStatus.CREATE_FAILED, reason=reason)	×
UNCOV 226	elif stack_status == StackStatus.UPDATE_IN_PROGRESS:	×
UNCOV 227	stack.set_stack_status(StackStatus.UPDATE_FAILED, reason=reason)	×
228	return	×
229	else:
230	event = ProgressEvent(OperationStatus.SUCCESS, resource_model={})	×
231
232	self.resources.setdefault(logical_resource_id, {"Properties": {}})	×
233	match event.status:	×
234	case OperationStatus.SUCCESS:	×
235	# merge the resources state with the external state
236	# TODO: this is likely a duplicate of updating from extra_resource_properties
UNCOV 237	self.resources[logical_resource_id]["Properties"].update(event.resource_model)	×
238	self.resources[logical_resource_id].update(extra_resource_properties)	×
239	# XXX for legacy delete_stack compatibility
240	self.resources[logical_resource_id]["LogicalResourceId"] = logical_resource_id	×
UNCOV 241	self.resources[logical_resource_id]["Type"] = resource_type	×
UNCOV 242	case OperationStatus.FAILED:	×
243	reason = event.message	×
244	LOG.warning(	×
245	"Resource provider operation failed: '%s'",
246	reason,
247	)
248	# TODO: duplication
249	stack = self._change_set.stack	×
250	stack_status = stack.status	×
UNCOV 251	if stack_status == StackStatus.CREATE_IN_PROGRESS:	×
UNCOV 252	stack.set_stack_status(StackStatus.CREATE_FAILED, reason=reason)	×
UNCOV 253	elif stack_status == StackStatus.UPDATE_IN_PROGRESS:	×
UNCOV 254	stack.set_stack_status(StackStatus.UPDATE_FAILED, reason=reason)	×
255	else:
256	raise NotImplementedError(f"Unhandled stack status: '{stack.status}'")
257	case any:	×
258	raise NotImplementedError(f"Event status '{any}' not handled")
259
260	def create_resource_provider_payload(	1✔
261	self,
262	action: ChangeAction,
263	logical_resource_id: str,
264	resource_type: str,
265	before_properties: Optional[PreprocProperties],
266	after_properties: Optional[PreprocProperties],
267	) -> Optional[ResourceProviderPayload]:
268	# FIXME: use proper credentials
UNCOV 269	creds: Credentials = {	×
270	"accessKeyId": self._change_set.stack.account_id,
271	"secretAccessKey": INTERNAL_AWS_SECRET_ACCESS_KEY,
272	"sessionToken": "",
273	}
UNCOV 274	before_properties_value = before_properties.properties if before_properties else None	×
275	after_properties_value = after_properties.properties if after_properties else None	×
276
UNCOV 277	match action:	×
UNCOV 278	case ChangeAction.Add:	×
UNCOV 279	resource_properties = after_properties_value or {}	×
280	previous_resource_properties = None	×
281	case ChangeAction.Modify \| ChangeAction.Dynamic:	×
UNCOV 282	resource_properties = after_properties_value or {}	×
283	previous_resource_properties = before_properties_value or {}	×
284	case ChangeAction.Remove:	×
285	resource_properties = before_properties_value or {}	×
286	previous_resource_properties = None	×
287	case _:	×
288	raise NotImplementedError(f"Action '{action}' not handled")
289
290	resource_provider_payload: ResourceProviderPayload = {	×
291	"awsAccountId": self._change_set.stack.account_id,
292	"callbackContext": {},
293	"stackId": self._change_set.stack.stack_name,
294	"resourceType": resource_type,
295	"resourceTypeVersion": "000000",
296	# TODO: not actually a UUID
297	"bearerToken": str(uuid.uuid4()),
298	"region": self._change_set.stack.region_name,
299	"action": str(action),
300	"requestData": {
301	"logicalResourceId": logical_resource_id,
302	"resourceProperties": resource_properties,
303	"previousResourceProperties": previous_resource_properties,
304	"callerCredentials": creds,
305	"providerCredentials": creds,
306	"systemTags": {},
307	"previousSystemTags": {},
308	"stackTags": {},
309	"previousStackTags": {},
310	},
311	}
UNCOV 312	return resource_provider_payload	×

localstack / localstack / 274ae585-9ad2-4b5f-8087-866ef08d3d6e

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