234004684

Build Type

github-actions

Committed by GitHub

Commit Message

Merge c44351568 into eeddff26c

Pull Request Pull Request #160: Implement s3 Hashnaming and artifacts option

Run Details

360 of 381 branches covered (94.49%)

Branch coverage included in aggregate %.

97 of 107 new or added lines in 6 files covered. (90.65%)

8 existing lines in 1 file now uncovered.

1172 of 1206 relevant lines covered (97.18%)

1.94 hits per line

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

97.62

/formica/loader.py

import json
import os
import sys
import re

import logging
import yaml
from jinja2 import Environment, FileSystemLoader
from jinja2.exceptions import TemplateSyntaxError, TemplateNotFound, UndefinedError
import arrow
import fnmatch

from .exceptions import FormicaArgumentException

from . import yaml_tags
from .helper import main_account_id

# To silence pyflakes warning of unused import
assert yaml_tags

logger = logging.getLogger(__name__)

FILE_TYPES = ["yml", "yaml", "json"]

RESOURCES_KEY = "Resources"
MODULE_KEY = "From"

ALLOWED_ATTRIBUTES = {
    "AWSTemplateFormatVersion": [str],
    "Description": [str],
    "Metadata": dict,
    "Parameters": dict,
    "Mappings": dict,
    "Conditions": dict,
    "Transform": [str, list],
    RESOURCES_KEY: dict,
    "Outputs": dict,
}


def code_escape(source):
    return source.replace("\n", "\\n").replace('"', '\\"')


def code_array(source):
    lines = source.split("\\n")
    return "[" + ",".join(['"%s"' % line for line in lines]) + "]"


def mandatory(a):
    from jinja2.runtime import Undefined

    if isinstance(a, Undefined) or a is None:
        raise FormicaArgumentException("Mandatory variable not set.")
    return a


def resource(name):
    if name is None:
        return ""
    else:
        return "".join(e for e in name.title() if e.isalnum())


def novalue(variable):
    return variable or '{"Ref": "AWS::NoValue"}'


