22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

36.84

/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✔
UNCOV 7	return node.value	×
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	"""
UNCOV 35	match tag_suffix:	×
UNCOV 36	case "Ref":	×
UNCOV 37	fn_name = "Ref"	×
UNCOV 38	case "Condition":	×
UNCOV 39	fn_name = "Condition"	×
UNCOV 40	case _:	×
UNCOV 41	fn_name = f"Fn::{tag_suffix}"	×
42
UNCOV 43	if tag_suffix == "GetAtt" and isinstance(node, yaml.ScalarNode):	×
44	# !GetAtt A.B.C => {"Fn::GetAtt": ["A", "B.C"]}
UNCOV 45	parts = node.value.partition(".")	×
UNCOV 46	if len(parts) != 3:	×
47	raise ValueError(f"Node value contains unexpected format for !GetAtt: {parts}")	×
UNCOV 48	return {fn_name: [parts[0], parts[2]]}	×
49
UNCOV 50	if isinstance(node, yaml.ScalarNode):	×
UNCOV 51	return {fn_name: node.value}	×
UNCOV 52	elif isinstance(node, yaml.SequenceNode):	×
UNCOV 53	return {fn_name: loader.construct_sequence(node)}	×
UNCOV 54	elif isinstance(node, yaml.MappingNode):	×
UNCOV 55	return {fn_name: loader.construct_mapping(node)}	×
56	else:
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✔
UNCOV 66	parsed = yaml.load(input_data, Loader=customloader)	×
67
UNCOV 68	if not isinstance(parsed, dict):	×
UNCOV 69	raise ValidationError("Template format error: unsupported structure.")	×
70
UNCOV 71	return parsed	×

localstack / localstack / 22709357475

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