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

96.3

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

from collections import OrderedDict

from localstack.services.cloudformation.engine.changes import ChangeConfig
from localstack.services.cloudformation.engine.parameters import StackParameter
from localstack.services.cloudformation.engine.template_utils import get_deps_for_resource


class NoResourceInStack(ValueError):
    """Raised when we preprocess the template and do not find a resource"""

    def __init__(self, logical_resource_id: str):
        msg = f"Template format error: Unresolved resource dependencies [{logical_resource_id}] in the Resources block of the template"

        super().__init__(msg)


def order_resources(
    resources: dict,
    resolved_parameters: dict[str, StackParameter],
    resolved_conditions: dict[str, bool],
    reverse: bool = False,
) -> OrderedDict:
    """
    Given a dictionary of resources, topologically sort the resources based on
    inter-resource dependencies (e.g. usages of intrinsic functions).
    """
    nodes: dict[str, list[str]] = {}
    for logical_resource_id, properties in resources.items():
        nodes.setdefault(logical_resource_id, [])
        deps = get_deps_for_resource(properties, resolved_conditions)
        for dep in deps:
            if dep in resolved_parameters:
                # we only care about other resources
                continue
            nodes.setdefault(dep, [])
            nodes[dep].append(logical_resource_id)

    # implementation from https://dev.to/leopfeiffer/topological-sort-with-kahns-algorithm-3dl1
    indegrees = dict.fromkeys(nodes.keys(), 0)
    for dependencies in nodes.values():
        for dependency in dependencies:
            indegrees[dependency] += 1

    # Place all elements with indegree 0 in queue
    queue = [k for k in nodes.keys() if indegrees[k] == 0]

    sorted_logical_resource_ids = []

    # Continue until all nodes have been dealt with
    while len(queue) > 0:
        # node of current iteration is the first one from the queue
        curr = queue.pop(0)
        sorted_logical_resource_ids.append(curr)

        # remove the current node from other dependencies
        for dependency in nodes[curr]:
            indegrees[dependency] -= 1

            if indegrees[dependency] == 0:
                queue.append(dependency)

    # check for circular dependencies
    if len(sorted_logical_resource_ids) != len(nodes):
        raise Exception("Circular dependency found.")

    sorted_mapping = []
    for logical_resource_id in sorted_logical_resource_ids:
        if properties := resources.get(logical_resource_id):
            sorted_mapping.append((logical_resource_id, properties))
        else:
            if (
                logical_resource_id not in resolved_parameters
                and logical_resource_id not in resolved_conditions
            ):
                raise NoResourceInStack(logical_resource_id)

    if reverse:
        sorted_mapping = sorted_mapping[::-1]
    return OrderedDict(sorted_mapping)


def order_changes(
    given_changes: list[ChangeConfig],
    resources: dict,
    resolved_parameters: dict[str, StackParameter],
    # TODO: remove resolved conditions somehow
    resolved_conditions: dict[str, bool],
    reverse: bool = False,
) -> list[ChangeConfig]:
    """
    Given a list of changes, a dictionary of resources and a dictionary of resolved conditions, topologically sort the
    changes based on inter-resource dependencies (e.g. usages of intrinsic functions).
    """
    ordered_resources = order_resources(
        resources=resources,
        resolved_parameters=resolved_parameters,
        resolved_conditions=resolved_conditions,
        reverse=reverse,
    )
    sorted_changes = []
    for logical_resource_id in ordered_resources.keys():
        for change in given_changes:
            if change["ResourceChange"]["LogicalResourceId"] == logical_resource_id:
                sorted_changes.append(change)
                break
    assert len(sorted_changes) > 0
    if reverse:
        sorted_changes = sorted_changes[::-1]
    return sorted_changes

