Recached

18895305385

Committed 28 Oct 2025 05:02PM UTC coverage: 86.9% (+0.008%) from 86.892%

Build # 18895305385

Build Type

push

github

Committed by

web-flow

Commit Message

Increase DNS server startup priority (#13314)

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

80 existing lines in 3 files now uncovered.

68360 of 78665 relevant lines covered (86.9%)

0.87 hits per line

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

94.74

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

import yaml

from localstack.services.cloudformation.engine.validations import ValidationError


def construct_raw(_, node):
    return node.value


class NoDatesSafeLoader(yaml.SafeLoader):
    @classmethod
    def remove_tag_constructor(cls, tag):
        """
        Remove the YAML constructor for a given tag and replace it with a raw constructor
        """
        # needed to make sure we're not changing the constructors of the base class
        # otherwise usage across the code base is affected as well
        if "yaml_constructors" not in cls.__dict__:
            cls.yaml_constructors = cls.yaml_constructors.copy()

        cls.yaml_constructors[tag] = construct_raw


NoDatesSafeLoader.remove_tag_constructor("tag:yaml.org,2002:timestamp")


def shorthand_constructor(loader: yaml.Loader, tag_suffix: str, node: yaml.Node):
    """
    TODO: proper exceptions (introduce this when fixing the provider)
    TODO: fix select & split (is this even necessary?)
    { "Fn::Select" : [ "2", { "Fn::Split": [",", {"Fn::ImportValue": "AccountSubnetIDs"}]}] }
    !Select [2, !Split [",", !ImportValue AccountSubnetIDs]]
    shorthand: 2 => canonical "2"
    """
    match tag_suffix:
        case "Ref":
            fn_name = "Ref"
        case "Condition":
            fn_name = "Condition"
        case _:
            fn_name = f"Fn::{tag_suffix}"

    if tag_suffix == "GetAtt" and isinstance(node, yaml.ScalarNode):
        # !GetAtt A.B.C => {"Fn::GetAtt": ["A", "B.C"]}
        parts = node.value.partition(".")
        if len(parts) != 3:
            raise ValueError(f"Node value contains unexpected format for !GetAtt: {parts}")
        return {fn_name: [parts[0], parts[2]]}

    if isinstance(node, yaml.ScalarNode):
        return {fn_name: node.value}
    elif isinstance(node, yaml.SequenceNode):
        return {fn_name: loader.construct_sequence(node)}
    elif isinstance(node, yaml.MappingNode):
        return {fn_name: loader.construct_mapping(node)}
    else:
        raise ValueError(f"Unexpected yaml Node type: {type(node)}")


customloader = NoDatesSafeLoader

yaml.add_multi_constructor("!", shorthand_constructor, customloader)


def parse_yaml(input_data: str) -> dict:
    parsed = yaml.load(input_data, Loader=customloader)

    if not isinstance(parsed, dict):
        raise ValidationError("Template format error: unsupported structure.")

    return parsed

1	import yaml	1✔
2
3	from localstack.services.cloudformation.engine.validations import ValidationError	1✔
4
5
6	def construct_raw(_, node):	1✔
7	return node.value	1✔
8
9
10	class NoDatesSafeLoader(yaml.SafeLoader):	1✔
11	@classmethod	1✔
12	def remove_tag_constructor(cls, tag):	1✔
13	"""
14	Remove the YAML constructor for a given tag and replace it with a raw constructor
15	"""
16	# needed to make sure we're not changing the constructors of the base class
17	# otherwise usage across the code base is affected as well
18	if "yaml_constructors" not in cls.__dict__:	1✔
19	cls.yaml_constructors = cls.yaml_constructors.copy()	1✔
20
21	cls.yaml_constructors[tag] = construct_raw	1✔
22
23
24	NoDatesSafeLoader.remove_tag_constructor("tag:yaml.org,2002:timestamp")	1✔
25
26
27	def shorthand_constructor(loader: yaml.Loader, tag_suffix: str, node: yaml.Node):	1✔
28	"""
29	TODO: proper exceptions (introduce this when fixing the provider)
30	TODO: fix select & split (is this even necessary?)
31	{ "Fn::Select" : [ "2", { "Fn::Split": [",", {"Fn::ImportValue": "AccountSubnetIDs"}]}] }
32	!Select [2, !Split [",", !ImportValue AccountSubnetIDs]]
33	shorthand: 2 => canonical "2"
34	"""
35	match tag_suffix:	1✔
36	case "Ref":	1✔
37	fn_name = "Ref"	1✔
38	case "Condition":	1✔
39	fn_name = "Condition"	1✔
40	case _:	1✔
41	fn_name = f"Fn::{tag_suffix}"	1✔
42
43	if tag_suffix == "GetAtt" and isinstance(node, yaml.ScalarNode):	1✔
44	# !GetAtt A.B.C => {"Fn::GetAtt": ["A", "B.C"]}
45	parts = node.value.partition(".")	1✔
46	if len(parts) != 3:	1✔
UNCOV 47	raise ValueError(f"Node value contains unexpected format for !GetAtt: {parts}")	×
48	return {fn_name: [parts[0], parts[2]]}	1✔
49
50	if isinstance(node, yaml.ScalarNode):	1✔
51	return {fn_name: node.value}	1✔
52	elif isinstance(node, yaml.SequenceNode):	1✔
53	return {fn_name: loader.construct_sequence(node)}	1✔
54	elif isinstance(node, yaml.MappingNode):	1✔
55	return {fn_name: loader.construct_mapping(node)}	1✔
56	else:
UNCOV 57	raise ValueError(f"Unexpected yaml Node type: {type(node)}")	×
58
59
60	customloader = NoDatesSafeLoader	1✔
61
62	yaml.add_multi_constructor("!", shorthand_constructor, customloader)	1✔
63
64
65	def parse_yaml(input_data: str) -> dict:	1✔
66	parsed = yaml.load(input_data, Loader=customloader)	1✔
67
68	if not isinstance(parsed, dict):	1✔
69	raise ValidationError("Template format error: unsupported structure.")	1✔
70
71	return parsed	1✔

localstack / localstack / 18895305385

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