"""Template langage for Rougail Created by: EOLE (http://eole.orion.education.fr) Copyright (C) 2005-2018 Forked by: Cadoles (http://www.cadoles.com) Copyright (C) 2019-2021 distribued with GPL-2 or later license This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA """ from shutil import copy import logging from typing import Dict, Any from subprocess import call from os import listdir, makedirs, getcwd, chdir from os.path import dirname, join, isfile, isdir, abspath try: from tiramisu3 import Config from tiramisu3.error import PropertiesOptionError # pragma: no cover except ModuleNotFoundError: # pragma: no cover from tiramisu import Config from tiramisu.error import PropertiesOptionError from ..config import RougailConfig from ..error import FileNotFound, TemplateError from ..i18n import _ from ..utils import load_modules from . import engine as engines ENGINES = {} for engine in engines.__all__: ENGINES[engine] = getattr(engines, engine) log = logging.getLogger(__name__) log.addHandler(logging.NullHandler()) class RougailLeaderIndex: """This object is create when access to a specified Index of the variable """ def __init__(self, value, follower, index, ) -> None: self._value = value self._follower = follower self._index = index def __getattr__(self, name): if name not in self._follower: raise AttributeError() value = self._follower[name] if isinstance(value, PropertiesOptionError): raise AttributeError() return value def __str__(self): return str(self._value) def __lt__(self, value): return self._value.__lt__(value) def __le__(self, value): return self._value.__le__(value) def __eq__(self, value): return self._value.__eq__(value) def __ne__(self, value): return self._value.__ne__(value) def __gt__(self, value): return self._value.__gt__(value) def __ge__(self, value): return self._value >= value def __add__(self, value): return self._value.__add__(value) def __radd__(self, value): return value + self._value class RougailLeader: """Implement access to leader and follower variable For examples: %%leader, %%leader[0].follower1 """ def __init__(self, value, ) -> None: self._value = value self._follower = {} def __getitem__(self, index): """Get a leader.follower at requested index. """ followers = {key: values[index] for key, values in self._follower.items()} return RougailLeaderIndex(self._value[index], followers, index, ) def __iter__(self): """Iterate over leader.follower. Return synchronised value of leader.follower. """ for index in range(len(self._value)): yield self.__getitem__(index) def __len__(self): return len(self._value) def __contains__(self, value): return self._value.__contains__(value) async def add_follower(self, config, name: str, path: str, ): """Add a new follower """ self._follower[name] = [] for index in range(len(self._value)): try: value = await config.option(path, index).value.get() except PropertiesOptionError as err: value = err self._follower[name].append(value) class RougailExtra: """Object that implement access to extra variable For example %%extra1.family.variable """ def __init__(self, suboption: Dict) -> None: self.suboption = suboption def __getattr__(self, key: str, ) -> Any: try: return self.suboption[key] except KeyError: raise AttributeError def __iter__(self): return iter(self.suboption.values()) class RougailBaseTemplate: """Engine to process Creole cheetah template """ def __init__(self, # pylint: disable=R0913 config: Config, rougailconfig: RougailConfig=None, ) -> None: if rougailconfig is None: rougailconfig = RougailConfig self.config = config self.destinations_dir = abspath(rougailconfig['destinations_dir']) self.tmp_dir = abspath(rougailconfig['tmp_dir']) self.templates_dir = abspath(rougailconfig['templates_dir']) self.patches_dir = abspath(rougailconfig['patches_dir']) eos = {} functions_file = rougailconfig['functions_file'] if isfile(functions_file): eosfunc = load_modules(functions_file) for func in dir(eosfunc): if not func.startswith('_'): eos[func] = getattr(eosfunc, func) self.eosfunc = eos self.rougail_variables_dict = {} self.rougailconfig = rougailconfig self.log = log self.engines = ENGINES def patch_template(self, filename: str, ) -> None: """Apply patch to a template """ patch_cmd = ['patch', '-d', self.tmp_dir, '-N', '-p1', '-f'] patch_no_debug = ['-s', '-r', '-', '--backup-if-mismatch'] patch_file = join(self.patches_dir, f'{filename}.patch') if isfile(patch_file): self.log.info(_("Patching template '{filename}' with '{patch_file}'")) ret = call(patch_cmd + patch_no_debug + ['-i', patch_file]) if ret: # pragma: no cover patch_cmd_err = ' '.join(patch_cmd + ['-i', patch_file]) msg = _(f"Error applying patch: '{patch_file}'\n" f"To reproduce and fix this error {patch_cmd_err}") self.log.error(_(msg)) copy(join(self.templates_dir, filename), self.tmp_dir) def prepare_template(self, filename: str, ) -> None: """Prepare template source file """ self.log.info(_("Copy template: '{filename}' -> '{self.tmp_dir}'")) if not isdir(self.tmp_dir): raise TemplateError(_(f'cannot find tmp_dir {self.tmp_dir}')) copy(filename, self.tmp_dir) self.patch_template(filename) def instance_file(self, filevar: Dict, type: str, service_name: str, ) -> None: """Run templatisation on one file """ self.log.info(_("Instantiating file '{filename}'")) if 'variable' in filevar: variable = filevar['variable'] else: variable = None filenames = filevar['name'] if not isinstance(filenames, list): filenames = [filenames] if variable and not isinstance(variable, list): variable = [variable] if not isdir(self.destinations_dir): raise TemplateError(_(f'cannot find destinations_dir {self.destinations_dir}')) for idx, filename, in enumerate(filenames): if variable: var = variable[idx] else: var = None func = f'_instance_{type}' filename, source, destfile, var = getattr(self, func)(filevar, filename, service_name, variable, idx, ) destfilename = join(self.destinations_dir, destfile[1:]) makedirs(dirname(destfilename), exist_ok=True) self.log.info(_(f"{filevar['engine']} processing: '{destfilename}'")) self.engines[filevar['engine']].process(filename=filename, source=source, true_destfilename=destfile, destfilename=destfilename, variable=var, rougail_variables_dict=self.rougail_variables_dict, eosfunc=self.eosfunc, ) async def instance_files(self) -> None: """Run templatisation on all files """ ori_dir = getcwd() chdir(self.templates_dir) for option in await self.config.option.list(type='all'): namespace = await option.option.name() is_var_namespace = namespace == self.rougailconfig['variable_namespace'] self.rougail_variables_dict[namespace] = await self.load_variables(option, is_var_namespace, ) for template in listdir('.'): self.prepare_template(template) for service_obj in await self.config.option('services').list('all'): service_name = await service_obj.option.name() for fills in await service_obj.list('all'): type_ = await fills.option.name() for fill_obj in await fills.list('all'): fill = await fill_obj.value.dict() if fill['activate']: self.instance_file(fill, type_, service_name) else: self.log.debug(_("Instantiation of file '{filename}' disabled")) self.post_instance() chdir(ori_dir) def post_instance(self): pass def _instance_ip(self, *args, ) -> None: raise NotImplementedError(_('cannot instanciate this service type ip')) def _instance_files(self, *args, ) -> None: raise NotImplementedError(_('cannot instanciate this service type file')) def _instance_overrides(self, *args, ) -> None: raise NotImplementedError(_('cannot instanciate this service type override')) async def load_variables(self, optiondescription, is_variable_namespace, ) -> RougailExtra: """Load all variables and set it in RougailExtra objects """ variables = {} for option in await optiondescription.list('all'): if await option.option.isoptiondescription(): if await option.option.isleadership(): for idx, suboption in enumerate(await option.list('all')): if idx == 0: leader = RougailLeader(await suboption.value.get()) leader_name = await suboption.option.name() if is_variable_namespace: self.rougail_variables_dict[await suboption.option.name()] = leader else: await leader.add_follower(self.config, await suboption.option.name(), await suboption.option.path(), ) variables[leader_name] = leader else: subfamilies = await self.load_variables(option, is_variable_namespace, ) variables[await option.option.name()] = subfamilies else: if is_variable_namespace: value = await option.value.get() self.rougail_variables_dict[await option.option.name()] = value if await option.option.issymlinkoption() and await option.option.isfollower(): value = [] path = await option.option.path() for index in range(await option.value.len()): value.append(await self.config.option(path, index).value.get()) else: value = await option.value.get() variables[await option.option.name()] = value return RougailExtra(variables)