Verification of parameters and their processing is transferred to the template engine. Not tested yet.

packages
Иванов Денис 4 years ago
parent 9c339bb03b
commit 037b8238ac

@ -1,14 +1,381 @@
# vim: fileencoding=utf-8 # vim: fileencoding=utf-8
# #
from jinja2.ext import Extension from jinja2.ext import Extension
from jinja2 import Environment, FileSystemLoader, TemplateSyntaxError, nodes from jinja2 import Environment, FileSystemLoader, TemplateSyntaxError, nodes,\
contextfunction
from jinja2.utils import missing from jinja2.utils import missing
from jinja2.runtime import Context, Undefined from jinja2.runtime import Context, Undefined
from collections.abc import MutableMapping from collections.abc import MutableMapping
from collections import OrderedDict
from importlib import import_module
import re
import os
from ..utils.package import PackageAtom, PackageAtomError
from ..utils.files import join_paths
# Типы шаблона: директория или файл.
DIR, FILE = range(2)
class IncorrectParameter(Exception):
pass
class DefaultParameterError(Exception):
pass
class ConditionFailed(TemplateSyntaxError):
pass
class TemplateParametersChecker:
'''Класс для хранения, проверки и разбора параметров шаблона.'''
available_parameters = {'name', 'path', 'append', 'chmod', 'chown',
'autoupdate', 'env', 'force', 'source', 'format',
'protected', 'mirror', 'run', 'exec', 'env',
'package', 'merge', 'postmerge', 'action',
'rebuild', 'restart', 'stop', 'start'}
inheritable_parameters = {'chmod': (None, None), 'chown': (None, None),
'autoupdate': (None, None), 'env': (None, None),
'package': (None, None), 'action': (None, None)}
available_appends = set()
directory_default_values = {'chown': 'root:root',
'chmod': '755'}
file_default_values = {'chown': 'root:root',
'chmod': '644'}
available_formats = set()
format_is_inspected = False
chmod_value_regular = re.compile(
r'([r-][w-][x-])([r-][w-][x-])([r-][w-][x-])')
package_atom_parser = PackageAtom()
def __init__(self, parameters_container, template_type, parameters=dict(),
chroot_path='/'):
self.template_type = template_type
self.chroot_path = chroot_path
self._parameters_container = parameters_container
self._inspect_formats_package()
self.checkers_list = OrderedDict({
'package': self.check_package_parameter,
'append': self.check_append_parameter,
'rebuild': self.check_rebuild_parameter,
'restart': self.check_restart_parameter,
'stop': self.check_stop_parameter,
'start': self.check_start_parameter,
'chown': self.check_chown_parameter,
'chmod': self.check_chmod_parameter,
'autoupdate': self.check_autoupdate_parameter,
'source': self.check_source_parameter,
'force': self.check_force_parameter
})
try:
if template_type == DIR:
self.check_template_parameters(self.directory_default_values)
elif template_type == FILE:
self.check_template_parameters(self.file_default_values)
except IncorrectParameter as error:
raise DefaultParameterError('Default parameter value error: {}'.
format(str(error)))
if parameters:
self.check_template_parameters(parameters)
def __getattr__(self, parameter_name):
if parameter_name not in self.available_parameters:
raise IncorrectParameter("Unknown parameter: '{}'".
format(parameter_name))
elif parameter_name not in self._parameters_container:
return False
else:
return self._parameters_container[parameter_name]
def check_template_parameter(self, parameter_name, parameter_value):
# Если параметр наследуем и уже встречался до этого с тем же
# значением -- второй раз не проверяем его.
if (parameter_name in self.inheritable_parameters and
parameter_value == self.inheritable_parameters[parameter_name][0]):
return self.inheritable_parameters[parameter_name][1]
if parameter_name not in self.available_parameters:
raise IncorrectParameter("Unknown parameter '{0}'".
format(parameter_name))
elif parameter_name in self.checkers_list:
checked_value = self.checkers_list[parameter_name](
parameter_value)
if parameter_name in self.inheritable_parameters:
self.inheritable_parameters[parameter_name] =\
(parameter_value, checked_value)
return checked_value
def check_template_parameters(self, parameters):
for parameter_name in parameters:
if (parameter_name in self.inheritable_parameters and
parameters[parameter_name] ==
self.inheritable_parameters[parameter_name][0]):
self._parameters_container[parameter_name] =\
self.inheritable_parameters[parameter_name][1]
continue
if parameter_name not in self.available_parameters:
raise IncorrectParameter("Unknown parameter '{0}'".
format(parameter_name))
elif parameter_name in self.checkers_list:
parameter_value = self.checkers_list[parameter_name](
parameters[parameter_name]
)
self._parameters_container[parameter_name] = parameter_value
if parameter_name in self.inheritable_parameters:
self.inheritable_parameters[parameter_name] =\
(parameter_value, parameters[parameter_name])
def check_package_parameter(self, parameter_value):
try:
self.package_atom_parser.parse_package_parameter(parameter_value)
except PackageAtomError as error:
raise IncorrectParameter(str(error))
parameter_value = self.package_atom_parser.atom_dictionary
return parameter_value
def check_append_parameter(self, parameter_value):
if parameter_value not in self.available_appends:
raise IncorrectParameter("Unacceptable value '{}' of parameter"
" 'append'".format(parameter_value))
return parameter_value
def check_rebuild_parameter(self, parameter_value):
if isinstance(parameter_value, bool):
raise IncorrectParameter("'rebuild' parameter value is not bool")
elif 'package' not in self._parameters_container:
raise IncorrectParameter(("'source' parameter is set without "
"'package' parameter"))
return parameter_value
def check_restart_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter(
"'restart' parameter value is not correct")
def check_stop_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'stop' parameter value is not correct")
def check_start_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'start' parameter value is not correct")
def check_run_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'run' parameter value is nkt correct")
def check_exec_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'exec' parameter value is not correct")
def check_chown_parameter(self, parameter_value):
if not parameter_value or isinstance(parameter_value, bool):
raise IncorrectParameter("'chown' parameter value is empty.")
parameter_value = self.get_chown_values(parameter_value)
return parameter_value
def check_chmod_parameter(self, parameter_value):
result = self.chmod_value_regular.search(parameter_value)
if result:
parameter_value = ''
for group_number in range(3):
current_group = result.groups()[group_number]
num = ''
for sym_number in range(3):
if current_group[sym_number] != '-':
num = num + '1'
else:
num = num + '0'
parameter_value = parameter_value + num
return int(parameter_value, 2)
elif parameter_value.isdigit():
parameter_value = int(parameter_value, 8)
return parameter_value
else:
raise IncorrectParameter("'chmod' parameter value is not correct")
def check_source_parameter(self, parameter_value):
if self.chroot_path != '/':
real_path = join_paths(self.chroot_path, parameter_value)
else:
real_path = parameter_value
if not parameter_value or isinstance(parameter_value, bool):
raise IncorrectParameter("'source' parameter value is empty")
elif (self.template_type == DIR and
('append' not in self._parameters_container or
self._parameters_container['append'] != 'link')):
raise IncorrectParameter(
("'source' parameter is set without "
"'append = link' for directory template")
)
elif not os.path.exists(real_path):
raise IncorrectParameter(
"File from 'source' parameter does not exist")
return os.path.normpath(real_path)
def check_force_parameter(self, parameter_value):
if isinstance(parameter_value, bool):
return parameter_value
else:
raise IncorrectParameter("'force' parameter value is not bool")
def check_autoupdate_parameter(self, parameter_value):
print('autoupdate value = {}'.format(parameter_value))
if isinstance(parameter_value, bool):
return parameter_value
else:
raise IncorrectParameter(
"'autoupdate' parameter value is not bool")
def get_chown_values(self, chown: str):
"""Получить значения uid и gid из параметра chown."""
if chown and ':' in chown:
user_name, group_name = chown.split(':')
if user_name.isdigit():
uid = int(user_name)
else:
import pwd
try:
if self.chroot_path == '/':
uid = pwd.getpwnam(user_name).pw_uid
else:
uid = self.get_uid_from_passwd(user_name)
except (KeyError, TypeError):
self.output.set_error(
format(user_name))
raise IncorrectParameter(
("'chown' value '{0}' is not correct:"
"no such user in the system: {1}").
format(chown, user_name))
if group_name.isdigit():
gid = int(group_name)
else:
import grp
try:
if self.chroot_path == '/':
gid = grp.getgrnam(group_name).gr_gid
else:
gid = self.get_gid_from_group(group_name)
except (KeyError, TypeError):
raise IncorrectParameter(
("'chown' value '{0}' is not correct:"
"no such group in the system: {1}").
format(chown, group_name))
return {'uid': uid, 'gid': gid}
else:
raise IncorrectParameter("'chown' value '{0}' is not correct".
format(chown, self.template_path))
def get_uid_from_passwd(self, user_name: str):
"""Взять uid из chroot passwd файла."""
passwd_file_path = os.path.join(self.chroot_path, 'etc/passwd')
passwd_dictionary = []
if os.path.exists(passwd_file_path):
with open(passwd_file_path, 'r') as passwd_file:
for line in passwd_file:
if line.startswith('#'):
continue
passwd_item = tuple(line.split(':')[0:3:2])
if (len(passwd_item) > 1 and passwd_item[0]
and passwd_item[0]):
passwd_dictionary.append(passwd_item)
passwd_dictionary = dict(passwd_dictionary)
return int(passwd_dictionary[user_name])
else:
IncorrectParameter("passwd file was not found in {}".
format(passwd_file_path))
def get_gid_from_group(self, group_name: str):
"""Взять gid из chroot group файла."""
group_file_path = os.path.join(self.chroot_path, 'etc/group')
group_dictionary = []
if os.path.exists(group_file_path):
with open(group_file_path, 'r') as group_file:
for line in group_file:
if line.startswith('#'):
continue
group_item = tuple(line.split(':')[0:3:2])
if len(group_item) > 1 and group_item[0] and group_item[1]:
group_dictionary.append(group_item)
group_dictionary = dict(group_dictionary)
if group_name in group_dictionary:
return int(group_dictionary[group_name])
else:
IncorrectParameter("'{0}' gid was not found in {1}".
format(group_name, group_file_path))
else:
IncorrectParameter("group file was not found in {}".
format(group_file_path))
@classmethod
def _inspect_formats_package(cls):
'''Метод для определения множества доступных форматов и
предоставляемых ими параметров.'''
if cls.format_is_inspected:
return
parameters_set = set()
format_set = set()
format_directory_path = os.path.join(os.path.dirname(__file__),
'format')
for module_name in os.listdir(format_directory_path):
if (os.path.isdir(os.path.join('format', module_name)) or
module_name == '__init__.py'):
continue
if module_name.endswith('.py'):
module_name = module_name[:-3]
try:
module = import_module('calculate.templates.format.{}'.
format(module_name))
for obj in dir(module):
if obj.endswith('Format') and obj != 'BaseFormat':
format_class = getattr(module, obj, False)
if format_class:
format_set.add(format_class.FORMAT)
parameters = getattr(format_class,
'FORMAT_PARAMETERS', set())
parameters_set.update(parameters)
except Exception:
continue
cls.available_formats = format_set
cls.available_parameters.update(parameters_set)
cls.formats_inspected = True
class Variables(MutableMapping): class Variables(MutableMapping):
'''Класс заглушка вместо модуля переменных.''' '''Класс заглушка вместо модуля переменных для тестов.'''
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
self.__attrs = dict(*args, **kwargs) self.__attrs = dict(*args, **kwargs)
@ -83,20 +450,27 @@ class CalculateContext(Context):
env=self._env_set) env=self._env_set)
class ConditionFailed(TemplateSyntaxError):
pass
class ParametersContainer(MutableMapping): class ParametersContainer(MutableMapping):
'''Класс для хранения параметров, взятых из шаблона, и передачи '''Класс для хранения параметров, взятых из шаблона, и передачи
их шаблонизатору.''' их шаблонизатору.'''
def __init__(self, parameters_dictionary={}): def __init__(self, parameters_dictionary={}):
self.__parameters = parameters_dictionary self.__parameters = parameters_dictionary
self._new_template = True
def set_parameters(self, *args, **kwargs): def set_parameters(self, *args, **kwargs):
parameters = dict(*args, **kwargs) parameters = dict(*args, **kwargs)
self.__parameters.update(parameters) self.__parameters.update(parameters)
def __getattr__(self, parameter_name):
if (parameter_name not in
TemplateParametersChecker.available_parameters):
raise IncorrectParameter("Unknown parameter: '{}'".
format(parameter_name))
elif parameter_name not in self.__parameters:
return False
else:
return self.__parameters[parameter_name]
def __getitem__(self, name): def __getitem__(self, name):
return self.__parameters[name] return self.__parameters[name]
@ -126,6 +500,7 @@ class CalculateExtension(Extension):
_datavars = Variables() _datavars = Variables()
def __init__(self, environment): def __init__(self, environment):
print('EXTENSION IS INITIALIZED')
self.tags = {'calculate', 'save', 'set_var'} self.tags = {'calculate', 'save', 'set_var'}
self.CONDITION_TOKENS_TYPES = {'eq', 'ne', 'lt', 'gt', 'lteq', 'gteq'} self.CONDITION_TOKENS_TYPES = {'eq', 'ne', 'lt', 'gt', 'lteq', 'gteq'}
self.LITERAL_TOKENS_TYPES = {'string', 'integer', 'float'} self.LITERAL_TOKENS_TYPES = {'string', 'integer', 'float'}
@ -134,9 +509,12 @@ class CalculateExtension(Extension):
self.parse_methods = {'calculate': self.parse_calculate, self.parse_methods = {'calculate': self.parse_calculate,
'save': self.parse_save} 'save': self.parse_save}
self.parameters_checker = None
self.new_parameters_set
self.environment = environment self.environment = environment
def parse(self, parser): def parse(self, parser):
self.parameters_checker = None
self.parser = parser self.parser = parser
self.stream = parser.stream self.stream = parser.stream
tag_token = self.stream.current.value tag_token = self.stream.current.value
@ -196,7 +574,14 @@ class CalculateExtension(Extension):
and self.stream.look().type not in and self.stream.look().type not in
self.CONDITION_TOKENS_TYPES): self.CONDITION_TOKENS_TYPES):
# разбираем параметр. # разбираем параметр.
pairs_list.append(self.get_parameter_node()) # pairs_list.append(self.get_parameter_node())
name_node, value_node = self.get_parameter_node()
check_node = self.call_method('check_parameter',
[name_node,
value_node,
nodes.ContextReference()],
lineno=lineno)
pairs_list.append(check_node)
elif (self.stream.current.type == 'name' elif (self.stream.current.type == 'name'
or self.stream.current.type == 'lparen'): or self.stream.current.type == 'lparen'):
# разбираем условие. Если условие False -- кидаем исключение. # разбираем условие. Если условие False -- кидаем исключение.
@ -210,12 +595,30 @@ class CalculateExtension(Extension):
raise TemplateSyntaxError('Name is expected in calculate tag.', raise TemplateSyntaxError('Name is expected in calculate tag.',
lineno=self.stream.current.lineno) lineno=self.stream.current.lineno)
expect_comma_flag = True expect_comma_flag = True
dictionary_node = nodes.Dict(pairs_list) # dictionary_node = nodes.Dict(pairs_list)
save_node = self.call_method('save_parameters', # save_node = self.call_method('save_parameters',
[dictionary_node, # [dictionary_node,
nodes.ContextReference()], # nodes.ContextReference()],
lineno=lineno) # lineno=lineno)
return nodes.Output([save_node], lineno=lineno) # return nodes.Output([save_node], lineno=lineno)
return nodes.Output(pairs_list, lineno=lineno)
def check_parameter(self, parameter_name, parameter_value, context):
parameters_object = context.parent['__parameters__']
if not self.parameters_checker:
template_type = context.parent['__template_type__']
chroot_path = context.parent['__datavars__'].main.cl_chroot_path
self.parameters_checker = TemplateParametersChecker(
template_type,
chroot_path=chroot_path
)
parameters_object._new_template = True
elif not parameters_object._new_template:
print('pair = {0}: {1}'.format(parameter_name, parameter_value))
return ''
def get_condition_result(self): def get_condition_result(self):
'''Метод для разбора условий из тега calculate.''' '''Метод для разбора условий из тега calculate.'''
@ -287,7 +690,8 @@ class CalculateExtension(Extension):
CalculateContext._env_set.add(name.strip()) CalculateContext._env_set.add(name.strip())
else: else:
parameter_rvalue = nodes.Const(True, lineno=lineno) parameter_rvalue = nodes.Const(True, lineno=lineno)
return nodes.Pair(parameter_name_node, parameter_rvalue) # return nodes.Pair(parameter_name_node, parameter_rvalue)
return (parameter_name_node, parameter_rvalue)
def save_parameters(cls, parameters_dictionary, context): def save_parameters(cls, parameters_dictionary, context):
'''Метод для сохранения значений параметров.''' '''Метод для сохранения значений параметров.'''
@ -295,14 +699,28 @@ class CalculateExtension(Extension):
return '' return ''
@contextfunction
def pkg(context, * args):
if args:
package_atom = args[0]
else:
package_atom = context.parent['__parameters__']['package']
return package_atom
class TemplateEngine: class TemplateEngine:
def __init__(self, directory_path='/', def __init__(self, directory_path='/',
parameters_set=set(), datavars_module=Variables(),
env_set=set(), appends_set=set()):
datavars_module=Variables()): TemplateParametersChecker._inspect_formats_package()
CalculateExtension._parameters_set = parameters_set
CalculateExtension._parameters_set =\
TemplateParametersChecker.available_parameters
CalculateExtension._datavars = datavars_module CalculateExtension._datavars = datavars_module
self.available_formats = TemplateParametersChecker.available_formats
self.available_appends = set()
self._datavars_module = datavars_module self._datavars_module = datavars_module
self._parameters_object = ParametersContainer() self._parameters_object = ParametersContainer()
self._template_text = '' self._template_text = ''
@ -326,14 +744,16 @@ class TemplateEngine:
__parameters__=self._parameters_object __parameters__=self._parameters_object
) )
def process_template_from_string(self, string, env=set()): def process_template_from_string(self, string, template_type, env=set()):
'''Метод для обработки текста шаблона.''' '''Метод для обработки текста шаблона.'''
CalculateContext._env_set = env CalculateContext._env_set = env
template = self.environment.from_string(string) template = self.environment.from_string(string)
self._parameters_object = ParametersContainer(parameters_dictionary={}) self._parameters_object = ParametersContainer(parameters_dictionary={})
self._template_text = template.render( self._template_text = template.render(
__datavars__=self._datavars_module, __datavars__=self._datavars_module,
__parameters__=self._parameters_object __parameters__=self._parameters_object,
__template_type__=template_type,
__DIR__=DIR, __FILE__=FILE
) )
@property @property