1	from collections import OrderedDict	2✔
2
3	from localstack.services.cloudformation.engine.changes import ChangeConfig	2✔
4	from localstack.services.cloudformation.engine.parameters import StackParameter	2✔
5	from localstack.services.cloudformation.engine.template_utils import get_deps_for_resource	2✔
6
7
8	class NoResourceInStack(ValueError):	2✔
9	"""Raised when we preprocess the template and do not find a resource"""
10
11	def __init__(self, logical_resource_id: str):	2✔
UNCOV 12	msg = f"Template format error: Unresolved resource dependencies [{logical_resource_id}] in the Resources block of the template"	1✔
13
UNCOV 14	super().__init__(msg)	1✔
15
16
17	def order_resources(	2✔
18	resources: dict,
19	resolved_parameters: dict[str, StackParameter],
20	resolved_conditions: dict[str, bool],
21	reverse: bool = False,
22	) -> OrderedDict:
23	"""
24	Given a dictionary of resources, topologically sort the resources based on
25	inter-resource dependencies (e.g. usages of intrinsic functions).
26	"""
27	nodes: dict[str, list[str]] = {}	2✔
28	for logical_resource_id, properties in resources.items():	2✔
29	nodes.setdefault(logical_resource_id, [])	2✔
30	deps = get_deps_for_resource(properties, resolved_conditions)	2✔
31	for dep in deps:	2✔
32	if dep in resolved_parameters:	2✔
33	# we only care about other resources
UNCOV 34	continue	1✔
35	nodes.setdefault(dep, [])	2✔
36	nodes[dep].append(logical_resource_id)	2✔
37
38	# implementation from https://dev.to/leopfeiffer/topological-sort-with-kahns-algorithm-3dl1
39	indegrees = dict.fromkeys(nodes.keys(), 0)	2✔
40	for dependencies in nodes.values():	2✔
41	for dependency in dependencies:	2✔
42	indegrees[dependency] += 1	2✔
43
44	# Place all elements with indegree 0 in queue
45	queue = [k for k in nodes.keys() if indegrees[k] == 0]	2✔
46
47	sorted_logical_resource_ids = []	2✔
48
49	# Continue until all nodes have been dealt with
50	while len(queue) > 0:	2✔
51	# node of current iteration is the first one from the queue
52	curr = queue.pop(0)	2✔
53	sorted_logical_resource_ids.append(curr)	2✔
54
55	# remove the current node from other dependencies
56	for dependency in nodes[curr]:	2✔
57	indegrees[dependency] -= 1	2✔
58
59	if indegrees[dependency] == 0:	2✔
60	queue.append(dependency)	2✔
61
62	# check for circular dependencies
63	if len(sorted_logical_resource_ids) != len(nodes):	2✔
64	raise Exception("Circular dependency found.")	×
65
66	sorted_mapping = []	2✔
67	for logical_resource_id in sorted_logical_resource_ids:	2✔
68	if properties := resources.get(logical_resource_id):	2✔
69	sorted_mapping.append((logical_resource_id, properties))	2✔
70	else:
UNCOV 71	if (	1✔
72	logical_resource_id not in resolved_parameters
73	and logical_resource_id not in resolved_conditions
74	):
UNCOV 75	raise NoResourceInStack(logical_resource_id)	1✔
76
77	if reverse:	2✔
UNCOV 78	sorted_mapping = sorted_mapping[::-1]	1✔
79	return OrderedDict(sorted_mapping)	2✔
80
81
82	def order_changes(	2✔
83	given_changes: list[ChangeConfig],
84	resources: dict,
85	resolved_parameters: dict[str, StackParameter],
86	# TODO: remove resolved conditions somehow
87	resolved_conditions: dict[str, bool],
88	reverse: bool = False,
89	) -> list[ChangeConfig]:
90	"""
91	Given a list of changes, a dictionary of resources and a dictionary of resolved conditions, topologically sort the
92	changes based on inter-resource dependencies (e.g. usages of intrinsic functions).
93	"""
UNCOV 94	ordered_resources = order_resources(	1✔
95	resources=resources,
96	resolved_parameters=resolved_parameters,
97	resolved_conditions=resolved_conditions,
98	reverse=reverse,
99	)
UNCOV 100	sorted_changes = []	1✔
UNCOV 101	for logical_resource_id in ordered_resources.keys():	1✔
UNCOV 102	for change in given_changes:	1✔
UNCOV 103	if change["ResourceChange"]["LogicalResourceId"] == logical_resource_id:	1✔
UNCOV 104	sorted_changes.append(change)	1✔
UNCOV 105	break	1✔
UNCOV 106	assert len(sorted_changes) > 0	1✔
UNCOV 107	if reverse:	1✔
108	sorted_changes = sorted_changes[::-1]	×
UNCOV 109	return sorted_changes	1✔

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