class Loader(object):
    def __init__(self, path=".", filename="*", variables=None, main_account_parameter=False):
        if variables is None:
            variables = {}
        self.cftemplate = {}
        self.path = path
        self.filename = filename
        self.env = Environment(loader=FileSystemLoader("./", followlinks=True))
        self.env.filters.update(
            {
                "code_escape": code_escape,
                "code_array": code_array,
                "mandatory": mandatory,
                "resource": resource,
                "novalue": novalue,
            }
        )
        self.variables = variables
        self.main_account_parameter = main_account_parameter

    def include_file(self, filename, **args):
        source = self.render(filename, **args)
        return code_escape(source)

    def load_file(self, filename, **args):
        return self.render(filename, **args)

    def list_files(self, filter="*"):
        return [t for t in self.env.list_templates() if fnmatch.fnmatch(t, filter)]

    def render(self, filename, **args):
        template_path = os.path.normpath("{}/{}".format(self.path, filename))
        template = self.env.get_template(template_path)
        variables = {}
        variables.update(self.variables)
        variables.update(args)
        arguments = dict(
            code=self.include_file,
            file=self.load_file,
            files=self.list_files,
            now=arrow.now,
            utcnow=arrow.utcnow,
            **variables,
        )
        return template.render(**arguments)

    def template(self, indent=4, sort_keys=True, separators=(",", ":"), dumper=None):
        if dumper is not None:
            return dumper(self.cftemplate)
        return json.dumps(self.cftemplate, indent=indent, sort_keys=sort_keys, separators=separators)

    def template_dictionary(self):
        return self.cftemplate

    def merge(self, template, file):
        if template:
            for key in template.keys():
                new = template[key]
                new_type = type(new)
                types = ALLOWED_ATTRIBUTES.get(key)
                if type(types) != list:
                    types = [types]
                if key in ALLOWED_ATTRIBUTES.keys() and new_type in types:
                    if new_type == str or new_type == list:
                        self.cftemplate[key] = new
                    elif new_type == dict:
                        for element_key, element_value in template[key].items():
                            if (
                                key == RESOURCES_KEY
                                and isinstance(element_value, dict)
                                and MODULE_KEY in element_value
                            ):
                                self.load_module(element_value[MODULE_KEY], element_key, element_value)
                            else:
                                self.cftemplate.setdefault(key, {})[element_key] = element_value
                else:
                    logger.info("Key '{}' in file {} is not valid".format(key, file))
                    sys.exit(1)
        else:
            logger.info("File {} is empty".format(file))

    def load_module(self, module_path, element_key, element_value):
        path_elements = module_path.split("::")
        dir_pattern = re.compile(fnmatch.translate(path_elements.pop(0)), re.IGNORECASE)
        matched_dirs = [dir for dir in os.listdir(self.path) if dir_pattern.match(dir)]
        matched_dir = module_path
        if matched_dirs:
            matched_dir = matched_dirs[0]

        module_path = self.path + "/" + "/".join([matched_dir] + path_elements)

        file_name = "*"

        if not os.path.isdir(module_path):
            file_name = module_path.split("/")[-1]
            module_path = "/".join(module_path.split("/")[:-1])

        properties = element_value.get("Properties", {})
        properties["module_name"] = element_key
        vars = self.merge_variables(properties)

        loader = Loader(module_path, file_name, vars)
        loader.load()
        self.merge(loader.template_dictionary(), file=file_name)

    def merge_variables(self, module_vars):
        merged_vars = {}
        for k, v in self.variables.items():
            merged_vars[k] = v
        for k, v in module_vars.items():
            merged_vars[k] = v
        return merged_vars

    def load(self):
        files = []

        for file_type in FILE_TYPES:
            pattern = re.compile(fnmatch.translate("{}.template.{}".format(self.filename, file_type)), re.IGNORECASE)
            files.extend([filename for filename in os.listdir(self.path) if pattern.match(filename)])

        if not files:
            logger.info("Could not find any template files in {}".format(self.path))
            sys.exit(1)

        for file in files:
            try:
                result = str(self.render(os.path.basename(file), **self.variables))
                template = yaml.full_load(result)
            except TemplateNotFound as e:
                logger.info("File not found" + ": " + e.message)
                logger.info('In: "' + file + '"')
                sys.exit(1)
            except TemplateSyntaxError as e:
                logger.info(e.__class__.__name__ + ": " + e.message)
                logger.info('File: "' + (e.filename or file) + '", line ' + str(e.lineno))
                sys.exit(1)
            except UndefinedError as e:
                logger.info(e.__class__.__name__ + ": " + e.message)
                logger.info('In: "' + file + '"')
                sys.exit(1)
            except FormicaArgumentException as e:
                logger.info(e.__class__.__name__ + ": " + e.args[0])
                logger.info('For Template: "' + file + '"')
                logger.info("If you use it as a template make sure you're setting all necessary vars")
                sys.exit(1)
            except yaml.YAMLError as e:
                logger.info(e.__str__())
                logger.info("Following is the Yaml document formica is trying to load:")
                logger.info("---------------------------------------------------------------------------")
                logger.info(result)
                logger.info("---------------------------------------------------------------------------")
                sys.exit(1)
            self.merge(template, file)

        if self.main_account_parameter:
            self.cftemplate["Parameters"] = self.cftemplate.get("Parameters") or {}
            self.cftemplate["Parameters"]["MainAccount"] = {"Type": "String", "Default": main_account_id()}

