calculate
/
calculate-utils-4-lib
6
0
Fork 0
Code Issues 1 Pull Requests Releases Wiki Activity

Now template can be run without install package value.

Browse Source
master
Иванов Денис 3 years ago
parent 74e9ef8b9f
commit c480831c59
23 changed files with 450 additions and 260 deletions
Show Stats Download Patch File Download Diff File

12
calculate/scripts/scripts.py
Unescape View File

@ -46,18 +46,18 @@ class Script:
в себе. Создает экземпляры лаунчеров.''' в себе. Создает экземпляры лаунчеров.'''
def __init__(self, script_id: str, def __init__(self, script_id: str,
args: List[Any] = [], args: List[Any] = [],
success_message: str = None, success_message: Union[str, None] = None,
failed_message: str = None, failed_message: Union[str, None] = None,
interrupt_message: str = None): interrupt_message: Union[str, None] = None):
self._id: str = script_id self._id: str = script_id
self.__items: List[Union['Task', 'Block', 'Handler', 'Run']] = None self.__items: List[Union['Task', 'Block', 'Handler', 'Run']] = None
self.__tasks_is_set: bool = False self.__tasks_is_set: bool = False
self.__args: list = args self.__args: list = args
self.__success_message: str = success_message self.__success_message: Union[str, None] = success_message
self.__failed_message: str = failed_message self.__failed_message: Union[str, None] = failed_message
self.__interrupt_message: str = interrupt_message self.__interrupt_message: Union[str, None] = interrupt_message
@property @property
def id(self) -> str: def id(self) -> str:

101
calculate/templates/template_engine.py
Unescape View File

@ -18,10 +18,13 @@ import copy
import re import re
import os import os
import stat import stat
from typing import Union, Any, List
from ..utils.package import ( from ..utils.package import (
PackageAtomName,
PackageAtomParser, PackageAtomParser,
PackageAtomError, PackageAtomError,
Package,
NOTEXIST, NOTEXIST,
Version Version
) )
@ -123,48 +126,51 @@ class Variables(MutableMapping):
class ParametersProcessor: class ParametersProcessor:
'''Класс для проверки и разбора параметров шаблона.''' '''Класс для проверки и разбора параметров шаблона.'''
available_parameters = {'name', 'path', 'append', 'chmod', 'chown', available_parameters: set = {'name', 'path', 'append', 'chmod', 'chown',
'autoupdate', 'env', 'force', 'source', 'format', 'autoupdate', 'env', 'force', 'source',
'unbound', 'mirror', 'run', 'exec', 'env', 'format', 'unbound', 'mirror', 'run', 'exec',
'package', 'merge', 'postmerge', 'action', 'env', 'package', 'merge', 'postmerge',
'rebuild', 'restart', 'stop', 'start', 'handler', 'action', 'rebuild', 'restart', 'stop',
'notify', 'group'} 'start', 'handler', 'notify', 'group'}
inheritable_parameters = {'chmod', 'chown', 'autoupdate', 'env', inheritable_parameters: set = {'chmod', 'chown', 'autoupdate', 'env',
'package', 'action', 'handler', 'group'} 'package', 'action', 'handler', 'group'}
# Параметры по умолчанию для файлов -- # Параметры по умолчанию для файлов --
# будут заполняться из __datavars__ # будут заполняться из __datavars__
file_default_parameters = {} file_default_parameters: dict = {}
# Параметры по умолчанию для директорий -- # Параметры по умолчанию для директорий --
# будут заполняться из __datavars__ # будут заполняться из __datavars__
directory_default_parameters = {} directory_default_parameters: dict = {}
available_appends = set() available_appends: set = set()
available_formats = dict() available_formats: dict = dict()
format_is_inspected = False format_is_inspected: bool = False
chmod_value_regular = re.compile( chmod_value_regular = re.compile(
r'([r-][w-][x-])([r-][w-][x-])([r-][w-][x-])') r'([r-][w-][x-])([r-][w-][x-])([r-][w-][x-])')
def __init__(self, parameters_container=None, def __init__(self,
chroot_path='/', parameters_container: Union["ParametersContainer",
datavars_module=Variables(), None] = None,
for_package=None): chroot_path: str = '/',
self.chroot_path = chroot_path datavars_module: Union[Datavars, Variables] = Variables(),
for_package: Union[Package, None] = None):
self.chroot_path: str = chroot_path
self.template_type = DIR self.template_type: int = DIR
self.datavars_module = datavars_module self.datavars_module: Union[Datavars, Variables] = datavars_module
self._parameters_container = parameters_container self._parameters_container: ParametersContainer = parameters_container
self.package_atom_parser = PackageAtomParser(chroot_path=chroot_path) self.package_atom_parser: PackageAtomParser =\
PackageAtomParser(chroot_path=chroot_path)
self._groups = {} self._groups: dict = {}
try: try:
groups = list(datavars_module.main.cl.groups._variables.keys()) groups = list(datavars_module.main.cl.groups._variables.keys())
for group in groups: for group in groups:
@ -179,7 +185,7 @@ class ParametersProcessor:
self._inspect_formats_package() self._inspect_formats_package()
self._for_package = None self._for_package: Union[Package, None] = for_package
# Если добавляемый параметр нуждается в проверке -- добавляем сюда # Если добавляемый параметр нуждается в проверке -- добавляем сюда
# метод для проверки. # метод для проверки.
@ -222,20 +228,22 @@ class ParametersProcessor:
# указываем здесь эти условия. # указываем здесь эти условия.
self.inherit_conditions = {'chmod': self.is_chmod_inheritable} self.inherit_conditions = {'chmod': self.is_chmod_inheritable}
def set_parameters_container(self, parameters_container): def set_parameters_container(self,
parameters_container: "ParametersContainer"
) -> None:
'''Метод для установки текущего контейнера параметров.''' '''Метод для установки текущего контейнера параметров.'''
self._parameters_container = parameters_container self._parameters_container = parameters_container
self._added_parameters = set() self._added_parameters = set()
@property @property
def for_package(self): def for_package(self) -> Union[Package, None]:
return self._for_package return self._for_package
@for_package.setter @for_package.setter
def for_package(self, package): def for_package(self, package: Package):
self._for_package = package self._for_package = package
def __getattr__(self, parameter_name): def __getattr__(self, parameter_name: str) -> Any:
if parameter_name not in self.available_parameters: if parameter_name not in self.available_parameters:
raise IncorrectParameter("Unknown parameter: '{}'". raise IncorrectParameter("Unknown parameter: '{}'".
format(parameter_name)) format(parameter_name))
@ -244,8 +252,9 @@ class ParametersProcessor:
else: else:
return self._parameters_container[parameter_name] return self._parameters_container[parameter_name]
def check_template_parameter(self, parameter_name, parameter_value, def check_template_parameter(self, parameter_name: str,
template_type, lineno): parameter_value: Any,
template_type: int, lineno: int) -> None:
'''Метод, проверяющий указанный параметр.''' '''Метод, проверяющий указанный параметр.'''
self.lineno = lineno self.lineno = lineno
self.template_type = template_type self.template_type = template_type
@ -280,7 +289,7 @@ class ParametersProcessor:
self._parameters_container.set_parameter( self._parameters_container.set_parameter(
{parameter_name: checked_value}) {parameter_name: checked_value})
def check_postparse_parameters(self): def check_postparse_parameters(self) -> None:
'''Метод, запускающий проверку параметров после их разбора.''' '''Метод, запускающий проверку параметров после их разбора.'''
for parameter, parameter_checker in\ for parameter, parameter_checker in\
self.postparse_checkers_list.items(): self.postparse_checkers_list.items():
@ -292,7 +301,8 @@ class ParametersProcessor:
self._parameters_container.change_parameter(parameter, self._parameters_container.change_parameter(parameter,
result) result)
def check_template_parameters(self, parameters, template_type, lineno): def check_template_parameters(self, parameters: dict,
template_type: int, lineno: int) -> None:
'''Метод, запускающий проверку указанных параметров.''' '''Метод, запускающий проверку указанных параметров.'''
self.template_type = template_type self.template_type = template_type
self.lineno = lineno self.lineno = lineno
@ -326,14 +336,14 @@ class ParametersProcessor:
# Методы для проверки параметров во время разбора шаблона. # Методы для проверки параметров во время разбора шаблона.
def check_package_parameter(self, parameter_value): def check_package_parameter(self, parameter_value: Any) -> str:
if not isinstance(parameter_value, str): if not isinstance(parameter_value, str):
raise IncorrectParameter("'package' parameter must have value of" raise IncorrectParameter("'package' parameter must have value of"
" the 'str' type, not" " the 'str' type, not"
f" {type(parameter_value)}") f" {type(parameter_value)}")
return parameter_value return parameter_value
def check_group_parameter(self, parameter_value): def check_group_parameter(self, parameter_value: Any) -> List[str]:
if isinstance(parameter_value, str): if isinstance(parameter_value, str):
result = [group.strip() for group in parameter_value.split(',')] result = [group.strip() for group in parameter_value.split(',')]
elif isinstance(parameter_value, (list, tuple)): elif isinstance(parameter_value, (list, tuple)):
@ -350,13 +360,14 @@ class ParametersProcessor:
f" {', '.join(self._groups.keys())}") f" {', '.join(self._groups.keys())}")
return result return result
def check_append_parameter(self, parameter_value): def check_append_parameter(self, parameter_value: Any) -> str:
if parameter_value not in self.available_appends: if parameter_value not in self.available_appends:
raise IncorrectParameter("Unacceptable value '{}' of parameter" raise IncorrectParameter("Unacceptable value '{}' of parameter"
" 'append'".format(parameter_value)) " 'append'".format(parameter_value))
return parameter_value return parameter_value
def check_merge_parameter(self, parameter_value): def check_merge_parameter(self, parameter_value: Any
) -> List[PackageAtomName]:
packages_list = [] packages_list = []
packages_names = parameter_value.split(',') packages_names = parameter_value.split(',')
@ -369,15 +380,15 @@ class ParametersProcessor:
return packages_list return packages_list
def check_rebuild_parameter(self, parameter_value): def check_rebuild_parameter(self, parameter_value: Any) -> bool:
if isinstance(parameter_value, bool): if not isinstance(parameter_value, bool):
raise IncorrectParameter("'rebuild' parameter value is not bool") raise IncorrectParameter("'rebuild' parameter value is not bool")
elif 'package' not in self._parameters_container: elif 'package' not in self._parameters_container:
raise IncorrectParameter(("'rebuild' parameter is set without " raise IncorrectParameter(("'rebuild' parameter is set without "
"'package' parameter")) "'package' parameter"))
return parameter_value return parameter_value
def check_restart_parameter(self, parameter_value): def check_restart_parameter(self, parameter_value: Any) -> str:
if parameter_value and isinstance(parameter_value, str): if parameter_value and isinstance(parameter_value, str):
return parameter_value return parameter_value
else: else:
@ -614,6 +625,8 @@ class ParametersProcessor:
" in handler templates.") " in handler templates.")
def check_postparse_package(self, parameter_value): def check_postparse_package(self, parameter_value):
print('check package value:', parameter_value)
print('groups:', self._groups)
groups = [] groups = []
package_atom = PackageAtomParser.parse_atom_name(parameter_value) package_atom = PackageAtomParser.parse_atom_name(parameter_value)
@ -637,18 +650,24 @@ class ParametersProcessor:
raise IncorrectParameter(error.message) raise IncorrectParameter(error.message)
elif self._check_package_group(package_atom, elif self._check_package_group(package_atom,
self._groups[group]): self._groups[group]):
self._parameters_container.remove_parameter('package') print('successfully checked')
if (self._parameters_container is not None
and self._parameters_container.package):
self._parameters_container.remove_parameter('package')
return return
raise ConditionFailed(f"package '{parameter_value}'" raise ConditionFailed(f"package '{parameter_value}'"
" does not match the template condition", " does not match the template condition",
self.lineno) self.lineno if hasattr(self, 'lineno') else 0)
def _check_package_group(self, package: dict, group_packages: list): def _check_package_group(self, package: dict, group_packages: list):
'''Метод для проверки соответствия описания пакета, заданного словарем, '''Метод для проверки соответствия описания пакета, заданного словарем,
какому-либо описанию пакета, заданного в переменных groups.''' какому-либо описанию пакета, заданного в переменных groups.'''
print(f'CHECK\n{package}')
for group_package in group_packages: for group_package in group_packages:
print(f'WITH GROUP PACKAGE {group_package}')
for parameter in ['category', 'name', 'version', 'slot']: for parameter in ['category', 'name', 'version', 'slot']:
print(f"checkin' {parameter}")
if package[parameter] is not None: if package[parameter] is not None:
if (group_package[parameter] is None if (group_package[parameter] is None
or group_package[parameter] != package[parameter]): or group_package[parameter] != package[parameter]):

