"""Annotate services 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 os.path import basename from typing import Tuple from rougail.i18n import _ from rougail.utils import normalize_family from rougail.error import DictConsistencyError # a object's attribute has some annotations # that shall not be present in the exported (flatened) XML ERASED_ATTRIBUTES = ('redefine', 'namespace', 'xmlfiles', 'disabled', 'name', 'manage') ERASED_ATTRIBUTES2 = ('redefine', 'namespace', 'xmlfiles', 'disabled') ALLOW_ATTRIBUT_NOT_MANAGE = ['file', 'engine', 'target'] class Annotator: """Manage service's object for example:: """ level = 20 def __init__(self, objectspace, *args, ) -> None: self.objectspace = objectspace self.uniq_overrides = [] if 'network_type' not in self.objectspace.types: self.objectspace.types['network_type'] = self.objectspace.types['ip_type'] if hasattr(self.objectspace.space, 'services'): if not hasattr(self.objectspace.space.services, 'service'): del self.objectspace.space.services else: self.convert_services() def convert_services(self): """convert services to variables """ self.objectspace.space.services.hidden = True self.objectspace.space.services.name = 'services' self.objectspace.space.services.doc = 'services' self.objectspace.space.services.path = 'services' for service_name, service in self.objectspace.space.services.service.items(): service.name = normalize_family(service_name) activate_obj = self._generate_element('boolean', None, None, 'activate', not service.disabled, service, '.'.join(['services', service.name, 'activate']), ) service.disabled = None for elttype, values in dict(vars(service)).items(): if elttype == 'servicelist': self.objectspace.list_conditions.setdefault('servicelist', {}).setdefault( values, []).append(activate_obj) continue if elttype in ERASED_ATTRIBUTES: continue if not service.manage and elttype not in ALLOW_ATTRIBUT_NOT_MANAGE: msg = _(f'unmanage service cannot have "{elttype}"') raise DictConsistencyError(msg, 66, service.xmlfiles) if isinstance(values, (dict, list)): if elttype != 'ip': eltname = elttype + 's' else: eltname = elttype path = '.'.join(['services', service.name, eltname]) family = self._gen_family(eltname, path, service.xmlfiles, with_informations=False, ) if isinstance(values, dict): values = list(values.values()) family.family = self.make_group_from_elts(service_name, elttype, values, path, ) setattr(service, elttype, family) else: if not hasattr(service, 'information'): service.information = self.objectspace.information(service.xmlfiles) setattr(service.information, elttype, values) service.path = '.'.join(['services', service.name]) manage = self._generate_element('boolean', None, None, 'manage', service.manage, service, '.'.join(['services', service.name, 'manage']), ) service.variable = [activate_obj, manage] service.doc = service_name def make_group_from_elts(self, service_name, elttype, elts, path, ): """Splits each objects into a group (and `OptionDescription`, in tiramisu terms) and build elements and its attributes (the `Options` in tiramisu terms) """ families = [] listname = '{}list'.format(elttype) for elt in elts: # try to launch _update_xxxx() function update_elt = '_update_' + elttype if hasattr(self, update_elt): getattr(self, update_elt)(elt, service_name, ) c_name, subpath = self._get_name_path(elt, path, ) family = self._gen_family(c_name, subpath, elt.xmlfiles, ) family.variable = [] if hasattr(elt, 'disabled'): disabled = elt.disabled else: disabled = False activate_obj = self._generate_element('boolean', None, None, 'activate', not disabled, elt, '.'.join([subpath, 'activate']), ) for key in dir(elt): if key.startswith('_') or key.endswith('_type') or key in ERASED_ATTRIBUTES2: continue value = getattr(elt, key) if key == listname: self.objectspace.list_conditions.setdefault(listname, {}).setdefault( value, []).append(activate_obj) continue if key == 'name': dtd_key_type = elttype + '_type' else: dtd_key_type = key + '_type' elt_type = getattr(elt, dtd_key_type, None) if elt_type: if elt_type == 'variable': elt_type = 'symlink' family.variable.append(self._generate_element(elt_type, dtd_key_type, elttype, key, value, elt, f'{subpath}.{key}' )) else: setattr(family.information, key, value) family.variable.append(activate_obj) families.append(family) return families def _get_name_path(self, elt, path: str, ) -> Tuple[str, str]: # create element name, if already exists, add _xx to be uniq if hasattr(elt, 'source') and elt.source: name = elt.source else: name = elt.name idx = 0 while True: c_name = name if idx: c_name += f'_{idx}' subpath = '{}.{}'.format(path, normalize_family(c_name)) try: self.objectspace.paths.get_family(subpath, 'services') except DictConsistencyError as err: if err.errno == 42: return c_name, subpath idx += 1 def _gen_family(self, name, path, xmlfiles, with_informations=True, ): family = self.objectspace.family(xmlfiles) family.name = normalize_family(name) family.doc = name family.mode = None self.objectspace.paths.add_family('services', path, family, None, ) if with_informations: family.information = self.objectspace.information(xmlfiles) return family def _generate_element(self, type_, dtd_key_type, elttype, key, value, elt, path, ): # pylint: disable=R0913 variable = self.objectspace.variable(elt.xmlfiles) variable.name = normalize_family(key) variable.mode = None variable.type = type_ if type_ == 'symlink': variable.opt = self.objectspace.paths.get_variable(value, xmlfiles=elt.xmlfiles) variable.multi = None needed_type = self.objectspace.types[dtd_key_type] if needed_type not in ('variable', variable.opt.type): msg = _(f'"{value}" in "{elttype}" must be a variable with type ' f'"{needed_type}" not "{variable.opt.type}"') raise DictConsistencyError(msg, 58, elt.xmlfiles) else: variable.doc = key variable.default = value variable.namespace = 'services' self.objectspace.paths.add_variable('services', path, 'service', False, variable, ) return variable def _update_override(self, override, service_name, ): if service_name in self.uniq_overrides: msg = _('only one override is allowed by service, ' 'please use a variable multiple if you want have more than one IP') raise DictConsistencyError(msg, 69, override.xmlfiles) self.uniq_overrides.append(service_name) override.name = service_name if not hasattr(override, 'source'): override.source = service_name @staticmethod def _update_file(file_, service_name, ): if not hasattr(file_, 'file_type') or file_.file_type == "filename": if not hasattr(file_, 'source'): file_.source = basename(file_.name) elif not hasattr(file_, 'source'): msg = _(f'attribute "source" is mandatory for the file "{file_.name}" ' f'"({service_name})"') raise DictConsistencyError(msg, 34, file_.xmlfiles) def _update_ip(self, ip, service_name, ) -> None: variable = self.objectspace.paths.get_variable(ip.name, ip.xmlfiles) if variable.type not in ['ip', 'network', 'network_cidr']: msg = _(f'ip cannot be linked to "{variable.type}" variable "{ip.name}"') raise DictConsistencyError(msg, 70, ip.xmlfiles) if variable.type in ['ip', 'network_cidr'] and hasattr(ip, 'netmask'): msg = _(f'ip with ip_type "{variable.type}" must not have netmask') raise DictConsistencyError(msg, 59, ip.xmlfiles) if variable.type == 'network' and not hasattr(ip, 'netmask'): msg = _(f'ip with ip_type "{variable.type}" must have netmask') raise DictConsistencyError(msg, 64, ip.xmlfiles) if hasattr(ip, 'netmask'): netmask = self.objectspace.paths.get_variable(ip.netmask, ip.xmlfiles) if netmask.type != 'netmask': msg = _(f'netmask in ip must have type "netmask", not "{netmask.type}"') raise DictConsistencyError(msg, 65, ip.xmlfiles)