20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

39.2

/localstack-core/localstack/services/cloudformation/engine/entities.py

import logging
from typing import TypedDict

from localstack.aws.api.cloudformation import Capability, ChangeSetType, Parameter
from localstack.services.cloudformation.engine.parameters import (
    StackParameter,
    convert_stack_parameters_to_list,
    mask_no_echo,
    strip_parameter_type,
)
from localstack.services.cloudformation.engine.v2.change_set_model import (
    ChangeSetModel,
    NodeTemplate,
)
from localstack.utils.aws import arns
from localstack.utils.collections import select_attributes
from localstack.utils.id_generator import (
    ExistingIds,
    ResourceIdentifier,
    Tags,
    generate_short_uid,
    generate_uid,
)
from localstack.utils.json import clone_safe
from localstack.utils.objects import recurse_object
from localstack.utils.strings import long_uid, short_uid
from localstack.utils.time import timestamp_millis

LOG = logging.getLogger(__name__)


class StackSet:
    """A stack set contains multiple stack instances."""

    # FIXME: confusing name. metadata is the complete incoming request object
    def __init__(self, metadata: dict):
        self.metadata = metadata
        # list of stack instances
        self.stack_instances = []
        # maps operation ID to stack set operation details
        self.operations = {}

    @property
    def stack_set_name(self):
        return self.metadata.get("StackSetName")


class StackInstance:
    """A stack instance belongs to a stack set and is specific to a region / account ID."""

    # FIXME: confusing name. metadata is the complete incoming request object
    def __init__(self, metadata: dict):
        self.metadata = metadata
        # reference to the deployed stack belonging to this stack instance
        self.stack = None


class CreateChangeSetInput(TypedDict):
    StackName: str
    Capabilities: list[Capability]
    ChangeSetName: str | None
    ChangSetType: ChangeSetType | None
    Parameters: list[Parameter]


class StackTemplate(TypedDict):
    StackName: str
    ChangeSetName: str | None
    Outputs: dict
    Resources: dict


class StackIdentifier(ResourceIdentifier):
    service = "cloudformation"
    resource = "stack"

    def __init__(self, account_id: str, region: str, stack_name: str):
        super().__init__(account_id, region, stack_name)

    def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:
        return generate_short_uid(resource_identifier=self, existing_ids=existing_ids, tags=tags)


class StackIdentifierV2(StackIdentifier):
    def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:
        return generate_uid(resource_identifier=self, existing_ids=existing_ids, tags=tags)


