16766254352

Committed 05 Aug 2025 04:40PM UTC coverage: 86.892% (-0.02%) from 86.91%

Build # 16766254352

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: defer deletions for correcting deploy order (#12936)

Coverage Stats

92 of 99 new or added lines in 6 files covered. (92.93%)

185 existing lines in 21 files now uncovered.

66597 of 76643 relevant lines covered (86.89%)

0.87 hits per line

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

98.59

/localstack-core/localstack/services/cloudformation/stores.py

import logging
from typing import Optional

from localstack.aws.api.cloudformation import Export, StackStatus
from localstack.services.cloudformation.engine.entities import Stack, StackChangeSet, StackSet
from localstack.services.cloudformation.v2.entities import ChangeSet as ChangeSetV2
from localstack.services.cloudformation.v2.entities import Stack as StackV2
from localstack.services.cloudformation.v2.entities import StackSet as StackSetV2
from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute

LOG = logging.getLogger(__name__)


class CloudFormationStore(BaseStore):
    # maps stack ID to stack details
    stacks: dict[str, Stack] = LocalAttribute(default=dict)
    stacks_v2: dict[str, StackV2] = LocalAttribute(default=dict)

    change_sets: dict[str, ChangeSetV2] = LocalAttribute(default=dict)

    # maps stack set ID to stack set details
    stack_sets: dict[str, StackSet] = LocalAttribute(default=dict)
    stack_sets_v2: dict[str, StackSetV2] = LocalAttribute(default=dict)

    # maps macro ID to macros
    macros: dict[str, dict] = LocalAttribute(default=dict)

    @property
    def exports(self) -> dict[str, Export]:
        exports = {}

        for stack_id, stack in self.stacks.items():
            if stack.status == StackStatus.DELETE_COMPLETE:
                continue

            for output in stack.resolved_outputs:
                export_name = output.get("ExportName")
                if not export_name:
                    continue
                if export_name in exports.keys():
                    # TODO: raise exception on stack creation in case of duplicate exports
                    LOG.warning(
                        "Found duplicate export name %s in stacks: %s %s",
                        export_name,
                        output["OutputValue"],
                        stack.stack_id,
                    )
                exports[export_name] = Export(
                    ExportingStackId=stack.stack_id, Name=export_name, Value=output["OutputValue"]
                )

        return exports

    @property
    def exports_v2(self) -> dict[str, Export]:
        exports = dict()
        stacks_v2 = self.stacks_v2.values()
        for stack in stacks_v2:
            if stack.status == StackStatus.DELETE_COMPLETE:
                continue
            for export_name, export_value in stack.resolved_exports.items():
                exports[export_name] = Export(
                    ExportingStackId=stack.stack_id, Name=export_name, Value=export_value
                )

        return exports


cloudformation_stores = AccountRegionBundle("cloudformation", CloudFormationStore)


def get_cloudformation_store(account_id: str, region_name: str) -> CloudFormationStore:
    return cloudformation_stores[account_id][region_name]


# TODO: rework / fix usage of this
def find_stack(account_id: str, region_name: str, stack_name: str) -> Stack | None:
    # Warning: This function may not return the correct stack if multiple stacks with same name exist.
    state = get_cloudformation_store(account_id, region_name)
    return (
        [s for s in state.stacks.values() if stack_name in [s.stack_name, s.stack_id]] or [None]
    )[0]


def find_stack_by_id(account_id: str, region_name: str, stack_id: str) -> Stack | None:
    """
    Find the stack by id.

    :param account_id: account of the stack
    :param region_name: region of the stack
    :param stack_id: stack id
    :return: Stack if it is found, None otherwise
    """
    state = get_cloudformation_store(account_id, region_name)
    for stack in state.stacks.values():
        # there can only be one stack with an id
        if stack_id == stack.stack_id:
            return stack
    return None


def find_active_stack_by_name_or_id(
    account_id: str, region_name: str, stack_name_or_id: str
) -> Stack | None:
    """
    Find the active stack by name. Some cloudformation operations only allow referencing by slack name if the stack is
    "active", which we currently interpret as not DELETE_COMPLETE.

    :param account_id: account of the stack
    :param region_name: region of the stack
    :param stack_name_or_id: stack name or stack id
    :return: Stack if it is found, None otherwise
    """
    state = get_cloudformation_store(account_id, region_name)
    for stack in state.stacks.values():
        # there can only be one stack where this condition is true for each region
        # as there can only be one active stack with a given name
        if (
            stack_name_or_id in [stack.stack_name, stack.stack_id]
            and stack.status != "DELETE_COMPLETE"
        ):
            return stack
    return None


def find_change_set(
    account_id: str,
    region_name: str,
    cs_name: str,
    stack_name: Optional[str] = None,
    active_only: bool = False,
) -> Optional[StackChangeSet]:
    store = get_cloudformation_store(account_id, region_name)
    for stack in store.stacks.values():
        if active_only and stack.status == StackStatus.DELETE_COMPLETE:
            continue
        if stack_name in (stack.stack_name, stack.stack_id, None):
            for change_set in stack.change_sets:
                if cs_name in (change_set.change_set_id, change_set.change_set_name):
                    return change_set
    return None


def exports_map(account_id: str, region_name: str) -> dict[str, Export]:
    store = get_cloudformation_store(account_id, region_name)
    return {**store.exports, **store.exports_v2}

1	import logging	1✔
2	from typing import Optional	1✔
3
4	from localstack.aws.api.cloudformation import Export, StackStatus	1✔
5	from localstack.services.cloudformation.engine.entities import Stack, StackChangeSet, StackSet	1✔
6	from localstack.services.cloudformation.v2.entities import ChangeSet as ChangeSetV2	1✔
7	from localstack.services.cloudformation.v2.entities import Stack as StackV2	1✔
8	from localstack.services.cloudformation.v2.entities import StackSet as StackSetV2	1✔
9	from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute	1✔
10
11	LOG = logging.getLogger(__name__)	1✔
12
13
14	class CloudFormationStore(BaseStore):	1✔
15	# maps stack ID to stack details
16	stacks: dict[str, Stack] = LocalAttribute(default=dict)	1✔
17	stacks_v2: dict[str, StackV2] = LocalAttribute(default=dict)	1✔
18
19	change_sets: dict[str, ChangeSetV2] = LocalAttribute(default=dict)	1✔
20
21	# maps stack set ID to stack set details
22	stack_sets: dict[str, StackSet] = LocalAttribute(default=dict)	1✔
23	stack_sets_v2: dict[str, StackSetV2] = LocalAttribute(default=dict)	1✔
24
25	# maps macro ID to macros
26	macros: dict[str, dict] = LocalAttribute(default=dict)	1✔
27
28	@property	1✔
29	def exports(self) -> dict[str, Export]:	1✔
30	exports = {}	1✔
31
32	for stack_id, stack in self.stacks.items():	1✔
33	if stack.status == StackStatus.DELETE_COMPLETE:	1✔
34	continue	1✔
35
36	for output in stack.resolved_outputs:	1✔
37	export_name = output.get("ExportName")	1✔
38	if not export_name:	1✔
39	continue	1✔
40	if export_name in exports.keys():	1✔
41	# TODO: raise exception on stack creation in case of duplicate exports
UNCOV 42	LOG.warning(	×
43	"Found duplicate export name %s in stacks: %s %s",
44	export_name,
45	output["OutputValue"],
46	stack.stack_id,
47	)
48	exports[export_name] = Export(	1✔
49	ExportingStackId=stack.stack_id, Name=export_name, Value=output["OutputValue"]
50	)
51
52	return exports	1✔
53
54	@property	1✔
55	def exports_v2(self) -> dict[str, Export]:	1✔
56	exports = dict()	1✔
57	stacks_v2 = self.stacks_v2.values()	1✔
58	for stack in stacks_v2:	1✔
59	if stack.status == StackStatus.DELETE_COMPLETE:	1✔
60	continue	1✔
61	for export_name, export_value in stack.resolved_exports.items():	1✔
62	exports[export_name] = Export(	1✔
63	ExportingStackId=stack.stack_id, Name=export_name, Value=export_value
64	)
65
66	return exports	1✔
67
68
69	cloudformation_stores = AccountRegionBundle("cloudformation", CloudFormationStore)	1✔
70
71
72	def get_cloudformation_store(account_id: str, region_name: str) -> CloudFormationStore:	1✔
73	return cloudformation_stores[account_id][region_name]	1✔
74
75
76	# TODO: rework / fix usage of this
77	def find_stack(account_id: str, region_name: str, stack_name: str) -> Stack \| None:	1✔
78	# Warning: This function may not return the correct stack if multiple stacks with same name exist.
79	state = get_cloudformation_store(account_id, region_name)	1✔
80	return (	1✔
81	[s for s in state.stacks.values() if stack_name in [s.stack_name, s.stack_id]] or [None]
82	)[0]
83
84
85	def find_stack_by_id(account_id: str, region_name: str, stack_id: str) -> Stack \| None:	1✔
86	"""
87	Find the stack by id.
88
89	:param account_id: account of the stack
90	:param region_name: region of the stack
91	:param stack_id: stack id
92	:return: Stack if it is found, None otherwise
93	"""
94	state = get_cloudformation_store(account_id, region_name)	1✔
95	for stack in state.stacks.values():	1✔
96	# there can only be one stack with an id
97	if stack_id == stack.stack_id:	1✔
98	return stack	1✔
99	return None	1✔
100
101
102	def find_active_stack_by_name_or_id(	1✔
103	account_id: str, region_name: str, stack_name_or_id: str
104	) -> Stack \| None:
105	"""
106	Find the active stack by name. Some cloudformation operations only allow referencing by slack name if the stack is
107	"active", which we currently interpret as not DELETE_COMPLETE.
108
109	:param account_id: account of the stack
110	:param region_name: region of the stack
111	:param stack_name_or_id: stack name or stack id
112	:return: Stack if it is found, None otherwise
113	"""
114	state = get_cloudformation_store(account_id, region_name)	1✔
115	for stack in state.stacks.values():	1✔
116	# there can only be one stack where this condition is true for each region
117	# as there can only be one active stack with a given name
118	if (	1✔
119	stack_name_or_id in [stack.stack_name, stack.stack_id]
120	and stack.status != "DELETE_COMPLETE"
121	):
122	return stack	1✔
123	return None	1✔
124
125
126	def find_change_set(	1✔
127	account_id: str,
128	region_name: str,
129	cs_name: str,
130	stack_name: Optional[str] = None,
131	active_only: bool = False,
132	) -> Optional[StackChangeSet]:
133	store = get_cloudformation_store(account_id, region_name)	1✔
134	for stack in store.stacks.values():	1✔
135	if active_only and stack.status == StackStatus.DELETE_COMPLETE:	1✔
136	continue	1✔
137	if stack_name in (stack.stack_name, stack.stack_id, None):	1✔
138	for change_set in stack.change_sets:	1✔
139	if cs_name in (change_set.change_set_id, change_set.change_set_name):	1✔
140	return change_set	1✔
141	return None	1✔
142
143
144	def exports_map(account_id: str, region_name: str) -> dict[str, Export]:	1✔
145	store = get_cloudformation_store(account_id, region_name)	1✔
146	return {store.exports, store.exports_v2}	1✔

localstack / localstack / 16766254352

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