@ -5,20 +5,12 @@ import re
import stat import stat
import shutil import shutil
from ..utils.files import join_paths from ..utils.files import join_paths
from importlib import import_module from .template_engine import TemplateEngine, Variables, ConditionFailed,\
from .template_engine import TemplateEngine, Variables, ConditionFailed TemplateParametersChecker, DIR, FILE,\
IncorrectParameter
from ..utils.io_module import IOModule from ..utils.io_module import IOModule
from collections import OrderedDict from collections import OrderedDict
from ..utils.mount import Mounts from ..utils.mount import Mounts
from ..utils.package import PackageAtom, PackageAtomError
class IncorrectParameter(Exception):
pass
class DefaultParameterError(Exception):
pass
class TemplateActionError(Exception): class TemplateActionError(Exception):
@ -29,331 +21,8 @@ class TemplateError(Exception):
pass pass
# Типы шаблона: директория или файл. class SystemAction:
DIR, FILE = range(2) pass
class TemplateParameters:
'''Класс для хранения, проверки и разбора параметров шаблона.'''
available_parameters = {'name', 'path', 'append', 'chmod', 'chown',
'autoupdate', 'env', 'force', 'source', 'format',
'protected', 'mirror', 'run', 'exec', 'env',
'package', 'merge', 'postmerge', 'action',
'rebuild', 'restart', 'stop', 'start'}
inheritable_parameters = {'chmod': '', 'chown': '', 'autoupdate': False,
'env': '', 'package': '', 'action': ''}
available_appends = set()
directory_default_values = {'chown': 'root:root',
'chmod': '755'}
file_default_values = {'chown': 'root:root',
'chmod': '644'}
available_formats = set()
format_is_inspected = False
chmod_value_regular = re.compile(
r'([r-][w-][x-])([r-][w-][x-])([r-][w-][x-])')
package_atom_parser = PackageAtom()
def __init__(self, parameters: dict, template_type, chroot_path='/'):
self.template_type = template_type
self.chroot_path = chroot_path
self._parameters_dictionary = {}
self._inspect_formats_package()
self.checkers_list = OrderedDict({
'package': self.check_package_parameter,
'append': self.check_append_parameter,
'rebuild': self.check_rebuild_parameter,
'restart': self.check_restart_parameter,
'stop': self.check_stop_parameter,
'start': self.check_start_parameter,
'chown': self.check_chown_parameter,
'chmod': self.check_chmod_parameter,
'autoupdate': self.check_autoupdate_parameter,
'source': self.check_source_parameter,
'force': self.check_force_parameter,
})
try:
if template_type == DIR:
self.check_template_parameters(self.directory_default_values)
elif template_type == FILE:
self.check_template_parameters(self.file_default_values)
except IncorrectParameter as error:
raise TemplateError('Default values error: {}'.format(str(error)))
self.check_template_parameters(parameters)
def __getattr__(self, parameter_name):
if parameter_name not in self.available_parameters:
raise IncorrectParameter("Unknown parameter: '{}'".
format(parameter_name))
elif parameter_name not in self._parameters_dictionary:
return False
else:
return self._parameters_dictionary[parameter_name]
def check_template_parameters(self, parameters):
for parameter_name in parameters:
# Если параметр наследуем и уже встречался до этого --
# второй раз не проверяем его.
if (parameter_name in self.inheritable_parameters and
not isinstance(self.inheritable_parameters[parameter_name],
bool) and
parameters[parameter_name] ==
self.inheritable_parameters[parameter_name]):
continue
if parameter_name not in self.available_parameters:
raise IncorrectParameter("Unknown parameter '{0}'".
format(parameter_name))
elif parameter_name in self.checkers_list:
parameter_value = self.checkers_list[parameter_name](
parameters[parameter_name]
)
self._parameters_dictionary[parameter_name] = parameter_value
if parameter_name in self.inheritable_parameters:
self.inheritable_parameters[parameter_name] =\
parameter_value
def check_package_parameter(self, parameter_value):
try:
self.package_atom_parser.parse_package_parameter(parameter_value)
except PackageAtomError as error:
raise IncorrectParameter(str(error))
parameter_value = self.package_atom_parser.atom_dictionary
return parameter_value
def check_append_parameter(self, parameter_value):
if parameter_value not in self.available_appends:
raise IncorrectParameter("Unacceptable value '{}' of parameter"
" 'append'".format(parameter_value))
return parameter_value
def check_rebuild_parameter(self, parameter_value):
if isinstance(parameter_value, bool):
raise IncorrectParameter("'rebuild' parameter value is not bool")
elif 'package' not in self._parameters_dictionary:
raise IncorrectParameter(("'source' parameter is set without "
"'package' parameter"))
return parameter_value
def check_restart_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter(
"'restart' parameter value is not correct")
def check_stop_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'stop' parameter value is not correct")
def check_start_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'start' parameter value is not correct")
def check_run_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'run' parameter value is nkt correct")
def check_exec_parameter(self, parameter_value):
if parameter_value and isinstance(parameter_value, str):
return parameter_value
else:
raise IncorrectParameter("'exec' parameter value is not correct")
def check_chown_parameter(self, parameter_value):
if not parameter_value or isinstance(parameter_value, bool):
raise IncorrectParameter("'chown' parameter value is empty.")
parameter_value = self.get_chown_values(parameter_value)
return parameter_value
def check_chmod_parameter(self, parameter_value):
result = self.chmod_value_regular.search(parameter_value)
if result:
parameter_value = ''
for group_number in range(3):
current_group = result.groups()[group_number]
num = ''
for sym_number in range(3):
if current_group[sym_number] != '-':
num = num + '1'
else:
num = num + '0'
parameter_value = parameter_value + num
return int(parameter_value, 2)
elif parameter_value.isdigit():
parameter_value = int(parameter_value, 8)
return parameter_value
else:
raise IncorrectParameter("'chmod' parameter value is not correct")
def check_source_parameter(self, parameter_value):
if self.chroot_path != '/':
real_path = join_paths(self.chroot_path, parameter_value)
else:
real_path = parameter_value
if not parameter_value or isinstance(parameter_value, bool):
raise IncorrectParameter("'source' parameter value is empty")
elif (self.template_type == DIR and
('append' not in self._parameters_dictionary or
self._parameters_dictionary['append'] != 'link')):
raise IncorrectParameter(
("'source' parameter is set without "
"'append = link' for directory template")
)
elif not os.path.exists(real_path):
raise IncorrectParameter(
"File from 'source' parameter does not exist")
return os.path.normpath(real_path)
def check_force_parameter(self, parameter_value):
if isinstance(parameter_value, bool):
return parameter_value
else:
raise IncorrectParameter("'force' parameter value is not bool")
def check_autoupdate_parameter(self, parameter_value):
print('autoupdate value = {}'.format(parameter_value))
if isinstance(parameter_value, bool):
return parameter_value
else:
raise IncorrectParameter(
"'autoupdate' parameter value is not bool")
def get_chown_values(self, chown: str):
"""Получить значения uid и gid из параметра chown."""
if chown and ':' in chown:
user_name, group_name = chown.split(':')
if user_name.isdigit():
uid = int(user_name)
else:
import pwd
try:
if self.chroot_path == '/':
uid = pwd.getpwnam(user_name).pw_uid
else:
uid = self.get_uid_from_passwd(user_name)
except (KeyError, TypeError):
self.output.set_error(
format(user_name))
raise IncorrectParameter(
("'chown' value '{0}' is not correct:"
"no such user in the system: {1}").
format(chown, user_name))
if group_name.isdigit():
gid = int(group_name)
else:
import grp
try:
if self.chroot_path == '/':
gid = grp.getgrnam(group_name).gr_gid
else:
gid = self.get_gid_from_group(group_name)
except (KeyError, TypeError):
raise IncorrectParameter(
("'chown' value '{0}' is not correct:"
"no such group in the system: {1}").
format(chown, group_name))
return {'uid': uid, 'gid': gid}
else:
raise IncorrectParameter("'chown' value '{0}' is not correct".
format(chown, self.template_path))
def get_uid_from_passwd(self, user_name: str):
"""Взять uid из chroot passwd файла."""
passwd_file_path = os.path.join(self.chroot_path, 'etc/passwd')
passwd_dictionary = []
if os.path.exists(passwd_file_path):
with open(passwd_file_path, 'r') as passwd_file:
for line in passwd_file:
if line.startswith('#'):
continue
passwd_item = tuple(line.split(':')[0:3:2])
if (len(passwd_item) > 1 and passwd_item[0]
and passwd_item[0]):
passwd_dictionary.append(passwd_item)
passwd_dictionary = dict(passwd_dictionary)
return int(passwd_dictionary[user_name])
else:
IncorrectParameter("passwd file was not found in {}".
format(passwd_file_path))
def get_gid_from_group(self, group_name: str):
"""Взять gid из chroot group файла."""
group_file_path = os.path.join(self.chroot_path, 'etc/group')
group_dictionary = []
if os.path.exists(group_file_path):
with open(group_file_path, 'r') as group_file:
for line in group_file:
if line.startswith('#'):
continue
group_item = tuple(line.split(':')[0:3:2])
if len(group_item) > 1 and group_item[0] and group_item[1]:
group_dictionary.append(group_item)
group_dictionary = dict(group_dictionary)
if group_name in group_dictionary:
return int(group_dictionary[group_name])
else:
IncorrectParameter("'{0}' gid was not found in {1}".
format(group_name, group_file_path))
else:
IncorrectParameter("group file was not found in {}".
format(group_file_path))
@classmethod
def _inspect_formats_package(cls):
'''Метод для определения множества доступных форматов и
предоставляемых ими параметров.'''
if cls.format_is_inspected:
return
parameters_set = set()
format_set = set()
format_directory_path = os.path.join(os.path.dirname(__file__),
'format')
for module_name in os.listdir(format_directory_path):
if (os.path.isdir(os.path.join('format', module_name)) or
module_name == '__init__.py'):
continue
if module_name.endswith('.py'):
module_name = module_name[:-3]
try:
module = import_module('calculate.templates.format.{}'.
format(module_name))
for obj in dir(module):
if obj.endswith('Format') and obj != 'BaseFormat':
format_class = getattr(module, obj, False)
if format_class:
format_set.add(format_class.FORMAT)
parameters = getattr(format_class,
'FORMAT_PARAMETERS', set())
parameters_set.update(parameters)
except Exception:
continue
cls.available_formats = format_set
cls.available_parameters.update(parameters_set)
cls.formats_inspected = True
class TemplateAction: class TemplateAction:
@ -384,14 +53,12 @@ class TemplateAction:
'skip': self._append_skip_file 'skip': self._append_skip_file
} }
@property
def available_appends(self):
appends_set = set(self.DIRECTORY_APPENDS.keys()).union( appends_set = set(self.DIRECTORY_APPENDS.keys()).union(
set(self.FILE_APPENDS.keys())) set(self.FILE_APPENDS.keys()))
TemplateParameters.available_appends = appends_set return appends_set
TemplateParameters._inspect_formats_package()
self.available_parameters = TemplateParameters.available_parameters
self.available_formats = TemplateParameters.available_formats
def process_template_directory(self, target_path, template_path, def process_template_directory(self, target_path, template_path,
parameters={}): parameters={}):
@ -400,7 +67,7 @@ class TemplateAction:
self.target_path = target_path self.target_path = target_path
try: try:
self.template_parameters = TemplateParameters( self.template_parameters = TemplateParametersChecker(
parameters, parameters,
template_type=DIR, template_type=DIR,
chroot_path=self.chroot_path) chroot_path=self.chroot_path)
@ -574,7 +241,7 @@ class TemplateAction:
self.target_path = target_path self.target_path = target_path
try: try:
self.template_parameters = TemplateParameters( self.template_parameters = TemplateParametersChecker(
parameters, parameters,
template_type=FILE, template_type=FILE,
chroot_path=self.chroot_path) chroot_path=self.chroot_path)
@ -784,15 +451,10 @@ class DirectoryProcessor:
self.template_action = TemplateAction(output_module=self.output, self.template_action = TemplateAction(output_module=self.output,
chroot_path=self.chroot_path) chroot_path=self.chroot_path)
self.parameters_set = self.template_action.available_parameters
self.formats_set = self.template_action.available_formats
self.inheritable_parameters =\ self.inheritable_parameters =\
set(TemplateParameters.inheritable_parameters) set(TemplateParametersChecker.inheritable_parameters)
self.template_engine = TemplateEngine( self.template_engine = TemplateEngine(datavars_module=datavars_module)
datavars_module=datavars_module,
parameters_set=self.parameters_set
)
self.template_paths = (self.datavars_module. self.template_paths = (self.datavars_module.
main.cl_template_path.split(',')) main.cl_template_path.split(','))
@ -1076,15 +738,6 @@ class DirectoryProcessor:
target_file_path = join_paths(current_target_path, target_file_path = join_paths(current_target_path,
template_name) template_name)
if ('format' in template_parameters and
template_parameters['format'] not in self.formats_set):
self.output.set_error(
"Unknown parameter '{0}' in template: {1}".
format(template_parameters['format'],
target_file_path)
)
continue
# Выполняем действие с использованием шаблона. # Выполняем действие с использованием шаблона.
self.output.set_success('Processing template: {}...'. self.output.set_success('Processing template: {}...'.
format(template_name)) format(template_name))

@ -0,0 +1,22 @@
from calculate.templates.template_engine import TemplateEngine, DIR, Variables
from calculate.templates.template_processor import TemplateAction
main = Variables({'cl_template_path':
('tests/templates/testfiles/template_dir_1,'
'tests/templates/testfiles/template_dir_2'),
'cl_chroot_path': '/'})
datavars = Variables({'main': main})
TemplateAction()
template_engine = TemplateEngine(datavars_module=datavars)
template_engine.process_template_from_string(
"{% calculate package = 'dev-lang/python-3.6', action = 'install' %}",
DIR)
template_engine.process_template_from_string(
"{% calculate package = 'dev-lang/python-2.7', action = 'update' %}",
DIR)
Loading…
Cancel
Save