# TODO: remove metadata (flatten into individual fields)
class Stack:
    change_sets: list["StackChangeSet"]

    def __init__(
        self,
        account_id: str,
        region_name: str,
        metadata: CreateChangeSetInput | None = None,
        template: StackTemplate | None = None,
        template_body: str | None = None,
    ):
        self.account_id = account_id
        self.region_name = region_name

        if template is None:
            template = {}

        self.resolved_outputs = []  # TODO
        self.resolved_parameters: dict[str, StackParameter] = {}
        self.resolved_conditions: dict[str, bool] = {}

        self.metadata = metadata or {}
        self.template = template or {}
        self.template_body = template_body
        self._template_raw = clone_safe(self.template)
        self.template_original = clone_safe(self.template)
        # initialize resources
        for resource_id, resource in self.template_resources.items():
            # HACK: if the resource is a Fn::ForEach intrinsic call from the LanguageExtensions transform, then it is not a dictionary but a list
            if resource_id.startswith("Fn::ForEach"):
                # we are operating on an untransformed template, so ignore for now
                continue
            resource["LogicalResourceId"] = self.template_original["Resources"][resource_id][
                "LogicalResourceId"
            ] = resource.get("LogicalResourceId") or resource_id
        # initialize stack template attributes
        stack_id = self.metadata.get("StackId") or arns.cloudformation_stack_arn(
            self.stack_name,
            stack_id=StackIdentifier(
                account_id=account_id, region=region_name, stack_name=metadata.get("StackName")
            ).generate(tags=metadata.get("tags")),
            account_id=account_id,
            region_name=region_name,
        )
        self.template["StackId"] = self.metadata["StackId"] = stack_id
        self.template["Parameters"] = self.template.get("Parameters") or {}
        self.template["Outputs"] = self.template.get("Outputs") or {}
        self.template["Conditions"] = self.template.get("Conditions") or {}
        # initialize metadata
        self.metadata["Parameters"] = self.metadata.get("Parameters") or []
        self.metadata["StackStatus"] = "CREATE_IN_PROGRESS"
        self.metadata["CreationTime"] = self.metadata.get("CreationTime") or timestamp_millis()
        self.metadata["LastUpdatedTime"] = self.metadata["CreationTime"]
        self.metadata.setdefault("Description", self.template.get("Description"))
        self.metadata.setdefault("RollbackConfiguration", {})
        self.metadata.setdefault("DisableRollback", False)
        self.metadata.setdefault("EnableTerminationProtection", False)
        # maps resource id to resource state
        self._resource_states = {}
        # list of stack events
        self.events = []
        # list of stack change sets
        self.change_sets = []
        # self.evaluated_conditions = {}

    def set_resolved_parameters(self, resolved_parameters: dict[str, StackParameter]):
        self.resolved_parameters = resolved_parameters
        if resolved_parameters:
            self.metadata["Parameters"] = list(resolved_parameters.values())

    def set_resolved_stack_conditions(self, resolved_conditions: dict[str, bool]):
        self.resolved_conditions = resolved_conditions

    def describe_details(self):
        attrs = [
            "StackId",
            "StackName",
            "Description",
            "StackStatusReason",
            "StackStatus",
            "Capabilities",
            "ParentId",
            "RootId",
            "RoleARN",
            "CreationTime",
            "DeletionTime",
            "LastUpdatedTime",
            "ChangeSetId",
            "RollbackConfiguration",
            "DisableRollback",
            "EnableTerminationProtection",
            "DriftInformation",
        ]
        result = select_attributes(self.metadata, attrs)
        result["Tags"] = self.tags
        outputs = self.resolved_outputs
        if outputs:
            result["Outputs"] = outputs
        stack_parameters = convert_stack_parameters_to_list(self.resolved_parameters)
        if stack_parameters:
            result["Parameters"] = [
                mask_no_echo(strip_parameter_type(sp)) for sp in stack_parameters
            ]
        if not result.get("DriftInformation"):
            result["DriftInformation"] = {"StackDriftStatus": "NOT_CHECKED"}
        for attr in ["Tags", "NotificationARNs"]:
            result.setdefault(attr, [])
        return result

    def set_stack_status(self, status: str, status_reason: str | None = None):
        self.metadata["StackStatus"] = status
        if "FAILED" in status:
            self.metadata["StackStatusReason"] = status_reason or "Deployment failed"
        self.log_stack_errors()
        self.add_stack_event(
            self.stack_name, self.stack_id, status, status_reason=status_reason or ""
        )

    def log_stack_errors(self, level=logging.WARNING):
        for event in self.events:
            if event["ResourceStatus"].endswith("FAILED"):
                if reason := event.get("ResourceStatusReason"):
                    reason = reason.replace("\n", "; ")
                    LOG.log(
                        level,
                        "CFn resource failed to deploy: %s (%s)",
                        event["LogicalResourceId"],
                        reason,
                    )
                else:
                    LOG.warning("CFn resource failed to deploy: %s", event["LogicalResourceId"])

    def set_time_attribute(self, attribute, new_time=None):
        self.metadata[attribute] = new_time or timestamp_millis()

    def add_stack_event(
        self,
        resource_id: str = None,
        physical_res_id: str = None,
        status: str = "",
        status_reason: str = "",
    ):
        resource_id = resource_id or self.stack_name
        physical_res_id = physical_res_id or self.stack_id
        resource_type = (
            self.template.get("Resources", {})
            .get(resource_id, {})
            .get("Type", "AWS::CloudFormation::Stack")
        )

        event = {
            "EventId": long_uid(),
            "Timestamp": timestamp_millis(),
            "StackId": self.stack_id,
            "StackName": self.stack_name,
            "LogicalResourceId": resource_id,
            "PhysicalResourceId": physical_res_id,
            "ResourceStatus": status,
            "ResourceType": resource_type,
        }

        if status_reason:
            event["ResourceStatusReason"] = status_reason

        self.events.insert(0, event)

    def set_resource_status(self, resource_id: str, status: str, status_reason: str = ""):
        """Update the deployment status of the given resource ID and publish a corresponding stack event."""
        physical_res_id = self.resources.get(resource_id, {}).get("PhysicalResourceId")
        self._set_resource_status_details(resource_id, physical_res_id=physical_res_id)
        state = self.resource_states.setdefault(resource_id, {})
        state["PreviousResourceStatus"] = state.get("ResourceStatus")
        state["ResourceStatus"] = status
        state["LastUpdatedTimestamp"] = timestamp_millis()
        self.add_stack_event(resource_id, physical_res_id, status, status_reason=status_reason)

    def _set_resource_status_details(self, resource_id: str, physical_res_id: str = None):
        """Helper function to ensure that the status details for the given resource ID are up-to-date."""
        resource = self.resources.get(resource_id)
        if resource is None or resource.get("Type") == "Parameter":
            # make sure we delete the states for any non-existing/deleted resources
            self._resource_states.pop(resource_id, None)
            return
        state = self._resource_states.setdefault(resource_id, {})
        attr_defaults = (
            ("LogicalResourceId", resource_id),
            ("PhysicalResourceId", physical_res_id),
        )
        for res in [resource, state]:
            for attr, default in attr_defaults:
                res[attr] = res.get(attr) or default
        state["StackName"] = state.get("StackName") or self.stack_name
        state["StackId"] = state.get("StackId") or self.stack_id
        state["ResourceType"] = state.get("ResourceType") or self.resources[resource_id].get("Type")
        state["Timestamp"] = timestamp_millis()
        return state

    def resource_status(self, resource_id: str):
        result = self._lookup(self.resource_states, resource_id)
        return result

    def latest_template_raw(self):
        if self.change_sets:
            return self.change_sets[-1]._template_raw
        return self._template_raw

    @property
    def resource_states(self):
        for resource_id in list(self._resource_states.keys()):
            self._set_resource_status_details(resource_id)
        return self._resource_states

    @property
    def stack_name(self):
        return self.metadata["StackName"]

    @property
    def stack_id(self):
        return self.metadata["StackId"]

    @property
    def resources(self):
        """Return dict of resources"""
        return dict(self.template_resources)

    @resources.setter
    def resources(self, resources: dict):
        self.template["Resources"] = resources

    @property
    def template_resources(self):
        return self.template.setdefault("Resources", {})

    @property
    def tags(self):
        return self.metadata.get("Tags", [])

    @property
    def imports(self):
        def _collect(o, **kwargs):
            if isinstance(o, dict):
                import_val = o.get("Fn::ImportValue")
                if import_val:
                    result.add(import_val)
            return o

        result = set()
        recurse_object(self.resources, _collect)
        return result

    @property
    def template_parameters(self):
        return self.template["Parameters"]

    @property
    def conditions(self):
        """Returns the (mutable) dict of stack conditions."""
        return self.template.setdefault("Conditions", {})

    @property
    def mappings(self):
        """Returns the (mutable) dict of stack mappings."""
        return self.template.setdefault("Mappings", {})

    @property
    def outputs(self):
        """Returns the (mutable) dict of stack outputs."""
        return self.template.setdefault("Outputs", {})

    @property
    def status(self):
        return self.metadata["StackStatus"]

    @property
    def resource_types(self):
        return [r.get("Type") for r in self.template_resources.values()]

    def resource(self, resource_id):
        return self._lookup(self.resources, resource_id)

    def _lookup(self, resource_map, resource_id):
        resource = resource_map.get(resource_id)
        if not resource:
            raise Exception(
                f'Unable to find details for resource "{resource_id}" in stack "{self.stack_name}"'
            )
        return resource

    def copy(self):
        return Stack(
            account_id=self.account_id,
            region_name=self.region_name,
            metadata=dict(self.metadata),
            template=dict(self.template),
        )


