20357994067

Committed 18 Dec 2025 10:01PM UTC coverage: 86.913% (-0.02%) from 86.929%

Build # 20357994067

Build Type

push

github

Committed by

web-flow

Commit Message

Fix CloudWatch model annotation (#13545)

Coverage Stats

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

1391 existing lines in 33 files now uncovered.

70000 of 80540 relevant lines covered (86.91%)

1.72 hits per line

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

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

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