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

57.14

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

import logging

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

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