# FIXME: remove inheritance
# TODO: what functionality of the Stack object do we rely on here?
class StackChangeSet(Stack):
    update_graph: NodeTemplate | None
    change_set_type: ChangeSetType | None

    def __init__(
        self,
        account_id: str,
        region_name: str,
        stack: Stack,
        params=None,
        template=None,
        change_set_type: ChangeSetType | None = None,
    ):
        if template is None:
            template = {}
        if params is None:
            params = {}
        super().__init__(account_id, region_name, params, template)

        name = self.metadata["ChangeSetName"]
        if not self.metadata.get("ChangeSetId"):
            self.metadata["ChangeSetId"] = arns.cloudformation_change_set_arn(
                name, change_set_id=short_uid(), account_id=account_id, region_name=region_name
            )

        self.account_id = account_id
        self.region_name = region_name
        self.stack = stack
        self.metadata["StackId"] = stack.stack_id
        self.metadata["Status"] = "CREATE_PENDING"
        self.change_set_type = change_set_type

    @property
    def change_set_id(self):
        return self.metadata["ChangeSetId"]

    @property
    def change_set_name(self):
        return self.metadata["ChangeSetName"]

    @property
    def resources(self):
        return dict(self.stack.resources)

    @property
    def changes(self):
        result = self.metadata["Changes"] = self.metadata.get("Changes", [])
        return result

    # V2 only
    def populate_update_graph(
        self,
        before_template: dict | None,
        after_template: dict | None,
        before_parameters: dict | None,
        after_parameters: dict | None,
    ) -> None:
        change_set_model = ChangeSetModel(
            before_template=before_template,
            after_template=after_template,
            before_parameters=before_parameters,
            after_parameters=after_parameters,
        )
        self.update_graph = change_set_model.get_update_model()