1	import json	2✔
2	import os	2✔
3	import sys	2✔
4	import re	2✔
5
6	import logging	2✔
7	import yaml	2✔
8	from jinja2 import Environment, FileSystemLoader	2✔
9	from jinja2.exceptions import TemplateSyntaxError, TemplateNotFound, UndefinedError	2✔
10	import arrow	2✔
11	import fnmatch	2✔
12
13	from .exceptions import FormicaArgumentException	2✔
14
15	from . import yaml_tags	2✔
16	from .helper import main_account_id	2✔
17
18	# To silence pyflakes warning of unused import
19	assert yaml_tags	2✔
20
21	logger = logging.getLogger(__name__)	2✔
22
23	FILE_TYPES = ["yml", "yaml", "json"]	2✔
24
25	RESOURCES_KEY = "Resources"	2✔
26	MODULE_KEY = "From"	2✔
27
28	ALLOWED_ATTRIBUTES = {	2✔
29	"AWSTemplateFormatVersion": [str],
30	"Description": [str],
31	"Metadata": dict,
32	"Parameters": dict,
33	"Mappings": dict,
34	"Conditions": dict,
35	"Transform": [str, list],
36	RESOURCES_KEY: dict,
37	"Outputs": dict,
38	}
39
40
41	def code_escape(source):	2✔
42	return source.replace("\n", "\\n").replace('"', '\\"')	2✔
43
44
45	def code_array(source):	2✔
46	lines = source.split("\\n")	2✔
47	return "[" + ",".join(['"%s"' % line for line in lines]) + "]"	2✔
48
49
50	def mandatory(a):	2✔
51	from jinja2.runtime import Undefined	2✔
52
53	if isinstance(a, Undefined) or a is None:	2✔
54	raise FormicaArgumentException("Mandatory variable not set.")	2✔
55	return a	2✔
56
57
58	def resource(name):	2✔
59	if name is None:	2✔
60	return ""	2✔
61	else:
62	return "".join(e for e in name.title() if e.isalnum())	2✔
63
64
65	def novalue(variable):	2✔
66	return variable or '{"Ref": "AWS::NoValue"}'	2✔
67
68
69	class Loader(object):	2✔
70	def __init__(self, path=".", filename="*", variables=None, main_account_parameter=False):	2✔
71	if variables is None:	2✔
72	variables = {}	2✔
73	self.cftemplate = {}	2✔
74	self.path = path	2✔
75	self.filename = filename	2✔
76	self.env = Environment(loader=FileSystemLoader("./", followlinks=True))	2✔
77	self.env.filters.update(	2✔
78	{
79	"code_escape": code_escape,
80	"code_array": code_array,
81	"mandatory": mandatory,
82	"resource": resource,
83	"novalue": novalue,
84	}
85	)
86	self.variables = variables	2✔
87	self.main_account_parameter = main_account_parameter	2✔
88
89	def include_file(self, filename, **args):	2✔
90	source = self.render(filename, **args)	2✔
91	return code_escape(source)	2✔
92
93	def load_file(self, filename, **args):	2✔
94	return self.render(filename, **args)	2✔
95
96	def list_files(self, filter="*"):	2✔
97	return [t for t in self.env.list_templates() if fnmatch.fnmatch(t, filter)]	2✔
98
99	def render(self, filename, **args):	2✔
100	template_path = os.path.normpath("{}/{}".format(self.path, filename))	2✔
101	template = self.env.get_template(template_path)	2✔
102	variables = {}	2✔
103	variables.update(self.variables)	2✔
104	variables.update(args)	2✔
105	arguments = dict(	2✔
106	code=self.include_file,
107	file=self.load_file,
108	files=self.list_files,
109	now=arrow.now,
110	utcnow=arrow.utcnow,
111	**variables,
112	)
113	return template.render(**arguments)	2✔
114
115	def template(self, indent=4, sort_keys=True, separators=(",", ":"), dumper=None):	2✔
116	if dumper is not None:	2✔
117	return dumper(self.cftemplate)	2✔
118	return json.dumps(self.cftemplate, indent=indent, sort_keys=sort_keys, separators=separators)	2✔
119
120	def template_dictionary(self):	2✔
121	return self.cftemplate	2✔
122
123	def merge(self, template, file):	2✔
124	if template:	2✔
125	for key in template.keys():	2✔
126	new = template[key]	2✔
127	new_type = type(new)	2✔
128	types = ALLOWED_ATTRIBUTES.get(key)	2✔
129	if type(types) != list:	2✔
130	types = [types]	2✔
131	if key in ALLOWED_ATTRIBUTES.keys() and new_type in types:	2✔
132	if new_type == str or new_type == list:	2✔
133	self.cftemplate[key] = new	2✔
134	elif new_type == dict:	2!
135	for element_key, element_value in template[key].items():	2✔
136	if (	2✔
137	key == RESOURCES_KEY
138	and isinstance(element_value, dict)
139	and MODULE_KEY in element_value
140	):
141	self.load_module(element_value[MODULE_KEY], element_key, element_value)	2✔
142	else:
143	self.cftemplate.setdefault(key, {})[element_key] = element_value	2✔
144	else:
145	logger.info("Key '{}' in file {} is not valid".format(key, file))	2✔
146	sys.exit(1)	2✔
147	else:
148	logger.info("File {} is empty".format(file))	2✔
149
150	def load_module(self, module_path, element_key, element_value):	2✔
151	path_elements = module_path.split("::")	2✔
152	dir_pattern = re.compile(fnmatch.translate(path_elements.pop(0)), re.IGNORECASE)	2✔
153	matched_dirs = [dir for dir in os.listdir(self.path) if dir_pattern.match(dir)]	2✔
154	matched_dir = module_path	2✔
155	if matched_dirs:	2✔
156	matched_dir = matched_dirs[0]	2✔
157
158	module_path = self.path + "/" + "/".join([matched_dir] + path_elements)	2✔
159
160	file_name = "*"	2✔
161
162	if not os.path.isdir(module_path):	2✔
163	file_name = module_path.split("/")[-1]	2✔
164	module_path = "/".join(module_path.split("/")[:-1])	2✔
165
166	properties = element_value.get("Properties", {})	2✔
167	properties["module_name"] = element_key	2✔
168	vars = self.merge_variables(properties)	2✔
169
170	loader = Loader(module_path, file_name, vars)	2✔
171	loader.load()	2✔
172	self.merge(loader.template_dictionary(), file=file_name)	2✔
173
174	def merge_variables(self, module_vars):	2✔
175	merged_vars = {}	2✔
176	for k, v in self.variables.items():	2✔
177	merged_vars[k] = v	2✔
178	for k, v in module_vars.items():	2✔
179	merged_vars[k] = v	2✔
180	return merged_vars	2✔
181
182	def load(self):	2✔
183	files = []	2✔
184
185	for file_type in FILE_TYPES:	2✔
186	pattern = re.compile(fnmatch.translate("{}.template.{}".format(self.filename, file_type)), re.IGNORECASE)	2✔
187	files.extend([filename for filename in os.listdir(self.path) if pattern.match(filename)])	2✔
188
189	if not files:	2✔
190	logger.info("Could not find any template files in {}".format(self.path))	2✔
191	sys.exit(1)	2✔
192
193	for file in files:	2✔
194	try:	2✔
195	result = str(self.render(os.path.basename(file), **self.variables))	2✔
196	template = yaml.full_load(result)	2✔
197	except TemplateNotFound as e:	2✔
198	logger.info("File not found" + ": " + e.message)	2✔
199	logger.info('In: "' + file + '"')	2✔
200	sys.exit(1)	2✔
201	except TemplateSyntaxError as e:	2✔
202	logger.info(e.__class__.__name__ + ": " + e.message)	2✔
203	logger.info('File: "' + (e.filename or file) + '", line ' + str(e.lineno))	2✔
204	sys.exit(1)	2✔
205	except UndefinedError as e:	2!
NEW 206	logger.info(e.__class__.__name__ + ": " + e.message)	×
NEW 207	logger.info('In: "' + file + '"')	×
NEW 208	sys.exit(1)	×
209	except FormicaArgumentException as e:	2✔
210	logger.info(e.__class__.__name__ + ": " + e.args[0])	2✔
211	logger.info('For Template: "' + file + '"')	2✔
212	logger.info("If you use it as a template make sure you're setting all necessary vars")	2✔
213	sys.exit(1)	2✔
214	except yaml.YAMLError as e:	2✔
215	logger.info(e.__str__())	2✔
216	logger.info("Following is the Yaml document formica is trying to load:")	2✔
217	logger.info("---------------------------------------------------------------------------")	2✔
218	logger.info(result)	2✔
219	logger.info("---------------------------------------------------------------------------")	2✔
220	sys.exit(1)	2✔
221	self.merge(template, file)	2✔
222
223	if self.main_account_parameter:	2✔
224	self.cftemplate["Parameters"] = self.cftemplate.get("Parameters") or {}	2✔
225	self.cftemplate["Parameters"]["MainAccount"] = {"Type": "String", "Default": main_account_id()}	2✔

theserverlessway / formica / 234004684

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