21418642313

Committed 27 Jan 2026 11:40PM UTC coverage: 74.0% (-0.7%) from 74.747%

Build # 21418642313

Build Type

Pull #41

github

Committed by

web-flow

Commit Message

Merge a6043de30 into a8b6a9122

Pull Request Pull Request #41: [IT-4759] Update Python version

Run Details

74 of 100 relevant lines covered (74.0%)

0.74 hits per line

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

74.0

/explode/app.py

"""
CloudFormation template transform macro: Explode
from: https://github.com/awslabs/aws-cloudformation-templates/tree/master/aws/services/CloudFormation/MacrosExamples/Explode
"""

import re
import sys


EXPLODE_RE = re.compile(r'(?i)!Explode (?P<explode_key>\w+)')


def walk_resource(resource, map_data):
    """Recursively process a resource."""
    if isinstance(resource, dict):
        new_resource = {}
        for key, value in resource.items():
            if isinstance(value, dict):
                new_resource[key] = walk_resource(value, map_data)
            elif isinstance(value, list):
                new_resource[key] = [walk_resource(x, map_data) for x in value]
            elif isinstance(value, str):
                new_resource[key] = replace_explode_in_string(value, map_data)
            else:
                new_resource[key] = value
    else:
        # if the resource is of type string
        new_resource = replace_explode_in_string(resource, map_data)
    return new_resource


def replace_explode_in_string(value, map_data):
    """Recursively process and replace Explode instances in a string."""
    match = EXPLODE_RE.search(value)
    while match:
        explode_key = match.group('explode_key')
        try:
            replace_value = map_data[explode_key]
        except KeyError:
            print("Missing item {} in mapping while processing {}: {}".format(
                explode_key,
                match,
                value))
        if isinstance(replace_value, int):
            value = replace_value
            # No further explosion is possible on an int
            match = None
        else:
            value = value.replace(match.group(0), replace_value)
            match = EXPLODE_RE.search(value)
    return value


def handle_transform(template, parameters):
    """Go through template and explode resources."""
    resources = template['Resources']
    new_resources = {}
    for resource_name, resource in resources.items():
        try:
            explode_map = resource['ExplodeMap']
            mappings = template['Mappings']
            del resource['ExplodeMap']
        except KeyError:
            try:
                explode_map = resource['ExplodeParam']
                mappings = format_params(parameters, explode_map)
                del resource['ExplodeParam']
            except KeyError:
                new_resources[resource_name] = resource
                continue
        try:
            explode_map_data = mappings[explode_map]
        except KeyError:
            # This resource refers to a mapping entry which doesn't exist, so
            # fail
            print('Unable to find mapping for exploding resource {}'.format(resource_name))
            raise
        resource_instances = explode_map_data.keys()
        for resource_instance in resource_instances:
            new_resource = walk_resource(resource, explode_map_data[resource_instance])
            if 'ResourceName' in explode_map_data[resource_instance]:
                new_resource_name = explode_map_data[resource_instance]['ResourceName']
            else:
                new_resource_name = resource_name + resource_instance
            new_resources[new_resource_name] = new_resource
    template['Resources'] = new_resources
    return template

def format_params(parameters, name):
    """Format parameters from a list to a dict"""
    items = parameters[name]
    new_params = {}
    new_params.update({name:{}})
    for item in items:
        temp = {}
        key_name = name[:-1]
        key_list_name = key_name + str(item)
        temp.update({key_list_name: {key_name: item}})
        new_params[name].update(temp)

    return new_params

def handler(event, _context):
    """Handle invocation in Lambda (when CloudFormation processes the Macro)"""
    fragment = event["fragment"]
    parameters = event["templateParameterValues"]
    status = "success"

    try:
        fragment = handle_transform(event["fragment"], parameters)
    except:
        status = "failure"

    return {
        "requestId": event["requestId"],
        "status": status,
        "fragment": fragment,
    }


if __name__ == "__main__":
    """
    If run from the command line, parse the file specified and output it
    This is quite naive; CF YAML tags like !GetAtt will break it (as will
    !Explode, but you can hide that in a string). Probably best to use JSON.
    Releatedly, always outputs JSON.
    """
    if len(sys.argv) == 2:
        import json
        filename = sys.argv[1]
        if filename.endswith(".yml") or filename.endswith(".yaml"):
            try:
                import yaml
            except ImportError:
                print("Please install PyYAML to test yaml templates")
                sys.exit(1)
            with open(filename, 'r') as file_handle:
                loaded_fragment = yaml.safe_load(file_handle)
        elif filename.endswith(".json"):
            with open(sys.argv[1], 'r') as file_handle:
                loaded_fragment = json.load(file_handle)
        else:
            print("Test file needs to end .yaml, .yml or .json")
            sys.exit(1)
        new_fragment = handle_transform(loaded_fragment)
        print(json.dumps(new_fragment))