1	import logging	1✔
2	from typing import TypedDict	1✔
3
4	from localstack.aws.api.cloudformation import Capability, ChangeSetType, Parameter	1✔
5	from localstack.services.cloudformation.engine.parameters import (	1✔
6	StackParameter,
7	convert_stack_parameters_to_list,
8	mask_no_echo,
9	strip_parameter_type,
10	)
11	from localstack.services.cloudformation.engine.v2.change_set_model import (	1✔
12	ChangeSetModel,
13	NodeTemplate,
14	)
15	from localstack.utils.aws import arns	1✔
16	from localstack.utils.collections import select_attributes	1✔
17	from localstack.utils.id_generator import (	1✔
18	ExistingIds,
19	ResourceIdentifier,
20	Tags,
21	generate_short_uid,
22	generate_uid,
23	)
24	from localstack.utils.json import clone_safe	1✔
25	from localstack.utils.objects import recurse_object	1✔
26	from localstack.utils.strings import long_uid, short_uid	1✔
27	from localstack.utils.time import timestamp_millis	1✔
28
29	LOG = logging.getLogger(__name__)	1✔
30
31
32	class StackSet:	1✔
33	"""A stack set contains multiple stack instances."""
34
35	# FIXME: confusing name. metadata is the complete incoming request object
36	def __init__(self, metadata: dict):	1✔
37	self.metadata = metadata	×
38	# list of stack instances
39	self.stack_instances = []	×
40	# maps operation ID to stack set operation details
41	self.operations = {}	×
42
43	@property	1✔
44	def stack_set_name(self):	1✔
45	return self.metadata.get("StackSetName")	×
46
47
48	class StackInstance:	1✔
49	"""A stack instance belongs to a stack set and is specific to a region / account ID."""
50
51	# FIXME: confusing name. metadata is the complete incoming request object
52	def __init__(self, metadata: dict):	1✔
53	self.metadata = metadata	×
54	# reference to the deployed stack belonging to this stack instance
55	self.stack = None	×
56
57
58	class CreateChangeSetInput(TypedDict):	1✔
59	StackName: str	1✔
60	Capabilities: list[Capability]	1✔
61	ChangeSetName: str \| None	1✔
62	ChangSetType: ChangeSetType \| None	1✔
63	Parameters: list[Parameter]	1✔
64
65
66	class StackTemplate(TypedDict):	1✔
67	StackName: str	1✔
68	ChangeSetName: str \| None	1✔
69	Outputs: dict	1✔
70	Resources: dict	1✔
71
72
73	class StackIdentifier(ResourceIdentifier):	1✔
74	service = "cloudformation"	1✔
75	resource = "stack"	1✔
76
77	def __init__(self, account_id: str, region: str, stack_name: str):	1✔
78	super().__init__(account_id, region, stack_name)	1✔
79
80	def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:	1✔
81	return generate_short_uid(resource_identifier=self, existing_ids=existing_ids, tags=tags)	1✔
82
83
84	class StackIdentifierV2(StackIdentifier):	1✔
85	def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:	1✔
86	return generate_uid(resource_identifier=self, existing_ids=existing_ids, tags=tags)	1✔
87
88
89	# TODO: remove metadata (flatten into individual fields)
90	class Stack:	1✔
91	change_sets: list["StackChangeSet"]	1✔
92
93	def __init__(	1✔
94	self,
95	account_id: str,
96	region_name: str,
97	metadata: CreateChangeSetInput \| None = None,
98	template: StackTemplate \| None = None,
99	template_body: str \| None = None,
100	):
101	self.account_id = account_id	×
102	self.region_name = region_name	×
103
104	if template is None:	×
105	template = {}	×
106
107	self.resolved_outputs = [] # TODO	×
108	self.resolved_parameters: dict[str, StackParameter] = {}	×
109	self.resolved_conditions: dict[str, bool] = {}	×
110
111	self.metadata = metadata or {}	×
112	self.template = template or {}	×
113	self.template_body = template_body	×
114	self._template_raw = clone_safe(self.template)	×
115	self.template_original = clone_safe(self.template)	×
116	# initialize resources
117	for resource_id, resource in self.template_resources.items():	×
118	# HACK: if the resource is a Fn::ForEach intrinsic call from the LanguageExtensions transform, then it is not a dictionary but a list
119	if resource_id.startswith("Fn::ForEach"):	×
120	# we are operating on an untransformed template, so ignore for now
121	continue	×
122	resource["LogicalResourceId"] = self.template_original["Resources"][resource_id][	×
123	"LogicalResourceId"
124	] = resource.get("LogicalResourceId") or resource_id
125	# initialize stack template attributes
126	stack_id = self.metadata.get("StackId") or arns.cloudformation_stack_arn(	×
127	self.stack_name,
128	stack_id=StackIdentifier(
129	account_id=account_id, region=region_name, stack_name=metadata.get("StackName")
130	).generate(tags=metadata.get("tags")),
131	account_id=account_id,
132	region_name=region_name,
133	)
134	self.template["StackId"] = self.metadata["StackId"] = stack_id	×
135	self.template["Parameters"] = self.template.get("Parameters") or {}	×
136	self.template["Outputs"] = self.template.get("Outputs") or {}	×
137	self.template["Conditions"] = self.template.get("Conditions") or {}	×
138	# initialize metadata
139	self.metadata["Parameters"] = self.metadata.get("Parameters") or []	×
140	self.metadata["StackStatus"] = "CREATE_IN_PROGRESS"	×
141	self.metadata["CreationTime"] = self.metadata.get("CreationTime") or timestamp_millis()	×
142	self.metadata["LastUpdatedTime"] = self.metadata["CreationTime"]	×
143	self.metadata.setdefault("Description", self.template.get("Description"))	×
144	self.metadata.setdefault("RollbackConfiguration", {})	×
145	self.metadata.setdefault("DisableRollback", False)	×
146	self.metadata.setdefault("EnableTerminationProtection", False)	×
147	# maps resource id to resource state
148	self._resource_states = {}	×
149	# list of stack events
150	self.events = []	×
151	# list of stack change sets
152	self.change_sets = []	×
153	# self.evaluated_conditions = {}
154
155	def set_resolved_parameters(self, resolved_parameters: dict[str, StackParameter]):	1✔
156	self.resolved_parameters = resolved_parameters	×
157	if resolved_parameters:	×
158	self.metadata["Parameters"] = list(resolved_parameters.values())	×
159
160	def set_resolved_stack_conditions(self, resolved_conditions: dict[str, bool]):	1✔
161	self.resolved_conditions = resolved_conditions	×
162
163	def describe_details(self):	1✔
164	attrs = [	×
165	"StackId",
166	"StackName",
167	"Description",
168	"StackStatusReason",
169	"StackStatus",
170	"Capabilities",
171	"ParentId",
172	"RootId",
173	"RoleARN",
174	"CreationTime",
175	"DeletionTime",
176	"LastUpdatedTime",
177	"ChangeSetId",
178	"RollbackConfiguration",
179	"DisableRollback",
180	"EnableTerminationProtection",
181	"DriftInformation",
182	]
183	result = select_attributes(self.metadata, attrs)	×
184	result["Tags"] = self.tags	×
185	outputs = self.resolved_outputs	×
186	if outputs:	×
187	result["Outputs"] = outputs	×
188	stack_parameters = convert_stack_parameters_to_list(self.resolved_parameters)	×
189	if stack_parameters:	×
190	result["Parameters"] = [	×
191	mask_no_echo(strip_parameter_type(sp)) for sp in stack_parameters
192	]
193	if not result.get("DriftInformation"):	×
194	result["DriftInformation"] = {"StackDriftStatus": "NOT_CHECKED"}	×
195	for attr in ["Tags", "NotificationARNs"]:	×
196	result.setdefault(attr, [])	×
197	return result	×
198
199	def set_stack_status(self, status: str, status_reason: str \| None = None):	1✔
200	self.metadata["StackStatus"] = status	×
201	if "FAILED" in status:	×
202	self.metadata["StackStatusReason"] = status_reason or "Deployment failed"	×
203	self.log_stack_errors()	×
204	self.add_stack_event(	×
205	self.stack_name, self.stack_id, status, status_reason=status_reason or ""
206	)
207
208	def log_stack_errors(self, level=logging.WARNING):	1✔
209	for event in self.events:	×
210	if event["ResourceStatus"].endswith("FAILED"):	×
211	if reason := event.get("ResourceStatusReason"):	×
212	reason = reason.replace("\n", "; ")	×
213	LOG.log(	×
214	level,
215	"CFn resource failed to deploy: %s (%s)",
216	event["LogicalResourceId"],
217	reason,
218	)
219	else:
220	LOG.warning("CFn resource failed to deploy: %s", event["LogicalResourceId"])	×
221
222	def set_time_attribute(self, attribute, new_time=None):	1✔
223	self.metadata[attribute] = new_time or timestamp_millis()	×
224
225	def add_stack_event(	1✔
226	self,
227	resource_id: str = None,
228	physical_res_id: str = None,
229	status: str = "",
230	status_reason: str = "",
231	):
232	resource_id = resource_id or self.stack_name	×
233	physical_res_id = physical_res_id or self.stack_id	×
234	resource_type = (	×
235	self.template.get("Resources", {})
236	.get(resource_id, {})
237	.get("Type", "AWS::CloudFormation::Stack")
238	)
239
240	event = {	×
241	"EventId": long_uid(),
242	"Timestamp": timestamp_millis(),
243	"StackId": self.stack_id,
244	"StackName": self.stack_name,
245	"LogicalResourceId": resource_id,
246	"PhysicalResourceId": physical_res_id,
247	"ResourceStatus": status,
248	"ResourceType": resource_type,
249	}
250
251	if status_reason:	×
252	event["ResourceStatusReason"] = status_reason	×
253
254	self.events.insert(0, event)	×
255
256	def set_resource_status(self, resource_id: str, status: str, status_reason: str = ""):	1✔
257	"""Update the deployment status of the given resource ID and publish a corresponding stack event."""
258	physical_res_id = self.resources.get(resource_id, {}).get("PhysicalResourceId")	×
259	self._set_resource_status_details(resource_id, physical_res_id=physical_res_id)	×
260	state = self.resource_states.setdefault(resource_id, {})	×
261	state["PreviousResourceStatus"] = state.get("ResourceStatus")	×
262	state["ResourceStatus"] = status	×
263	state["LastUpdatedTimestamp"] = timestamp_millis()	×
264	self.add_stack_event(resource_id, physical_res_id, status, status_reason=status_reason)	×
265
266	def _set_resource_status_details(self, resource_id: str, physical_res_id: str = None):	1✔
267	"""Helper function to ensure that the status details for the given resource ID are up-to-date."""
268	resource = self.resources.get(resource_id)	×
269	if resource is None or resource.get("Type") == "Parameter":	×
270	# make sure we delete the states for any non-existing/deleted resources
271	self._resource_states.pop(resource_id, None)	×
272	return	×
273	state = self._resource_states.setdefault(resource_id, {})	×
274	attr_defaults = (	×
275	("LogicalResourceId", resource_id),
276	("PhysicalResourceId", physical_res_id),
277	)
278	for res in [resource, state]:	×
279	for attr, default in attr_defaults:	×
280	res[attr] = res.get(attr) or default	×
281	state["StackName"] = state.get("StackName") or self.stack_name	×
282	state["StackId"] = state.get("StackId") or self.stack_id	×
283	state["ResourceType"] = state.get("ResourceType") or self.resources[resource_id].get("Type")	×
284	state["Timestamp"] = timestamp_millis()	×
285	return state	×
286
287	def resource_status(self, resource_id: str):	1✔
288	result = self._lookup(self.resource_states, resource_id)	×
289	return result	×
290
291	def latest_template_raw(self):	1✔
292	if self.change_sets:	×
293	return self.change_sets[-1]._template_raw	×
294	return self._template_raw	×
295
296	@property	1✔
297	def resource_states(self):	1✔
298	for resource_id in list(self._resource_states.keys()):	×
299	self._set_resource_status_details(resource_id)	×
300	return self._resource_states	×
301
302	@property	1✔
303	def stack_name(self):	1✔
304	return self.metadata["StackName"]	×
305
306	@property	1✔
307	def stack_id(self):	1✔
308	return self.metadata["StackId"]	×
309
310	@property	1✔
311	def resources(self):	1✔
312	"""Return dict of resources"""
313	return dict(self.template_resources)	×
314
315	@resources.setter	1✔
316	def resources(self, resources: dict):	1✔
317	self.template["Resources"] = resources	×
318
319	@property	1✔
320	def template_resources(self):	1✔
321	return self.template.setdefault("Resources", {})	×
322
323	@property	1✔
324	def tags(self):	1✔
325	return self.metadata.get("Tags", [])	×
326
327	@property	1✔
328	def imports(self):	1✔
329	def _collect(o, **kwargs):	×
330	if isinstance(o, dict):	×
331	import_val = o.get("Fn::ImportValue")	×
332	if import_val:	×
333	result.add(import_val)	×
334	return o	×
335
336	result = set()	×
337	recurse_object(self.resources, _collect)	×
338	return result	×
339
340	@property	1✔
341	def template_parameters(self):	1✔
342	return self.template["Parameters"]	×
343
344	@property	1✔
345	def conditions(self):	1✔
346	"""Returns the (mutable) dict of stack conditions."""
347	return self.template.setdefault("Conditions", {})	×
348
349	@property	1✔
350	def mappings(self):	1✔
351	"""Returns the (mutable) dict of stack mappings."""
352	return self.template.setdefault("Mappings", {})	×
353
354	@property	1✔
355	def outputs(self):	1✔
356	"""Returns the (mutable) dict of stack outputs."""
357	return self.template.setdefault("Outputs", {})	×
358
359	@property	1✔
360	def status(self):	1✔
361	return self.metadata["StackStatus"]	×
362
363	@property	1✔
364	def resource_types(self):	1✔
365	return [r.get("Type") for r in self.template_resources.values()]	×
366
367	def resource(self, resource_id):	1✔
368	return self._lookup(self.resources, resource_id)	×
369
370	def _lookup(self, resource_map, resource_id):	1✔
371	resource = resource_map.get(resource_id)	×
372	if not resource:	×
373	raise Exception(	×
374	f'Unable to find details for resource "{resource_id}" in stack "{self.stack_name}"'
375	)
376	return resource	×
377
378	def copy(self):	1✔
379	return Stack(	×
380	account_id=self.account_id,
381	region_name=self.region_name,
382	metadata=dict(self.metadata),
383	template=dict(self.template),
384	)
385
386
387	# FIXME: remove inheritance
388	# TODO: what functionality of the Stack object do we rely on here?
389	class StackChangeSet(Stack):	1✔
390	update_graph: NodeTemplate \| None	1✔
391	change_set_type: ChangeSetType \| None	1✔
392
393	def __init__(	1✔
394	self,
395	account_id: str,
396	region_name: str,
397	stack: Stack,
398	params=None,
399	template=None,
400	change_set_type: ChangeSetType \| None = None,
401	):
402	if template is None:	×
403	template = {}	×
404	if params is None:	×
405	params = {}	×
406	super().__init__(account_id, region_name, params, template)	×
407
408	name = self.metadata["ChangeSetName"]	×
409	if not self.metadata.get("ChangeSetId"):	×
410	self.metadata["ChangeSetId"] = arns.cloudformation_change_set_arn(	×
411	name, change_set_id=short_uid(), account_id=account_id, region_name=region_name
412	)
413
414	self.account_id = account_id	×
415	self.region_name = region_name	×
416	self.stack = stack	×
417	self.metadata["StackId"] = stack.stack_id	×
418	self.metadata["Status"] = "CREATE_PENDING"	×
419	self.change_set_type = change_set_type	×
420
421	@property	1✔
422	def change_set_id(self):	1✔
423	return self.metadata["ChangeSetId"]	×
424
425	@property	1✔
426	def change_set_name(self):	1✔
427	return self.metadata["ChangeSetName"]	×
428
429	@property	1✔
430	def resources(self):	1✔
431	return dict(self.stack.resources)	×
432
433	@property	1✔
434	def changes(self):	1✔
435	result = self.metadata["Changes"] = self.metadata.get("Changes", [])	×
436	return result	×
437
438	# V2 only
439	def populate_update_graph(	1✔
440	self,
441	before_template: dict \| None,
442	after_template: dict \| None,
443	before_parameters: dict \| None,
444	after_parameters: dict \| None,
445	) -> None:
446	change_set_model = ChangeSetModel(	×
447	before_template=before_template,
448	after_template=after_template,
449	before_parameters=before_parameters,
450	after_parameters=after_parameters,
451	)
452	self.update_graph = change_set_model.get_update_model()	×

localstack / localstack / 20565403496

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