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)

Run Details

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

98.57

/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	2✔
2
3	from localstack.aws.api.cloudformation import Export, StackStatus	2✔
4	from localstack.services.cloudformation.engine.entities import Stack, StackChangeSet, StackSet	2✔
5	from localstack.services.cloudformation.v2.entities import ChangeSet as ChangeSetV2	2✔
6	from localstack.services.cloudformation.v2.entities import Stack as StackV2	2✔
7	from localstack.services.cloudformation.v2.entities import StackSet as StackSetV2	2✔
8	from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute	2✔
9
10	LOG = logging.getLogger(__name__)	2✔
11
12
13	class CloudFormationStore(BaseStore):	2✔
14	# maps stack ID to stack details
15	stacks: dict[str, Stack] = LocalAttribute(default=dict)	2✔
16	stacks_v2: dict[str, StackV2] = LocalAttribute(default=dict)	2✔
17
18	change_sets: dict[str, ChangeSetV2] = LocalAttribute(default=dict)	2✔
19
20	# maps stack set ID to stack set details
21	stack_sets: dict[str, StackSet] = LocalAttribute(default=dict)	2✔
22	stack_sets_v2: dict[str, StackSetV2] = LocalAttribute(default=dict)	2✔
23
24	# maps macro ID to macros
25	macros: dict[str, dict] = LocalAttribute(default=dict)	2✔
26
27	@property	2✔
28	def exports(self) -> dict[str, Export]:	2✔
29	exports = {}	2✔
30
31	for stack_id, stack in self.stacks.items():	2✔
UNCOV 32	if stack.status == StackStatus.DELETE_COMPLETE:	1✔
UNCOV 33	continue	1✔
34
UNCOV 35	for output in stack.resolved_outputs:	1✔
UNCOV 36	export_name = output.get("ExportName")	1✔
UNCOV 37	if not export_name:	1✔
UNCOV 38	continue	1✔
UNCOV 39	if export_name in exports.keys():	1✔
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	)
UNCOV 47	exports[export_name] = Export(	1✔
48	ExportingStackId=stack.stack_id, Name=export_name, Value=output["OutputValue"]
49	)
50
51	return exports	2✔
52
53	@property	2✔
54	def exports_v2(self) -> dict[str, Export]:	2✔
55	exports = {}	2✔
56	stacks_v2 = self.stacks_v2.values()	2✔
57	for stack in stacks_v2:	2✔
58	if stack.status == StackStatus.DELETE_COMPLETE:	2✔
59	continue	2✔
60	for export_name, export_value in stack.resolved_exports.items():	2✔
61	exports[export_name] = Export(	2✔
62	ExportingStackId=stack.stack_id, Name=export_name, Value=export_value
63	)
64
65	return exports	2✔
66
67
68	cloudformation_stores = AccountRegionBundle("cloudformation", CloudFormationStore)	2✔
69
70
71	def get_cloudformation_store(account_id: str, region_name: str) -> CloudFormationStore:	2✔
72	return cloudformation_stores[account_id][region_name]	2✔
73
74
75	# TODO: rework / fix usage of this
76	def find_stack(account_id: str, region_name: str, stack_name: str) -> Stack \| None:	2✔
77	# Warning: This function may not return the correct stack if multiple stacks with same name exist.
UNCOV 78	state = get_cloudformation_store(account_id, region_name)	1✔
UNCOV 79	return (	1✔
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:	2✔
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	"""
UNCOV 93	state = get_cloudformation_store(account_id, region_name)	1✔
UNCOV 94	for stack in state.stacks.values():	1✔
95	# there can only be one stack with an id
UNCOV 96	if stack_id == stack.stack_id:	1✔
UNCOV 97	return stack	1✔
UNCOV 98	return None	1✔
99
100
101	def find_active_stack_by_name_or_id(	2✔
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	"""
UNCOV 113	state = get_cloudformation_store(account_id, region_name)	1✔
UNCOV 114	for stack in state.stacks.values():	1✔
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
UNCOV 117	if (	1✔
118	stack_name_or_id in [stack.stack_name, stack.stack_id]
119	and stack.status != "DELETE_COMPLETE"
120	):
UNCOV 121	return stack	1✔
UNCOV 122	return None	1✔
123
124
125	def find_change_set(	2✔
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:
UNCOV 132	store = get_cloudformation_store(account_id, region_name)	1✔
UNCOV 133	for stack in store.stacks.values():	1✔
UNCOV 134	if active_only and stack.status == StackStatus.DELETE_COMPLETE:	1✔
UNCOV 135	continue	1✔
UNCOV 136	if stack_name in (stack.stack_name, stack.stack_id, None):	1✔
UNCOV 137	for change_set in stack.change_sets:	1✔
UNCOV 138	if cs_name in (change_set.change_set_id, change_set.change_set_name):	1✔
UNCOV 139	return change_set	1✔
UNCOV 140	return None	1✔
141
142
143	def exports_map(account_id: str, region_name: str) -> dict[str, Export]:	2✔
144	store = get_cloudformation_store(account_id, region_name)	2✔
145	return {store.exports, store.exports_v2}	2✔

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