1	"""
2	CloudFormation template transform macro: Explode
3	from: https://github.com/awslabs/aws-cloudformation-templates/tree/master/aws/services/CloudFormation/MacrosExamples/Explode
4	"""
5
6	import re	1✔
7	import sys	1✔
8
9
10	EXPLODE_RE = re.compile(r'(?i)!Explode (?P<explode_key>\w+)')	1✔
11
12
13	def walk_resource(resource, map_data):	1✔
14	"""Recursively process a resource."""
15	if isinstance(resource, dict):	1✔
16	new_resource = {}	1✔
17	for key, value in resource.items():	1✔
18	if isinstance(value, dict):	1✔
19	new_resource[key] = walk_resource(value, map_data)	1✔
20	elif isinstance(value, list):	1✔
21	new_resource[key] = [walk_resource(x, map_data) for x in value]	1✔
22	elif isinstance(value, str):	1✔
23	new_resource[key] = replace_explode_in_string(value, map_data)	1✔
24	else:
25	new_resource[key] = value	×
26	else:
27	# if the resource is of type string
28	new_resource = replace_explode_in_string(resource, map_data)	×
29	return new_resource	1✔
30
31
32	def replace_explode_in_string(value, map_data):	1✔
33	"""Recursively process and replace Explode instances in a string."""
34	match = EXPLODE_RE.search(value)	1✔
35	while match:	1✔
36	explode_key = match.group('explode_key')	1✔
37	try:	1✔
38	replace_value = map_data[explode_key]	1✔
39	except KeyError:	1✔
40	print("Missing item {} in mapping while processing {}: {}".format(	1✔
41	explode_key,
42	match,
43	value))
44	if isinstance(replace_value, int):	1✔
45	value = replace_value	1✔
46	# No further explosion is possible on an int
47	match = None	1✔
48	else:
49	value = value.replace(match.group(0), replace_value)	×
50	match = EXPLODE_RE.search(value)	×
51	return value	1✔
52
53
54	def handle_transform(template, parameters):	1✔
55	"""Go through template and explode resources."""
56	resources = template['Resources']	1✔
57	new_resources = {}	1✔
58	for resource_name, resource in resources.items():	1✔
59	try:	1✔
60	explode_map = resource['ExplodeMap']	1✔
61	mappings = template['Mappings']	1✔
62	del resource['ExplodeMap']	1✔
63	except KeyError:	1✔
64	try:	1✔
65	explode_map = resource['ExplodeParam']	1✔
66	mappings = format_params(parameters, explode_map)	1✔
67	del resource['ExplodeParam']	1✔
68	except KeyError:	1✔
69	new_resources[resource_name] = resource	1✔
70	continue	1✔
71	try:	1✔
72	explode_map_data = mappings[explode_map]	1✔
73	except KeyError:	×
74	# This resource refers to a mapping entry which doesn't exist, so
75	# fail
76	print('Unable to find mapping for exploding resource {}'.format(resource_name))	×
77	raise	×
78	resource_instances = explode_map_data.keys()	1✔
79	for resource_instance in resource_instances:	1✔
80	new_resource = walk_resource(resource, explode_map_data[resource_instance])	1✔
81	if 'ResourceName' in explode_map_data[resource_instance]:	1✔
82	new_resource_name = explode_map_data[resource_instance]['ResourceName']	1✔
83	else:
84	new_resource_name = resource_name + resource_instance	1✔
85	new_resources[new_resource_name] = new_resource	1✔
86	template['Resources'] = new_resources	1✔
87	return template	1✔
88
89	def format_params(parameters, name):	1✔
90	"""Format parameters from a list to a dict"""
91	items = parameters[name]	1✔
92	new_params = {}	1✔
93	new_params.update({name:{}})	1✔
94	for item in items:	1✔
95	temp = {}	1✔
96	key_name = name[:-1]	1✔
97	key_list_name = key_name + str(item)	1✔
98	temp.update({key_list_name: {key_name: item}})	1✔
99	new_params[name].update(temp)	1✔
100
101	return new_params	1✔
102
103	def handler(event, _context):	1✔
104	"""Handle invocation in Lambda (when CloudFormation processes the Macro)"""
105	fragment = event["fragment"]	1✔
106	parameters = event["templateParameterValues"]	1✔
107	status = "success"	1✔
108
109	try:	1✔
110	fragment = handle_transform(event["fragment"], parameters)	1✔
111	except:	1✔
112	status = "failure"	1✔
113
114	return {	1✔
115	"requestId": event["requestId"],
116	"status": status,
117	"fragment": fragment,
118	}
119
120
121	if __name__ == "__main__":	1✔
122	"""	×
123	If run from the command line, parse the file specified and output it
124	This is quite naive; CF YAML tags like !GetAtt will break it (as will
125	!Explode, but you can hide that in a string). Probably best to use JSON.
126	Releatedly, always outputs JSON.
127	"""
128	if len(sys.argv) == 2:	×
129	import json	×
130	filename = sys.argv[1]	×
131	if filename.endswith(".yml") or filename.endswith(".yaml"):	×
132	try:	×
133	import yaml	×
134	except ImportError:	×
135	print("Please install PyYAML to test yaml templates")	×
136	sys.exit(1)	×
137	with open(filename, 'r') as file_handle:	×
138	loaded_fragment = yaml.safe_load(file_handle)	×
139	elif filename.endswith(".json"):	×
140	with open(sys.argv[1], 'r') as file_handle:	×
141	loaded_fragment = json.load(file_handle)	×
142	else:
143	print("Test file needs to end .yaml, .yml or .json")	×
144	sys.exit(1)	×
145	new_fragment = handle_transform(loaded_fragment)	×
146	print(json.dumps(new_fragment))	×

Sage-Bionetworks-IT / cfn-explode-macro / 21418642313

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