410
calculate/templates/template_processor.py
Unescape View File

@ -36,6 +36,14 @@ from calculate.variables.datavars import (
TableType, TableType,
VariableNotFoundError VariableNotFoundError
) )
from typing import (
Union,
Dict,
List,
Tuple,
Iterator,
Callable
)
from calculate.variables.loader import Datavars from calculate.variables.loader import Datavars
from .format.base_format import Format from .format.base_format import Format
from ..utils.io_module import IOModule from ..utils.io_module import IOModule
@ -70,15 +78,15 @@ class TemplateCollisionError(Exception):
class CalculateConfigFile: class CalculateConfigFile:
'''Класс для работы с файлом /var/lib/calculate/config.''' '''Класс для работы с файлом /var/lib/calculate/config.'''
def __init__(self, cl_config_path='/var/lib/calculate/config', def __init__(self, cl_config_path: str = '/var/lib/calculate/config',
cl_chroot_path='/'): cl_chroot_path: str = '/'):
self.chroot_path = cl_chroot_path self.chroot_path: str = cl_chroot_path
self.cl_config_path = cl_config_path self.cl_config_path: str = cl_config_path
self._config_dictionary = self._get_cl_config_dictionary() self._config_dictionary: OrderedDict = self._get_cl_config_dictionary()
self._unsaved_changes = False self._unsaved_changes: bool = False
def __contains__(self, file_path: str) -> bool: def __contains__(self, file_path: str) -> bool:
file_path = self._remove_chroot(file_path) file_path = self._remove_chroot(file_path)
@ -158,12 +166,13 @@ class TemplateWrapper:
'''Класс связывающий шаблон с целевым файлом и определяющий параметры '''Класс связывающий шаблон с целевым файлом и определяющий параметры
наложения шаблона, обусловленные состоянием целевого файла.''' наложения шаблона, обусловленные состоянием целевого файла.'''
type_checks = {DIR: os.path.isdir, type_checks: Dict[int,
FILE: os.path.isfile} Callable[[str], bool]] = {DIR: os.path.isdir,
FILE: os.path.isfile}
_protected_is_set = False _protected_is_set: bool = False
_protected_set = set() _protected_set: set = set()
_unprotected_set = set() _unprotected_set: set = set()
def __new__(cls, *args, **kwargs): def __new__(cls, *args, **kwargs):
if not cls._protected_is_set: if not cls._protected_is_set:
@ -175,63 +184,64 @@ class TemplateWrapper:
cls._set_protected(chroot_path) cls._set_protected(chroot_path)
return super().__new__(cls) return super().__new__(cls)
def __init__(self, target_file_path, def __init__(
parameters, self, target_file_path: str,
template_type, parameters: ParametersContainer,
template_path, template_type: int,
template_text='', template_path: str,
target_package=None, template_text: str = '',
chroot_path='/', target_package: Union[Package, None] = None,
config_archive_path='/var/lib/calculate/config-archive', chroot_path: str = '/',
dbpkg=True): config_archive_path: str = '/var/lib/calculate/config-archive',
self.target_path = target_file_path dbpkg: bool = True):
self.template_path = template_path self.target_path: str = target_file_path
self.chroot_path = chroot_path self.template_path: str = template_path
self.config_archive_path = config_archive_path self.chroot_path: str = chroot_path
self.config_archive_path: str = config_archive_path
self.target_package_name = None
self.package_atom_parser = PackageAtomParser( self.target_package_name: Union[PackageAtomName, None] = None
self.package_atom_parser: PackageAtomParser = PackageAtomParser(
chroot_path=self.chroot_path) chroot_path=self.chroot_path)
# Вспомогательный флаг, включается, если по целевому пути лежит файл, # Вспомогательный флаг, включается, если по целевому пути лежит файл,
# для которого не определился никакой пакет. # для которого не определился никакой пакет.
self.target_without_package = False self.target_without_package: bool = False
self.parameters = parameters self.parameters: ParametersContainer = parameters
self.output_path = self.target_path self.output_path: str = self.target_path
self.input_path = None self.input_path: Union[str, None] = None
self.template_type = template_type self.template_type: int = template_type
self.template_text = template_text self.template_text: str = template_text
self.contents_matching = True self.contents_matching: bool = True
self.ca_is_missed = False self.ca_is_missed: bool = False
# Флаг, указывающий, что нужно удалить файл из target_path перед # Флаг, указывающий, что нужно удалить файл из target_path перед
# применением шаблона. # применением шаблона.
self.remove_original = False self.remove_original: bool = False
# Флаг, указывающий, что целевой путь был изменен. # Флаг, указывающий, что целевой путь был изменен.
self.target_path_is_changed = False self.target_path_is_changed: bool = False
# Флаг, указывающий, что файл по целевому пути является ссылкой. # Флаг, указывающий, что файл по целевому пути является ссылкой.
self.target_is_link = False self.target_is_link: bool = False
# Пакет, к которому относится файл. # Пакет, к которому относится файл.
self.target_package = target_package self.target_package: Package = target_package
# Флаг, разрешающий работу с CONTENTS. Если False, то выключает # Флаг, разрешающий работу с CONTENTS. Если False, то выключает
# protected для всех файлов блокирует все операции с CONTENTS и ._cfg. # protected для всех файлов блокирует все операции с CONTENTS и ._cfg.
self.dbpkg = dbpkg self.dbpkg: bool = dbpkg
# Флаг, указывающий, что файл является PROTECTED. # Флаг, указывающий, что файл является PROTECTED.
self.protected = False self.protected: bool = False
# Временный флаг для определения того, является ли шаблон userspace. # Временный флаг для определения того, является ли шаблон userspace.
self.is_userspace = False self.is_userspace: bool = False
self.format_class = None self.format_class: Union[Format, None] = None
if self.parameters.run or self.parameters.exec: if self.parameters.run or self.parameters.exec:
# Если есть параметр run или exec, то кроме текста шаблона ничего # Если есть параметр run или exec, то кроме текста шаблона ничего
@ -410,8 +420,8 @@ class TemplateWrapper:
self.target_package_name = parameter_package self.target_package_name = parameter_package
elif file_package != parameter_package and self.template_type != DIR: elif file_package != parameter_package and self.template_type != DIR:
target_name = self._check_packages_slots(parameter_package, target_name = self._compare_packages(parameter_package,
file_package) file_package)
if (target_name is not None and self.target_package is not None if (target_name is not None and self.target_package is not None
and self.target_package.package_name == target_name): and self.target_package.package_name == target_name):
target_name = self.target_package.package_name target_name = self.target_package.package_name
@ -453,7 +463,10 @@ class TemplateWrapper:
self.target_package = Package(self.target_package_name, self.target_package = Package(self.target_package_name,
chroot_path=self.chroot_path) chroot_path=self.chroot_path)
def _check_packages_slots(self, lpackage, rpackage): def _compare_packages(self, lpackage: PackageAtomName,
rpackage: PackageAtomName
) -> Union[None, PackageAtomName]:
'''Метод, сравнивающий пакеты по их именам, возвращает старший.'''
if lpackage.category != rpackage.category: if lpackage.category != rpackage.category:
return None return None
@ -606,7 +619,7 @@ class TemplateWrapper:
if self.template_type == DIR and self.target_package is not None: if self.template_type == DIR and self.target_package is not None:
self.target_package.clear_dir(self.target_path) self.target_package.clear_dir(self.target_path)
def add_to_contents(self, file_md5=None) -> None: def add_to_contents(self, file_md5: Union[str, None] = None) -> None:
'''Метод для добавления целевого файла в CONTENTS.''' '''Метод для добавления целевого файла в CONTENTS.'''
if self.target_package is None: if self.target_package is None:
return return
@ -684,60 +697,65 @@ class TemplateWrapper:
class TemplateExecutor: class TemplateExecutor:
'''Класс исполнительного модуля.''' '''Класс исполнительного модуля.'''
def __init__(self, datavars_module=Variables(), chroot_path='/', def __init__(self,
cl_config_archive='/var/lib/calculate/config-archive', datavars_module: Union[Datavars, Variables] = Variables(),
cl_config_path='/var/lib/calculate/config', chroot_path: str = '/',
execute_archive_path='/var/lib/calculate/.execute/', cl_config_archive: str = '/var/lib/calculate/config-archive',
dbpkg=True): cl_config_path: str = '/var/lib/calculate/config',
execute_archive_path: str = '/var/lib/calculate/.execute/',
dbpkg: bool = True):
# TODO добавить список измененных файлов. # TODO добавить список измененных файлов.
self.datavars_module = datavars_module self.datavars_module: Union[Datavars, Variables] = datavars_module
self.chroot_path = chroot_path self.chroot_path: str = chroot_path
# Объект для проверки файловых систем. Пока не инициализируем. # Объект для проверки файловых систем. Пока не инициализируем.
self.mounts = None self.mounts: Union[Mounts, None] = None
# Директория для хранения полученных при обработке exec скриптов. # Директория для хранения полученных при обработке exec скриптов.
self.execute_archive_path = execute_archive_path self.execute_archive_path: str = execute_archive_path
self.execute_files = OrderedDict() self.execute_files: OrderedDict = OrderedDict()
self.dbpkg = dbpkg self.dbpkg: bool = dbpkg
# Словарь с измененными файлами и статусами их изменений. # Словарь с измененными файлами и статусами их изменений.
self.changed_files = {} self.changed_files: dict = {}
# Список целевых путей измененных файлов. Нужен для корректного # Список целевых путей измененных файлов. Нужен для корректного
# формирования calculate-заголовка. # формирования calculate-заголовка.
self.processed_targets = [] self.processed_targets: list = []
self.directory_default_parameters =\ # TODO разобраться с этим.
# Значения параметров по умолчанию, пока не используются.
self.directory_default_parameters: dict =\
ParametersProcessor.directory_default_parameters ParametersProcessor.directory_default_parameters
self.file_default_parameters =\ self.file_default_parameters: dict =\
ParametersProcessor.file_default_parameters ParametersProcessor.file_default_parameters
# Отображение имен действий для директорий на методы их реализующие. # Отображение имен действий для директорий на методы их реализующие.
self.directory_appends = {'join': self._append_join_directory, self.directory_appends: dict = {
'remove': self._append_remove_directory, 'join': self._append_join_directory,
'skip': self._append_skip_directory, 'remove': self._append_remove_directory,
'clear': self._append_clear_directory, 'skip': self._append_skip_directory,
'link': self._append_link_directory, 'clear': self._append_clear_directory,
'replace': self._append_replace_directory} 'link': self._append_link_directory,
'replace': self._append_replace_directory}
# Отображение имен действий для файлов на методы их реализующие. # Отображение имен действий для файлов на методы их реализующие.
self.file_appends = {'join': self._append_join_file, self.file_appends: dict = {'join': self._append_join_file,
'after': self._append_after_file, 'after': self._append_after_file,
'before': self._append_before_file, 'before': self._append_before_file,
'replace': self._append_replace_file, 'replace': self._append_replace_file,
'remove': self._append_remove_file, 'remove': self._append_remove_file,
'skip': self._append_skip_file, 'skip': self._append_skip_file,
'clear': self._append_clear_file, 'clear': self._append_clear_file,
'link': self._append_link_file} 'link': self._append_link_file}
self.formats_classes = ParametersProcessor.available_formats self.formats_classes: set = ParametersProcessor.available_formats
self.calculate_config_file = CalculateConfigFile( self.calculate_config_file: CalculateConfigFile = CalculateConfigFile(
cl_config_path=cl_config_path, cl_config_path=cl_config_path,
cl_chroot_path=chroot_path) cl_chroot_path=chroot_path)
self.cl_config_archive_path = cl_config_archive self.cl_config_archive_path: str = cl_config_archive
Format.CALCULATE_VERSION = CALCULATE_VERSION Format.CALCULATE_VERSION = CALCULATE_VERSION
@property @property
@ -750,8 +768,9 @@ class TemplateExecutor:
def execute_template(self, target_path: str, def execute_template(self, target_path: str,
parameters: ParametersContainer, template_type: int, parameters: ParametersContainer, template_type: int,
template_path: str, template_text='', template_path: str, template_text: str = '',
save_changes=True, target_package=None) -> dict: save_changes: bool = True,
target_package: Union[Package, None] = None) -> dict:
'''Метод для запуска выполнения шаблонов.''' '''Метод для запуска выполнения шаблонов.'''
# Словарь с данными о результате работы исполнительного метода. # Словарь с данными о результате работы исполнительного метода.
self.executor_output = {'target_path': None, self.executor_output = {'target_path': None,
@ -831,7 +850,7 @@ class TemplateExecutor:
return self.executor_output return self.executor_output
def save_changes(self): def save_changes(self) -> None:
'''Метод для сохранения чего-нибудь после выполнения всех шаблонов.''' '''Метод для сохранения чего-нибудь после выполнения всех шаблонов.'''
# Пока сохраняем только получившееся содержимое config-файла. # Пока сохраняем только получившееся содержимое config-файла.
self.calculate_config_file.save_changes() self.calculate_config_file.save_changes()
@ -851,8 +870,8 @@ class TemplateExecutor:
if self.dbpkg: if self.dbpkg:
template_object.add_to_contents() template_object.add_to_contents()
def _append_remove_directory(self, def _append_remove_directory(self, template_object: TemplateWrapper
template_object: TemplateWrapper) -> None: ) -> None:
'''Метод описывающий действия для append = "remove", если шаблон -- '''Метод описывающий действия для append = "remove", если шаблон --
директория. Удаляет директорию со всем содержимым, если она есть.''' директория. Удаляет директорию со всем содержимым, если она есть.'''
if template_object.target_type is not None: if template_object.target_type is not None:
@ -964,7 +983,8 @@ class TemplateExecutor:
template_object.clear_dir_contents() template_object.clear_dir_contents()
def _append_join_file(self, template_object: TemplateWrapper, def _append_join_file(self, template_object: TemplateWrapper,
join_before=False, replace=False) -> None: join_before: bool = False, replace: bool = False
) -> None:
'''Метод описывающий действия при append = "join", если шаблон -- файл. '''Метод описывающий действия при append = "join", если шаблон -- файл.
Объединяет шаблон с целевым файлом.''' Объединяет шаблон с целевым файлом.'''
output_path = template_object.output_path output_path = template_object.output_path
@ -1178,7 +1198,8 @@ class TemplateExecutor:
self.calculate_config_file.remove_file(template_object.target_path) self.calculate_config_file.remove_file(template_object.target_path)
def _copy_from_source(self, template_object: TemplateWrapper, def _copy_from_source(self, template_object: TemplateWrapper,
chown=None, chmod=None) -> str: chown: Union[dict, None] = None,
chmod: Union[int, None] = None) -> str:
'''Метод для копирования файла, указанного в source.''' '''Метод для копирования файла, указанного в source.'''
output_path = template_object.output_path output_path = template_object.output_path
source_path = template_object.input_path source_path = template_object.input_path
@ -1361,7 +1382,7 @@ class TemplateExecutor:
return hashlib.md5(source_path.encode()).hexdigest() return hashlib.md5(source_path.encode()).hexdigest()
def _create_directory(self, template_object: TemplateWrapper, def _create_directory(self, template_object: TemplateWrapper,
path_to_create=None) -> None: path_to_create: Union[str, None] = None) -> None:
'''Метод для создания директории и, при необходимости, изменения '''Метод для создания директории и, при необходимости, изменения
владельца и доступа все директорий на пути к целевой.''' владельца и доступа все директорий на пути к целевой.'''
if path_to_create is None: if path_to_create is None:
@ -1817,7 +1838,7 @@ class TemplateExecutor:
return hasattr(error, 'errno') and error.errno == errno.EACCES and\ return hasattr(error, 'errno') and error.errno == errno.EACCES and\
'var/calculate/remote' in path_to_check 'var/calculate/remote' in path_to_check
def _is_vfat(self, path_to_check): def _is_vfat(self, path_to_check: str):
'''Метод, проверяющий является ли файловая система vfat. Нужно для того, '''Метод, проверяющий является ли файловая система vfat. Нужно для того,
чтобы знать о возможности применения chown, chmod и т.д.''' чтобы знать о возможности применения chown, chmod и т.д.'''
# Инициализируем объект для проверки примонтированных файловых систем. # Инициализируем объект для проверки примонтированных файловых систем.
@ -1833,7 +1854,7 @@ class TemplateExecutor:
class DirectoryTree: class DirectoryTree:
'''Класс реализующий дерево каталогов для пакета.''' '''Класс реализующий дерево каталогов для пакета.'''
def __init__(self, base_directory): def __init__(self, base_directory: str):
self.base_directory = base_directory self.base_directory = base_directory
self._tree = {} self._tree = {}
@ -1856,7 +1877,7 @@ class DirectoryTree:
def show_tree(self) -> None: def show_tree(self) -> None:
pprint(self._tree) pprint(self._tree)
def get_directory_tree(self, directory: str): def get_directory_tree(self, directory: str) -> "DirectoryTree":
'''Метод для получения нового дерева из ветви данного дерева, '''Метод для получения нового дерева из ветви данного дерева,
соответствующей некоторому каталогу, содержащемуся в корне данного соответствующей некоторому каталогу, содержащемуся в корне данного
дерева.''' дерева.'''
@ -1872,10 +1893,10 @@ class DirectoryTree:
else: else:
return None return None
def __setitem__(self, name: str, value): def __setitem__(self, name: str, value: Union[None, dict]) -> None:
self._tree[name] = value self._tree[name] = value
def __iter__(self): def __iter__(self) -> Iterator[str]:
if self._tree is not None: if self._tree is not None:
return iter(self._tree.keys()) return iter(self._tree.keys())
else: else:
@ -1890,35 +1911,40 @@ class DirectoryTree:
class DirectoryProcessor: class DirectoryProcessor:
'''Класс обработчика директорий шаблонов.''' '''Класс обработчика директорий шаблонов.'''
def __init__(self, action: str, datavars_module=Variables(), package='', def __init__(self, action: str,
output_module=IOModule(), dbpkg=True, datavars_module: Union[Datavars, Variables] = Variables(),
namespace: NamespaceNode = None, **groups): install: Union[str, PackageAtomName] = '',
output_module: IOModule = IOModule(), dbpkg: bool = True,
namespace: NamespaceNode = None, **groups: dict):
if isinstance(action, list): if isinstance(action, list):
self.action = action self.action = action
else: else:
self.action = [action] self.action = [action]
self.output = output_module self.output: IOModule = output_module
self.datavars_module = datavars_module self.datavars_module: Variables = datavars_module
self._namespace = namespace self._namespace: NamespaceNode = namespace
# Корневая директория. # Корневая директория.
self.cl_chroot_path: str
if 'cl_chroot_path' in datavars_module.main: if 'cl_chroot_path' in datavars_module.main:
self.cl_chroot_path = datavars_module.main.cl_chroot_path self.cl_chroot_path = datavars_module.main.cl_chroot_path
else: else:
self.cl_chroot_path = '/' self.cl_chroot_path = '/'
# Корневая директория. # Корневой путь шаблонов.
self.templates_root: str
if 'cl_root_path' in datavars_module.main: if 'cl_root_path' in datavars_module.main:
self.templates_root = join_paths(self.cl_chroot_path, self.templates_root = join_paths(self.cl_chroot_path,
datavars_module.main.cl_root_path) datavars_module.main.cl_root_path)
else: else:
self.templates_root = self.cl_chroot_path self.templates_root = self.cl_chroot_path
self.cl_ignore_files = self._get_cl_ignore_files() self.cl_ignore_files: List[str] = self._get_cl_ignore_files()
# Путь к файлу config с хэш-суммами файлов, для которых уже # Путь к файлу config с хэш-суммами файлов, для которых уже
# предлагались изменения. # предлагались изменения.
self.cl_config_path: str
if 'cl_config_path' in datavars_module.main: if 'cl_config_path' in datavars_module.main:
self.cl_config_path = self._add_chroot_path( self.cl_config_path = self._add_chroot_path(
self.datavars_module.main.cl_config_path) self.datavars_module.main.cl_config_path)
@ -1928,6 +1954,7 @@ class DirectoryProcessor:
# Путь к директории config-archive для хранения оригинальной ветки # Путь к директории config-archive для хранения оригинальной ветки
# конфигурационных файлов. # конфигурационных файлов.
self.cl_config_archive: str
if 'cl_config_archive' in datavars_module.main: if 'cl_config_archive' in datavars_module.main:
self.cl_config_archive = self._add_chroot_path( self.cl_config_archive = self._add_chroot_path(
self.datavars_module.main.cl_config_archive) self.datavars_module.main.cl_config_archive)
@ -1937,6 +1964,7 @@ class DirectoryProcessor:
# Путь к директории .execute для хранения хранения файлов скриптов, # Путь к директории .execute для хранения хранения файлов скриптов,
# полученных из шаблонов с параметром exec. # полученных из шаблонов с параметром exec.
self.cl_exec_dir_path: str
if 'cl_exec_dir_path' in datavars_module.main: if 'cl_exec_dir_path' in datavars_module.main:
self.cl_exec_dir_path = self._add_chroot_path( self.cl_exec_dir_path = self._add_chroot_path(
self.datavars_module.main.cl_exec_dir_path) self.datavars_module.main.cl_exec_dir_path)
@ -1945,7 +1973,7 @@ class DirectoryProcessor:
'/var/lib/calculate/.execute/') '/var/lib/calculate/.execute/')
# Инициализируем исполнительный модуль. # Инициализируем исполнительный модуль.
self.template_executor = TemplateExecutor( self.template_executor: TemplateExecutor = TemplateExecutor(
datavars_module=self.datavars_module, datavars_module=self.datavars_module,
chroot_path=self.cl_chroot_path, chroot_path=self.cl_chroot_path,
cl_config_archive=self.cl_config_archive, cl_config_archive=self.cl_config_archive,
@ -1966,22 +1994,23 @@ class DirectoryProcessor:
self._make_current_template_var() self._make_current_template_var()
# Инициализируем шаблонизатор. # Инициализируем шаблонизатор.
self.template_engine = TemplateEngine( self.template_engine: TemplateEngine = TemplateEngine(
datavars_module=self.datavars_module, datavars_module=self.datavars_module,
chroot_path=self.cl_chroot_path, chroot_path=self.cl_chroot_path,
appends_set=self.template_executor.available_appends) appends_set=self.template_executor.available_appends)
# Разбираем atom имя пакета, для которого накладываем шаблоны. # Разбираем atom имя пакета, для которого накладываем шаблоны.
self.for_package = False self.for_package: Union[PackageAtomName, None] = None
if package: if install:
if isinstance(package, PackageAtomName): if isinstance(install, PackageAtomName):
self.for_package = package self.for_package = install
elif isinstance(package, str): elif isinstance(install, str):
try: try:
self.for_package = self.template_engine.\ self.for_package = self.template_engine.\
parameters_processor.check_postparse_package(package) parameters_processor.check_postparse_package(
install)
except ConditionFailed as error: except ConditionFailed as error:
# ConfitionFailed потому что для проверки значения пакета, # ConditionFailed потому что для проверки значения пакета,
# используется тот же метод, что проверяет параметр package # используется тот же метод, что проверяет параметр package
# в шаблонах, а в них этот параметр играет роль условия. # в шаблонах, а в них этот параметр играет роль условия.
self.output.set_error(str(error)) self.output.set_error(str(error))
@ -1990,6 +2019,7 @@ class DirectoryProcessor:
# Получаем список директорий шаблонов. # Получаем список директорий шаблонов.
# TODO переменная список. # TODO переменная список.
self.template_paths: List[str]
if isinstance(self.datavars_module, (Datavars, NamespaceNode)): if isinstance(self.datavars_module, (Datavars, NamespaceNode)):
var_type = self.datavars_module.main[ var_type = self.datavars_module.main[
'cl_template_path'].variable_type 'cl_template_path'].variable_type
@ -2012,11 +2042,11 @@ class DirectoryProcessor:
self.packages_file_trees = OrderedDict() self.packages_file_trees = OrderedDict()
# Список обработчиков. # Список обработчиков.
self._handlers = {} self._handlers: Dict[tuple] = {}
self._handlers_queue = [] self._handlers_queue: List[str] = []
self._handling = None self._handling: Union[str, None] = None
def _get_cl_ignore_files(self) -> list: def _get_cl_ignore_files(self) -> List[str]:
'''Метод для получения из соответствующей переменной списка паттернов '''Метод для получения из соответствующей переменной списка паттернов
для обнаружения игнорируемых в ходе обработки шаблонов файлов.''' для обнаружения игнорируемых в ходе обработки шаблонов файлов.'''
if 'cl_ignore_files' in self.datavars_module.main: if 'cl_ignore_files' in self.datavars_module.main:
@ -2050,7 +2080,7 @@ class DirectoryProcessor:
package_atom: str) -> None: package_atom: str) -> None:
try: try:
groups_namespace = self.datavars_module.main.cl.groups groups_namespace = self.datavars_module.main.cl.groups
except VariableNotFoundError: except (VariableNotFoundError, AttributeError):
namespaces = ['cl', 'groups'] namespaces = ['cl', 'groups']
groups_namespace = self.datavars_module.main groups_namespace = self.datavars_module.main
for namespace in namespaces: for namespace in namespaces:
@ -2065,6 +2095,9 @@ class DirectoryProcessor:
groups_namespace[namespace] = Variables() groups_namespace[namespace] = Variables()
groups_namespace = groups_namespace[namespace] groups_namespace = groups_namespace[namespace]
print('GROUPS NAMESPACE:')
print(groups_namespace)
atom_dict = PackageAtomParser.parse_atom_name(package_atom) atom_dict = PackageAtomParser.parse_atom_name(package_atom)
if isinstance(self.datavars_module, (Datavars, NamespaceNode)): if isinstance(self.datavars_module, (Datavars, NamespaceNode)):
if group_name not in groups_namespace: if group_name not in groups_namespace:
@ -2074,24 +2107,59 @@ class DirectoryProcessor:
else: else:
group_var = groups_namespace[group_name] group_var = groups_namespace[group_name]
group_table = group_var.get_value().get_table() group_table = group_var.get_value().get_table()
if not self.check_existance_in_group(atom_dict, group_table): check_result = self.check_existance_in_group(atom_dict,
group_table)
if check_result == -2:
group_table.append(atom_dict) group_table.append(atom_dict)
elif check_result >= 0:
group_table[check_result] = atom_dict
if check_result != -1:
group_var.source = group_table group_var.source = group_table
else: else:
if group_name not in groups_namespace: if group_name not in groups_namespace:
groups_namespace[group_name] = [atom_dict] groups_namespace[group_name] = [atom_dict]
else: else:
group_table = groups_namespace[group_name] group_table = groups_namespace[group_name]
if not self.check_existance_in_group(atom_dict, group_table): check_result = self.check_existance_in_group(atom_dict,
group_table)
if check_result == -2:
group_table.append(atom_dict) group_table.append(atom_dict)
elif check_result >= 0:
group_table[check_result] = atom_dict
if check_result != -1:
groups_namespace[group_name] = group_table groups_namespace[group_name] = group_table
def check_existance_in_group(self, atom_dict: dict, group: list): def check_existance_in_group(self, atom_dict: dict, group: list) -> bool:
'''Метод для проверки наличия в таблице групп указанного пакета, а
также для сравнения данных о пакете из таблицы и из атом словаря.
Возвращает:
-2 -- если пакета нет;
-1 -- если присутствует и полностью совпадает;
>=0 -- если имеющиеся в таблице данные о пакете должены быть заменены
на указанные. В этом случае возвращается индекс старых данных в
таблице. '''
index = 0
for group_atom in group: for group_atom in group:
for field in ['category', 'name', 'version', 'slot']: if (atom_dict['category'] != group_atom['category'] and
if atom_dict[field] != group_atom[field]: atom_dict['name'] != group_atom['name']):
continue index += 1
return True continue
if (atom_dict['slot'] is not None and
group_atom['slot'] is not None):
if atom_dict['slot'] != group_atom['slot']:
return -2
if atom_dict['version'] is not None:
if (group_atom['version'] is None or
atom_dict['version'] != group_atom['version']):
return index
return -1
return -2
def _make_current_template_var(self) -> None: def _make_current_template_var(self) -> None:
var_path = ['main', 'cl'] var_path = ['main', 'cl']
@ -2120,6 +2188,7 @@ class DirectoryProcessor:
# Режим заполнения очередей директорий пакетов, необходимых для более # Режим заполнения очередей директорий пакетов, необходимых для более
# быстрой обработки параметра merge. # быстрой обработки параметра merge.
self.fill_trees = bool(self.for_package) self.fill_trees = bool(self.for_package)
if self.for_package: if self.for_package:
if self.for_package is NonePackage: if self.for_package is NonePackage:
package = self.for_package package = self.for_package
@ -2129,6 +2198,7 @@ class DirectoryProcessor:
else: else:
package = None package = None
self.base_directory: str
for directory_path in self.template_paths: for directory_path in self.template_paths:
self.base_directory = directory_path.strip() self.base_directory = directory_path.strip()
entries = os.scandir(self.base_directory) entries = os.scandir(self.base_directory)
@ -2163,11 +2233,12 @@ class DirectoryProcessor:
self.template_executor.save_changes() self.template_executor.save_changes()
return self.template_executor.changed_files return self.template_executor.changed_files
def _execute_handlers(self): def _execute_handlers(self) -> None:
'''Метод для запуска обработчиков добавленных в очередь обработчиков '''Метод для запуска обработчиков добавленных в очередь обработчиков
с помощью параметра notify.''' с помощью параметра notify.'''
self.output.set_info('Processing handlers...') self.output.set_info('Processing handlers...')
index = 0 index = 0
while index < len(self._handlers_queue): while index < len(self._handlers_queue):
handler = self._handlers_queue[index] handler = self._handlers_queue[index]
index += 1 index += 1
@ -2219,7 +2290,7 @@ class DirectoryProcessor:
FILE, handler_path, FILE, handler_path,
template_text=handler_text) template_text=handler_text)
def _merge_packages(self): def _merge_packages(self) -> None:
'''Метод для выполнения шаблонов относящихся к пакетам, указанным во '''Метод для выполнения шаблонов относящихся к пакетам, указанным во
всех встреченных значениях параметра merge.''' всех встреченных значениях параметра merge.'''
not_merged_packages = [] not_merged_packages = []
@ -2283,7 +2354,7 @@ class DirectoryProcessor:
else: else:
self.output.set_success('All packages are merged.') self.output.set_success('All packages are merged.')
def _run_exec_files(self): def _run_exec_files(self) -> None:
'''Метод для выполнения скриптов, полученных в результате обработки '''Метод для выполнения скриптов, полученных в результате обработки
шаблонов с параметром exec.''' шаблонов с параметром exec.'''
for exec_file_path, exec_info in\ for exec_file_path, exec_info in\
@ -2306,28 +2377,11 @@ class DirectoryProcessor:
except TemplateExecutorError as error: except TemplateExecutorError as error:
self.output.set_error(str(error)) self.output.set_error(str(error))
def _get_directories_queue(self, path: str) -> tuple:
'''Уже не актуальный метод для построение очередей из путей к
шаблонам. Хотя возможно еще пригодится.'''
directories_queue = []
for base_dir in self.template_paths:
base_dir = base_dir.strip()
if path.startswith(base_dir):
base_directory = base_dir
break
while path != base_directory:
path, dir_name = os.path.split(path)
directories_queue.append(dir_name)
return base_directory, directories_queue
def _walk_directory_tree(self, current_directory_path: str, def _walk_directory_tree(self, current_directory_path: str,
current_target_path: str, current_target_path: str,
directory_parameters: ParametersContainer(), directory_parameters: ParametersContainer,
directory_tree={}, directory_tree: Union[dict, DirectoryTree] = {},
package=None) -> None: package: Union[Package, None] = None) -> None:
'''Метод для рекурсивного обхода директорий с шаблонами, а также, при '''Метод для рекурсивного обхода директорий с шаблонами, а также, при
необходимости, заполнения деревьев директорий шаблонов, с помощью необходимости, заполнения деревьев директорий шаблонов, с помощью
которых далее выполняются шаблоны пакетов из merge.''' которых далее выполняются шаблоны пакетов из merge.'''
@ -2389,6 +2443,7 @@ class DirectoryProcessor:
directory_tree = {} directory_tree = {}
return return
# Хэндлеры, вложенные в другие хэндлеры не разрешены.
if self._handling is not None and directory_parameters.handler: if self._handling is not None and directory_parameters.handler:
if directory_parameters.handler != self._handling: if directory_parameters.handler != self._handling:
self.output.set_error("'handler' parameter is not" self.output.set_error("'handler' parameter is not"
@ -2403,6 +2458,8 @@ class DirectoryProcessor:
if pkg not in self.processed_packages: if pkg not in self.processed_packages:
self.packages_to_merge.add(pkg) self.packages_to_merge.add(pkg)
# Если присутствует параметр notify, в котором указаны хэндлеры для
# последующего выполнения -- добавляем их в очередь.
if directory_parameters.notify: if directory_parameters.notify:
for handler_id in directory_parameters.notify: for handler_id in directory_parameters.notify:
if handler_id not in self._handlers_queue: if handler_id not in self._handlers_queue:
@ -2425,6 +2482,7 @@ class DirectoryProcessor:
current_target_path = os.path.join(current_target_path, current_target_path = os.path.join(current_target_path,
directory_name) directory_name)
# Для директорий по умолчанию append = join.
if not directory_parameters.append: if not directory_parameters.append:
directory_parameters.set_parameter({'append': 'join'}) directory_parameters.set_parameter({'append': 'join'})
@ -2459,7 +2517,7 @@ class DirectoryProcessor:
template_path = os.path.join(current_directory_path, template_path = os.path.join(current_directory_path,
template_name) template_name)
# Применяем к файлу шаблона шаблонизатор. # Обрабатываем файл шаблона шаблонизатором.
template_text = self._parse_template(template_parameters, template_text = self._parse_template(template_parameters,
template_name, template_name,
FILE, current_directory_path) FILE, current_directory_path)
@ -2507,6 +2565,8 @@ class DirectoryProcessor:
if pkg not in self.processed_packages: if pkg not in self.processed_packages:
self.packages_to_merge.add(pkg) self.packages_to_merge.add(pkg)
# Если присутствует параметр notify, в котором указаны хэндлеры для
# последующего выполнения -- добавляем их в очередь.
if template_parameters.notify: if template_parameters.notify:
for handler_id in template_parameters.notify: for handler_id in template_parameters.notify:
if handler_id not in self._handlers_queue: if handler_id not in self._handlers_queue:
@ -2572,7 +2632,8 @@ class DirectoryProcessor:
directory_tree = {} directory_tree = {}
return return
def _scan_directory(self, directory_path: str) -> tuple: def _scan_directory(self, directory_path: str
) -> Tuple[List[str], List[str]]:
'''Метод для получения и фильтрования списка файлов и директорий, '''Метод для получения и фильтрования списка файлов и директорий,
содержащихся в директории шаблонов.''' содержащихся в директории шаблонов.'''
template_files = [] template_files = []
@ -2604,9 +2665,11 @@ class DirectoryProcessor:
return False return False
return True return True
def _get_files_and_dirs_from_tree(self, template_files, def _get_files_and_dirs_from_tree(self, template_files: List[str],
template_directories, template_directories: List[str],
directory_tree: DirectoryTree): directory_tree: DirectoryTree
) -> Tuple[List[str],
List[DirectoryTree]]:
'''Метод для получения списков файлов и директорий из дерева '''Метод для получения списков файлов и директорий из дерева
директорий.''' директорий.'''
tree_files = [] tree_files = []
@ -2622,8 +2685,8 @@ class DirectoryProcessor:
return tree_directories, tree_files return tree_directories, tree_files
def _make_target_path(self, target_path, template_name, def _make_target_path(self, target_path: str, template_name: str,
parameters): parameters: ParametersContainer) -> str:
'''Метод для получения пути к целевому файлу с учетом наличия '''Метод для получения пути к целевому файлу с учетом наличия
параметров name, path и append = skip.''' параметров name, path и append = skip.'''
# Если есть параметр name -- меняем имя шаблона. # Если есть параметр name -- меняем имя шаблона.
@ -2642,10 +2705,10 @@ class DirectoryProcessor:
template_name) template_name)
return target_path return target_path
def _parse_template(self, parameters, def _parse_template(self, parameters: ParametersContainer,
template_name, template_name: str,
template_type, template_type: int,
template_directory): template_directory: str) -> Union[str, bool]:
'''Метод для разбора шаблонов, получения значений их параметров и их '''Метод для разбора шаблонов, получения значений их параметров и их
текста после отработки шаблонизитора.''' текста после отработки шаблонизитора.'''
if template_type == DIR: if template_type == DIR:
@ -2657,6 +2720,7 @@ class DirectoryProcessor:
self.datavars_module.main.cl['current_template'].set(template_path) self.datavars_module.main.cl['current_template'].set(template_path)
else: else:
self.datavars_module.main.cl['current_template'] = template_path self.datavars_module.main.cl['current_template'] = template_path
try: try:
self.template_engine.process_template(template_name, self.template_engine.process_template(template_name,
template_type, template_type,
@ -2673,12 +2737,13 @@ class DirectoryProcessor:
template_path)) template_path))
return False return False
def _execute_template(self, target_path, def _execute_template(self, target_path: str,
parameters, parameters: ParametersContainer,
template_type, template_type: int,
template_path, template_path: str,
template_text='', template_text: str = '',
package=None): package: Union[Package, None] = None
) -> Union[bool, str]:
'''Метод для наложения шаблонов и обработки информации полученной после '''Метод для наложения шаблонов и обработки информации полученной после
наложения.''' наложения.'''
try: try:
@ -2727,7 +2792,8 @@ class DirectoryProcessor:
format(template_path)) format(template_path))
return target_path return target_path
def _update_package_tree(self, package, current_level_tree): def _update_package_tree(self, package: Package,
current_level_tree: Union[None, dict]) -> None:
'''Метод для обновления деревьев директорий пакетов, необходимых для '''Метод для обновления деревьев директорий пакетов, необходимых для
обработки шаблонов пакетов из значения параметра merge.''' обработки шаблонов пакетов из значения параметра merge.'''
# Если текущему уровню соответствует заглушка None или он содержит # Если текущему уровню соответствует заглушка None или он содержит
@ -2743,12 +2809,13 @@ class DirectoryProcessor:
# Если для данного пакета еще нет дерева -- # Если для данного пакета еще нет дерева --
# копируем для него текущее. # копируем для него текущее.
directory_tree = DirectoryTree(self.base_directory) directory_tree = DirectoryTree(self.base_directory)
directory_tree.update_tree( directory_tree.update_tree(copy.deepcopy(self.directory_tree))
copy.deepcopy(self.directory_tree))
self.packages_file_trees[package] = directory_tree self.packages_file_trees[package] = directory_tree
def _check_package_and_action(self, parameters, template_path, def _check_package_and_action(self, parameters: ParametersContainer,
directory_tree=None): template_path: str,
directory_tree: Union[dict, None] = None
) -> bool:
'''Метод для проверки параметров action и package во время обработки '''Метод для проверки параметров action и package во время обработки
каталогов с шаблонами. Если среди аргументов указано также каталогов с шаблонами. Если среди аргументов указано также
дерево каталогов, то в случае несовпадения значений package для файла дерево каталогов, то в случае несовпадения значений package для файла
@ -2790,6 +2857,9 @@ class DirectoryProcessor:
if parameters.package != self.for_package: if parameters.package != self.for_package:
if directory_tree is not None: if directory_tree is not None:
# Если есть дерево, которое собирается для текущих шаблонов
# и параметр package шаблона не совпадает с текущим,
# ставим заглушку, в качестве которой используется None.
template_name = os.path.basename(template_path) template_name = os.path.basename(template_path)
directory_tree[template_name] = None directory_tree[template_name] = None
self.output.set_warning( self.output.set_warning(
@ -2804,17 +2874,17 @@ class DirectoryProcessor:
return True return True
@contextmanager @contextmanager
def _start_handling(self, handler): def _start_handling(self, handler_id: str):
'''Метод для перевода обработчика каталогов в режим обработки '''Метод для перевода обработчика каталогов в режим обработки
хэндлеров.''' хэндлеров.'''
try: try:
self._handling = handler self._handling = handler_id
yield self yield self
finally: finally:
self._handling = None self._handling = None
@contextmanager @contextmanager
def _set_current_package(self, package): def _set_current_package(self, package: Package):
'''Метод для указания в шаблонизаторе пакета, для которого в данный '''Метод для указания в шаблонизаторе пакета, для которого в данный
момент проводим настройку. Пока не используется.''' момент проводим настройку. Пока не используется.'''
try: try:

6
run_templates.py
Unescape View File

@ -11,7 +11,7 @@ def main():
parser = argparse.ArgumentParser('Run templates.') parser = argparse.ArgumentParser('Run templates.')
parser.add_argument('-a', '--action', action='append', type=str, nargs='+', parser.add_argument('-a', '--action', action='append', type=str, nargs='+',
help="action parameter value.") help="action parameter value.")
parser.add_argument('-p', '--package', type=str, parser.add_argument('-p', '--install', type=str,
help="atom name of a target package.") help="atom name of a target package.")
parser.add_argument('--dbpkg', action='store_true', parser.add_argument('--dbpkg', action='store_true',
help=("flag for switching template engine's mode from" help=("flag for switching template engine's mode from"
@ -27,8 +27,10 @@ def main():
datavars = Datavars() datavars = Datavars()
io_module = IOModule() io_module = IOModule()
if args.package == 'None': if args.install is None:
package = NonePackage package = NonePackage
elif args.install.strip().casefold() == 'all':
package = None
else: else:
package = args.package package = args.package

87
tests/templates/test_directory_processor.py
Unescape View File

@ -663,7 +663,7 @@ class TestDirectoryProcessor:
'templates_17') 'templates_17')
directory_processor = DirectoryProcessor('install', directory_processor = DirectoryProcessor('install',
datavars_module=datavars, datavars_module=datavars,
package=other_package_name) install=other_package_name)
directory_processor.process_template_directories() directory_processor.process_template_directories()
test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH) test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH)
@ -689,7 +689,7 @@ class TestDirectoryProcessor:
'templates_18') 'templates_18')
directory_processor = DirectoryProcessor('install', directory_processor = DirectoryProcessor('install',
datavars_module=datavars, datavars_module=datavars,
package=other_package_name) install=other_package_name)
directory_processor.process_template_directories() directory_processor.process_template_directories()
test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH) test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH)
@ -777,7 +777,7 @@ class TestDirectoryProcessor:
'templates_21') 'templates_21')
directory_processor = DirectoryProcessor('install', directory_processor = DirectoryProcessor('install',
datavars_module=datavars, datavars_module=datavars,
package=other_package_name) install=other_package_name)
directory_processor.process_template_directories() directory_processor.process_template_directories()
test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH) test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH)
@ -820,7 +820,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/other-package') install='test-category/other-package')
directory_processor.process_template_directories() directory_processor.process_template_directories()
test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH) test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH)
@ -863,7 +863,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/other-package') install='test-category/other-package')
directory_processor.process_template_directories() directory_processor.process_template_directories()
test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH) test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH)
@ -1109,7 +1109,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH) test_package = Package(test_package_name_1, chroot_path=CHROOT_PATH)
@ -1205,7 +1205,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
@ -1233,9 +1233,9 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package', install='test-category/test-package',
build='test-category/build-package-2.1:0[use_1,use_2]', build='test-category/build-package-2.1:0[use_1 use_2]',
uninstall='test-category/unmerged-package-1.0.1:1[use_1,use_2]' uninstall='test-category/unmerged-package-1.0.1:1[use_1 use_2]'
) )
assert 'build' in datavars.main.cl.groups assert 'build' in datavars.main.cl.groups
assert 'uninstall' in datavars.main.cl.groups assert 'uninstall' in datavars.main.cl.groups
@ -1257,6 +1257,7 @@ class TestDirectoryProcessor:
'etc/dir_51'): 'N', 'etc/dir_51'): 'N',
join_paths(CHROOT_PATH, join_paths(CHROOT_PATH,
'etc/dir_51/file_0'): 'N'} 'etc/dir_51/file_0'): 'N'}
datavars.main.cl['groups'] = Variables({})
def test_group_parameter_with_merge_parameter(self): def test_group_parameter_with_merge_parameter(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH, datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
@ -1264,9 +1265,9 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package', install='test-category/test-package',
build='test-category/build-package-2.1:0[use_1,use_2]', build='test-category/build-package-2.1:0[use_1 use_2]',
uninstall='test-category/unmerged-package-1.0.1:1[use_1,use_2]' uninstall='test-category/unmerged-package-1.0.1:1[use_1 use_2]'
) )
assert 'build' in datavars.main.cl.groups assert 'build' in datavars.main.cl.groups
assert 'uninstall' in datavars.main.cl.groups assert 'uninstall' in datavars.main.cl.groups
@ -1287,6 +1288,7 @@ class TestDirectoryProcessor:
'etc/dir_56'): 'N', 'etc/dir_56'): 'N',
join_paths(CHROOT_PATH, join_paths(CHROOT_PATH,
'etc/dir_56/file_0'): 'N'} 'etc/dir_56/file_0'): 'N'}
# datavars.main.cl['groups'] = Variables({})
def test_solving_collisions_for_the_same_packages_from_different_slots(self): def test_solving_collisions_for_the_same_packages_from_different_slots(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH, datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
@ -1294,7 +1296,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
header = ('#' + '-' * 79 + '\n' + header = ('#' + '-' * 79 + '\n' +
@ -1361,7 +1363,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package:1' install='test-category/test-package:1'
) )
output_text = "options {\n parameter-0 yes;\n};\n" output_text = "options {\n parameter-0 yes;\n};\n"
@ -1389,7 +1391,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/copy.gif')) assert os.path.exists(join_paths(CHROOT_PATH, '/etc/copy.gif'))
@ -1400,7 +1402,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_60/file_0')) assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_60/file_0'))
@ -1413,7 +1415,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_61/file_0')) assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_61/file_0'))
@ -1443,7 +1445,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_62/file_0')) assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_62/file_0'))
@ -1458,7 +1460,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_63/file_0')) assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_63/file_0'))
@ -1469,7 +1471,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'update', 'update',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package' install='test-category/test-package'
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_65/file_0')) assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_65/file_0'))
@ -1481,7 +1483,7 @@ class TestDirectoryProcessor:
directory_processor = DirectoryProcessor( directory_processor = DirectoryProcessor(
'install', 'install',
datavars_module=datavars, datavars_module=datavars,
package='test-category/test-package', install='test-category/test-package',
output_module=io output_module=io
) )
directory_processor.process_template_directories() directory_processor.process_template_directories()
@ -1491,6 +1493,47 @@ class TestDirectoryProcessor:
"Warning: package 'test-category/new-package-0.1.1'" "Warning: package 'test-category/new-package-0.1.1'"
" not found for action 'install'.") " not found for action 'install'.")
def test_group_and_package_parameter_for_unexisting_package(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_49')
directory_processor = DirectoryProcessor(
'merge',
datavars_module=datavars,
uninstall='test-category/unmerged-package-1.0.1:1[use_1 use_2]'
)
directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_67/file_0'))
def test_group_and_some_templates_ignored(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_50')
directory_processor = DirectoryProcessor(
'unmerge',
datavars_module=datavars,
uninstall='test-category/unmerged-package-1.0.1:1[use_1 use_2]'
)
directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_68/file_0'))
assert not os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_69/file_0'))
def test_group_with_NonePackage(self):
datavars.main['cl_template_path'] = os.path.join(CHROOT_PATH,
'templates_51')
directory_processor = DirectoryProcessor(
'unmerge',
datavars_module=datavars,
install=NonePackage,
uninstall='test-category/unmerged-package-1.0.1:1[use_1 use_2]',
build='test-category/build-package'
)
directory_processor.process_template_directories()
assert os.path.exists(join_paths(CHROOT_PATH, '/etc/dir_70/file_0'))
assert not os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_71/file_0'))
assert not os.path.exists(join_paths(CHROOT_PATH,
'/etc/dir_72/file_0'))
def test_view_tree(self): def test_view_tree(self):
list_path = join_paths(CHROOT_PATH, '/etc') list_path = join_paths(CHROOT_PATH, '/etc')
show_tree(list_path) show_tree(list_path)

9
tests/templates/testfiles/test_dir_processor_root/etc.backup/._cfg0001_file_12.backup
Unescape View File

@ -1,9 +0,0 @@
#-------------------------------------------------------------------------------
# Modified by Calculate Utilities 4.0
# Processed template files:
# /path/to/template
#-------------------------------------------------------------------------------
section-name {
parameter-1 yes;
parameter-2 no;
};

9
tests/templates/testfiles/test_dir_processor_root/etc.backup/._cfg0002_file_12.backup
Unescape View File

@ -1,9 +0,0 @@
#-------------------------------------------------------------------------------
# Modified by Calculate Utilities 4.0
# Processed template files:
# /home/divanov/Home/development/calculate-lib/tests/templates/testfiles/test_dir_processor_root/templates_28/root/file_12
#-------------------------------------------------------------------------------
section-name {
parameter-1 yes;
parameter-3 10;
};

2
tests/templates/testfiles/test_dir_processor_root/templates_40/install/.calculate_directory
Unescape View File

@ -1 +1 @@
{% calculate append = 'skip', path = '/etc', action = 'install' %} {% calculate append = 'skip', action = 'install' %}

1
tests/templates/testfiles/test_dir_processor_root/templates_49/merge/.calculate_directory
Unescape View File

@ -0,0 +1 @@
{% calculate action = "merge", append = "skip" %}

1
tests/templates/testfiles/test_dir_processor_root/templates_49/merge/dir_67/.calculate_directory
Unescape View File

@ -0,0 +1 @@
{% calculate append = "skip", group = "uninstall" %}

12
tests/templates/testfiles/test_dir_processor_root/templates_49/merge/dir_67/script
Unescape View File

@ -0,0 +1,12 @@
{% calculate run = '/usr/bin/python' %}
import os
text = '''Please, take this and run
far away, far away from me
I am tainted
And happyness, and peace of mind
Was never meant for me.'''
os.mkdir('./dir_67')
with open('./dir_67/file_0', 'w') as output_file:
output_file.write(text)

1
tests/templates/testfiles/test_dir_processor_root/templates_50/unmerge/.calculate_directory
Unescape View File

@ -0,0 +1 @@
{% calculate action = "unmerge", append = "skip" %}

2
tests/templates/testfiles/test_dir_processor_root/templates_50/unmerge/dir_68/.calculate_directory
Unescape View File

@ -0,0 +1,2 @@
{% calculate append = "skip", group = "uninstall",
package = 'test-category/unmerged-package' %}

12
tests/templates/testfiles/test_dir_processor_root/templates_50/unmerge/dir_68/script
Unescape View File

@ -0,0 +1,12 @@
{% calculate run = '/usr/bin/python' %}
import os
text = '''Please, take this and run
far away, far away from me
I am tainted
And happyness, and peace of mind
Was never meant for me.'''
os.mkdir('./dir_68')
with open('./dir_68/file_0', 'w') as output_file:
output_file.write(text)

2
tests/templates/testfiles/test_dir_processor_root/templates_50/unmerge/dir_69/.calculate_directory
Unescape View File

@ -0,0 +1,2 @@
{% calculate append = "skip", group = "uninstall",
package = 'test-category/build-package'%}

12
tests/templates/testfiles/test_dir_processor_root/templates_50/unmerge/dir_69/script
Unescape View File

@ -0,0 +1,12 @@
{% calculate run = '/usr/bin/python' %}
import os
text = '''Please, take this and run
far away, far away from me
I am tainted
And happyness, and peace of mind
Was never meant for me.'''
os.mkdir('./dir_69')
with open('./dir_69/file_0', 'w') as output_file:
output_file.write(text)

1
tests/templates/testfiles/test_dir_processor_root/templates_51/merge/.calculate_directory
Unescape View File

@ -0,0 +1 @@
{% calculate action = "unmerge", append = "skip" %}

2
tests/templates/testfiles/test_dir_processor_root/templates_51/merge/dir_70/.calculate_directory
Unescape View File

@ -0,0 +1,2 @@
{% calculate append = "skip", group = "uninstall",
package = 'test-category/unmerged-package' %}

12
tests/templates/testfiles/test_dir_processor_root/templates_51/merge/dir_70/script
Unescape View File

@ -0,0 +1,12 @@
{% calculate run = '/usr/bin/python' %}
import os
text = '''Please, take this and run
far away, far away from me
I am tainted
And happyness, and peace of mind
Was never meant for me.'''
os.mkdir('./dir_70')
with open('./dir_70/file_0', 'w') as output_file:
output_file.write(text)

2
tests/templates/testfiles/test_dir_processor_root/templates_51/merge/dir_71/.calculate_directory
Unescape View File

@ -0,0 +1,2 @@
{% calculate append = "skip", group = "uninstall",
package = 'test-category/build-package'%}

12
tests/templates/testfiles/test_dir_processor_root/templates_51/merge/dir_71/script
Unescape View File

@ -0,0 +1,12 @@
{% calculate run = '/usr/bin/python' %}
import os
text = '''Please, take this and run
far away, far away from me
I am tainted
And happyness, and peace of mind
Was never meant for me.'''
os.mkdir('./dir_71')
with open('./dir_71/file_0', 'w') as output_file:
output_file.write(text)

1
tests/templates/testfiles/test_dir_processor_root/templates_51/merge/dir_72/.calculate_directory
Unescape View File

@ -0,0 +1 @@
{% calculate append = 'join', package = 'test-category/test-package' %}

1
tests/templates/testfiles/test_dir_processor_root/templates_51/merge/dir_72/file_0
Unescape View File

@ -0,0 +1 @@
{% calculate format = 'raw', append = 'join' %}
Write Preview
Loading…
Cancel
Save