5f1fdd48-92f9-4508-ac88-80c0245b2bad

Committed 09 Apr 2025 07:07PM UTC coverage: 86.783% (-0.03%) from 86.809%

Build # 5f1fdd48-92f9-4508-ac88-80c0245b2bad

Build Type

push

circleci

Committed by

web-flow

Commit Message

EC2: generate security group ids using id manager concept (#12494)

Co-authored-by: Mathieu Cloutier <cloutier.mat0@gmail.com>

Coverage Stats

14 of 14 new or added lines in 1 file covered. (100.0%)

615 existing lines in 23 files now uncovered.

63754 of 73464 relevant lines covered (86.78%)

0.87 hits per line

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

91.02

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

import logging
from typing import Optional, 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
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 StackMetadata(TypedDict):
    StackName: str
    Capabilities: list[Capability]
    ChangeSetName: Optional[str]
    ChangSetType: Optional[ChangeSetType]
    Parameters: list[Parameter]


class StackTemplate(TypedDict):
    StackName: str
    ChangeSetName: Optional[str]
    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)


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

    def __init__(
        self,
        account_id: str,
        region_name: str,
        metadata: Optional[StackMetadata] = None,
        template: Optional[StackTemplate] = None,
        template_body: Optional[str] = None,
    ):
        self.account_id = account_id
        self.region_name = region_name

        if template is None:
            template = {}

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

localstack / localstack / 5f1fdd48-92f9-4508-ac88-80c0245b2bad

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