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

66.67

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

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