595 lines
25 KiB
Python
595 lines
25 KiB
Python
"""creole loader
|
|
flattened XML specific
|
|
"""
|
|
from os.path import join, isfile, isdir
|
|
from os import listdir
|
|
#from ast import literal_eval
|
|
from lxml.etree import parse, DTD
|
|
|
|
from tiramisu import (StrOption, OptionDescription, DynOptionDescription, PortOption,
|
|
IntOption, ChoiceOption, BoolOption, SymLinkOption, IPOption,
|
|
NetworkOption, NetmaskOption, DomainnameOption, BroadcastOption,
|
|
URLOption, EmailOption, FilenameOption, UsernameOption, DateOption,
|
|
PasswordOption, BoolOption, MACOption, Leadership, submulti,
|
|
Params, ParamSelfOption, ParamOption, ParamValue, Calculation, calc_value,
|
|
groups, owners)
|
|
from tiramisu.error import ConfigError
|
|
|
|
from .config import dtdfilename
|
|
from .i18n import _
|
|
#For compatibility
|
|
from .xmlreflector import HIGH_COMPATIBILITY
|
|
#from . import eosfunc
|
|
from .objspace import CreoleObjSpace
|
|
from .utils import normalize_family
|
|
import imp
|
|
|
|
|
|
FUNC_TO_DICT = ['valid_not_equal']
|
|
|
|
|
|
class ConvertDynOptionDescription(DynOptionDescription):
|
|
def convert_suffix_to_path(self, suffix):
|
|
if not isinstance(suffix, str):
|
|
suffix = str(suffix)
|
|
return normalize_family(suffix,
|
|
check_name=False)
|
|
|
|
|
|
class CreoleLoaderError(Exception):
|
|
pass
|
|
|
|
|
|
def convert_tiramisu_value(value, obj):
|
|
"""
|
|
convertit les variables dans le bon type si nécessaire
|
|
"""
|
|
if value is None:
|
|
return value
|
|
def _convert_boolean(value):
|
|
if isinstance(value, bool):
|
|
return value
|
|
prop = {'True': True,
|
|
'False': False,
|
|
'None': None}
|
|
if value not in prop:
|
|
raise Exception('unknown value {} while trying to cast {} to boolean'.format(value, obj))
|
|
return prop[value]
|
|
|
|
func = {IntOption: int, StrOption: str, PortOption: str,
|
|
DomainnameOption: str, EmailOption: str, URLOption: str,
|
|
IPOption: str, NetmaskOption: str, NetworkOption: str,
|
|
BroadcastOption: str, FilenameOption: str,
|
|
BoolOption: _convert_boolean}.get(obj, None)
|
|
if func is None:
|
|
return value
|
|
if isinstance(value, list):
|
|
return [func(val) for val in value]
|
|
else:
|
|
return func(value)
|
|
|
|
|
|
CONVERT_OPTION = {'number': dict(opttype=IntOption),
|
|
'choice': dict(opttype=ChoiceOption),
|
|
'string': dict(opttype=StrOption),
|
|
'password': dict(opttype=PasswordOption),
|
|
'mail': dict(opttype=EmailOption),
|
|
'boolean': dict(opttype=BoolOption),
|
|
'symlink': dict(opttype=SymLinkOption),
|
|
'filename': dict(opttype=FilenameOption),
|
|
'date': dict(opttype=DateOption),
|
|
'unix_user': dict(opttype=UsernameOption),
|
|
'ip': dict(opttype=IPOption, initkwargs={'allow_reserved': True}),
|
|
'local_ip': dict(opttype=IPOption, initkwargs={'private_only': True, 'warnings_only': True}),
|
|
'netmask': dict(opttype=NetmaskOption),
|
|
'network': dict(opttype=NetworkOption),
|
|
'broadcast': dict(opttype=BroadcastOption),
|
|
'netbios': dict(opttype=DomainnameOption, initkwargs={'type': 'netbios', 'warnings_only': True}),
|
|
'domain': dict(opttype=DomainnameOption, initkwargs={'type': 'domainname', 'allow_ip': True, 'allow_without_dot': True}),
|
|
'domain_strict': dict(opttype=DomainnameOption, initkwargs={'type': 'domainname', 'allow_ip': False}),
|
|
'hostname': dict(opttype=DomainnameOption, initkwargs={'type': 'hostname', 'allow_ip': True}),
|
|
'hostname_strict': dict(opttype=DomainnameOption, initkwargs={'type': 'hostname', 'allow_ip': False}),
|
|
'web_address': dict(opttype=URLOption, initkwargs={'allow_ip': True, 'allow_without_dot': True}),
|
|
'port': dict(opttype=PortOption, initkwargs={'allow_private': True}),
|
|
'mac': dict(opttype=MACOption),
|
|
'cidr': dict(opttype=IPOption, initkwargs={'cidr': True}),
|
|
'network_cidr': dict(opttype=NetworkOption, initkwargs={'cidr': True}),
|
|
}
|
|
|
|
|
|
class Elt(object):
|
|
def __init__(self, attrib):
|
|
self.attrib = attrib
|
|
|
|
|
|
class PopulateTiramisuObjects(object):
|
|
def __init__(self):
|
|
self.storage = ElementStorage()
|
|
self.booleans = []
|
|
self.force_store_values = set()
|
|
self.separators = {}
|
|
self.groups = {}
|
|
|
|
def parse_dtd(self, dtdfilename):
|
|
"""Loads the Creole DTD
|
|
|
|
:raises IOError: if the DTD is not found
|
|
|
|
:param dtdfilename: the full filename of the Creole DTD
|
|
"""
|
|
if not isfile(dtdfilename):
|
|
raise IOError(_("no such DTD file: {}").format(dtdfilename))
|
|
with open(dtdfilename, 'r') as dtdfd:
|
|
dtd = DTD(dtdfd)
|
|
for elt in dtd.iterelements():
|
|
if elt.name == 'variable':
|
|
for attr in elt.iterattributes():
|
|
if set(attr.itervalues()) == set(['True', 'False']):
|
|
self.booleans.append(attr.name)
|
|
|
|
def make_tiramisu_objects(self, xmlroot, creolefunc_file):
|
|
elt = Elt({'name': 'baseoption'})
|
|
if creolefunc_file is None:
|
|
self.eosfunc = None
|
|
else:
|
|
self.eosfunc = imp.load_source('eosfunc', creolefunc_file)
|
|
family = Family(elt, self.booleans, self.storage, self.eosfunc)
|
|
self.storage.add('.', family)
|
|
|
|
elts = {}
|
|
for elt in xmlroot:
|
|
elts.setdefault(elt.tag, []).append(elt)
|
|
list_elts = list(elts.keys())
|
|
if 'family' in list_elts:
|
|
list_elts.remove('family')
|
|
list_elts.insert(0, 'family')
|
|
for elt in list_elts:
|
|
xmlelts_ = elts[elt]
|
|
if elt == 'family':
|
|
xmlelts = []
|
|
actions = None
|
|
# `creole` family has to be loaded before any other family
|
|
# because `extra` family could use `creole` variables.
|
|
# `actions` family has to be loaded at the very end
|
|
# because it may use `creole` or `extra` variables
|
|
for xml in xmlelts_:
|
|
if xml.attrib['name'] == 'creole':
|
|
xmlelts.insert(0, xml)
|
|
elif xml.attrib['name'] == 'actions':
|
|
actions = xml
|
|
else:
|
|
xmlelts.append(xml)
|
|
if actions is not None:
|
|
xmlelts.append(actions)
|
|
else:
|
|
xmlelts = xmlelts_
|
|
for xmlelt in xmlelts:
|
|
if xmlelt.tag != 'family':
|
|
raise CreoleLoaderError(_('unknown tag {}').format(xmlelt.tag))
|
|
self._iter_family(xmlelt, family)
|
|
|
|
def _populate_variable(self, elt, subpath, is_follower, is_leader):
|
|
variable = Variable(elt, self.booleans, self.storage, is_follower, is_leader, self.eosfunc)
|
|
path = self._build_path(subpath, elt)
|
|
properties = variable.attrib.get('properties', [])
|
|
if 'force_store_value' in properties or "auto_freeze" in properties:
|
|
self.force_store_values.add(path)
|
|
self.storage.add(path, variable)
|
|
return variable
|
|
|
|
def _populate_family(self, elt, subpath):
|
|
if subpath is None:
|
|
force_icon = False
|
|
else:
|
|
force_icon = not subpath.startswith('containers') and not subpath.startswith('actions')
|
|
family = Family(elt, self.booleans, self.storage, self.eosfunc, force_icon)
|
|
path = self._build_path(subpath, elt)
|
|
self.storage.add(path, family)
|
|
return family
|
|
|
|
def _build_path(self, subpath, elt):
|
|
if subpath is None:
|
|
subpath = elt.attrib['name']
|
|
else:
|
|
subpath += '.' + elt.attrib['name']
|
|
return subpath
|
|
|
|
def _iter_leader(self, leader, subpath):
|
|
subpath = self._build_path(subpath, leader)
|
|
family = Family(leader, self.booleans, self.storage, self.eosfunc)
|
|
family.set_leader()
|
|
self.storage.add(subpath, family)
|
|
leader_name = None
|
|
for var in leader:
|
|
if var.tag == 'property':
|
|
self._parse_properties(family, var)
|
|
elif var.tag == 'variable':
|
|
if leader_name is None:
|
|
leader_name = var.attrib['name']
|
|
self.groups[leader_name] = []
|
|
else:
|
|
self.groups[leader_name].append(var.attrib['name'])
|
|
self._iter_family(var, family, subpath=subpath)
|
|
else:
|
|
raise CreoleLoaderError(_('unknown tag {}').format(var.tag))
|
|
return family
|
|
|
|
def _iter_family(self, child, family, subpath=None):
|
|
if child.tag not in ['family', 'variable', 'separators', 'leader', 'property']:
|
|
raise CreoleLoaderError(_('unknown tag {}').format(child.tag))
|
|
if child.tag == 'family':
|
|
old_family = family
|
|
family = self._populate_family(child, subpath)
|
|
if old_family is not None:
|
|
old_family.add(family)
|
|
if len(child) != 0:
|
|
subpath = self._build_path(subpath, child)
|
|
for c in child:
|
|
self._iter_family(c, family, subpath=subpath)
|
|
elif child.tag == 'leader':
|
|
leader = self._iter_leader(child, subpath)
|
|
family.add(leader)
|
|
elif child.tag == 'separators':
|
|
self._parse_separators(child)
|
|
elif child.tag == 'variable':
|
|
if family is None:
|
|
raise CreoleLoaderError(_('variable without family'))
|
|
|
|
is_follower = False
|
|
is_leader = False
|
|
if family.is_leader:
|
|
if child.attrib['name'] != family.attrib['name']:
|
|
is_follower = True
|
|
else:
|
|
is_leader = True
|
|
variable = self._populate_variable(child, subpath, is_follower, is_leader)
|
|
family.add(variable)
|
|
elif child.tag == 'property':
|
|
self._parse_properties(family, child)
|
|
else:
|
|
raise Exception('unknown tag {}'.format(child.tag))
|
|
|
|
def _parse_properties(self, family, child):
|
|
if child.get('type') == 'calculation':
|
|
kwargs = {'condition': child.attrib['source'],
|
|
'expected': ParamValue(child.attrib.get('expected'))}
|
|
if child.attrib['inverse'] == 'True':
|
|
kwargs['reverse_condition'] = ParamValue(True)
|
|
family.attrib['properties'].append((ParamValue(child.text), kwargs))
|
|
else:
|
|
family.attrib['properties'].append(child.text)
|
|
|
|
def _parse_separators(self, separators):
|
|
for separator in separators:
|
|
elt = self.storage.get(separator.attrib['name'])
|
|
never_hidden = separator.attrib.get('never_hidden')
|
|
if never_hidden == 'True':
|
|
never_hidden = True
|
|
else:
|
|
never_hidden = None
|
|
info = (separator.text, never_hidden)
|
|
self.separators[separator.attrib['name']] = info
|
|
elt.add_information('separator', info)
|
|
|
|
|
|
class ElementStorage:
|
|
def __init__(self):
|
|
self.paths = {}
|
|
|
|
def add(self, path, elt):
|
|
if path in self.paths:
|
|
raise CreoleLoaderError(_('path already loaded {}').format(path))
|
|
self.paths[path] = elt
|
|
|
|
def add_information(self, path, name, information):
|
|
elt = self.get(path)
|
|
elt.add_information(name, information)
|
|
|
|
def get(self, path):
|
|
if path not in self.paths:
|
|
raise CreoleLoaderError(_('there is no element for path {}').format(path))
|
|
return self.paths[path]
|
|
|
|
|
|
class Common:
|
|
def build_properties(self):
|
|
for index, prop in enumerate(self.attrib['properties']):
|
|
if isinstance(prop, tuple):
|
|
action, kwargs = prop
|
|
kwargs['condition'] = ParamOption(self.storage.get(kwargs['condition']).get(), todict=True)
|
|
prop = Calculation(calc_value,
|
|
Params(action,
|
|
kwargs=kwargs))
|
|
self.attrib['properties'][index] = prop
|
|
if self.attrib['properties']:
|
|
self.attrib['properties'] = tuple(self.attrib['properties'])
|
|
else:
|
|
del self.attrib['properties']
|
|
|
|
|
|
class Variable(Common):
|
|
def __init__(self, elt, booleans, storage, is_follower, is_leader, eosfunc):
|
|
self.option = None
|
|
self.informations = {}
|
|
self.attrib = {}
|
|
self.attrib['properties'] = []
|
|
self.attrib['validators'] = []
|
|
self.eosfunc = eosfunc
|
|
self.storage = storage
|
|
is_submulti = False
|
|
for key, value in elt.attrib.items():
|
|
if key in booleans:
|
|
if value == 'True':
|
|
value = True
|
|
elif value == 'False':
|
|
value = False
|
|
elif key == 'multi' and value == 'submulti':
|
|
is_submulti = True
|
|
value = submulti
|
|
else:
|
|
raise CreoleLoaderError(_('unknown value {} for {}').format(value, key))
|
|
if key in ['help', 'test']:
|
|
self.add_information(key, value)
|
|
elif key == 'type':
|
|
pass
|
|
else:
|
|
self.attrib[key] = value
|
|
convert_option = CONVERT_OPTION[elt.attrib['type']]
|
|
self.object_type = convert_option['opttype']
|
|
if elt.attrib['type'] == 'choice':
|
|
if self.attrib.get('choice'):
|
|
self.attrib['values'] = getattr(self.eosfunc, self.attrib.get('choice'))
|
|
else:
|
|
self.attrib['values'] = []
|
|
for child in elt:
|
|
if child.tag == 'choice':
|
|
value = child.text
|
|
if 'type' in child.attrib and child.attrib['type'] == 'number':
|
|
value = int(value)
|
|
if value is None:
|
|
value = u''
|
|
self.attrib['values'].append(value)
|
|
self.attrib['values'] = tuple(self.attrib['values'])
|
|
for child in elt:
|
|
if child.tag == 'property':
|
|
if child.get('type') == 'calculation':
|
|
kwargs = {'condition': child.attrib['source'],
|
|
'expected': ParamValue(child.attrib.get('expected'))}
|
|
if child.attrib['inverse'] == 'True':
|
|
kwargs['reverse_condition'] = ParamValue(True)
|
|
self.attrib['properties'].append((ParamValue(child.text), kwargs))
|
|
else:
|
|
self.attrib['properties'].append(child.text)
|
|
elif child.tag == 'value':
|
|
if child.attrib.get('type') == 'calculation':
|
|
if child.text is not None and child.text.strip():
|
|
self.attrib['default'] = (child.text.strip(),)
|
|
else:
|
|
params = []
|
|
for param in child:
|
|
params.append(self.parse_param(param))
|
|
self.attrib['default'] = (child.attrib['name'], params, False)
|
|
else:
|
|
if "type" in child.attrib:
|
|
type_ = CONVERT_OPTION[child.attrib['type']]['opttype']
|
|
else:
|
|
type_ = self.object_type
|
|
if self.attrib['multi'] is True and not is_follower:
|
|
if 'default' not in self.attrib:
|
|
self.attrib['default'] = []
|
|
value = convert_tiramisu_value(child.text, type_)
|
|
self.attrib['default'].append(value)
|
|
if 'default_multi' not in self.attrib and not is_leader:
|
|
self.attrib['default_multi'] = value
|
|
elif self.attrib['multi'] == submulti:
|
|
if 'default' not in self.attrib:
|
|
self.attrib['default'] = []
|
|
value = convert_tiramisu_value(child.text, type_)
|
|
if not isinstance(value, list) and not is_follower:
|
|
value = [value]
|
|
self.attrib['default'].append(value)
|
|
if 'default_multi' not in self.attrib and not is_leader:
|
|
self.attrib['default_multi'] = value
|
|
else:
|
|
if 'default' in self.attrib:
|
|
raise CreoleLoaderError(_('default value already set for {}'
|
|
'').format(self.attrib['path']))
|
|
value = convert_tiramisu_value(child.text, type_)
|
|
if value is None: # and (elt.attrib['type'] != 'choice' or value not in self.attrib['values']):
|
|
value = u''
|
|
if is_follower:
|
|
self.attrib['default_multi'] = value
|
|
else:
|
|
self.attrib['default'] = value
|
|
elif child.tag == 'choice':
|
|
# already load
|
|
pass
|
|
elif child.tag == 'check':
|
|
params = []
|
|
for param in child:
|
|
params.append(self.parse_param(param))
|
|
#check.params = params
|
|
self.attrib['validators'].append((child.attrib['name'], params, child.attrib['warnings_only']))
|
|
else:
|
|
raise Exception('unknown tag {}'.format(child.tag))
|
|
if 'initkwargs' in convert_option:
|
|
self.attrib.update(convert_option['initkwargs'])
|
|
if elt.attrib['type'] == 'symlink':
|
|
del self.attrib['properties']
|
|
del self.attrib['multi']
|
|
self.attrib['opt'] = storage.get(self.attrib['opt'])
|
|
|
|
def parse_param(self, param):
|
|
name = param.attrib.get('name', '')
|
|
if param.attrib['type'] == 'string':
|
|
value = param.text
|
|
elif param.attrib['type'] == 'variable':
|
|
transitive = param.attrib.get('transitive', 'False')
|
|
if transitive == 'True':
|
|
transitive = True
|
|
elif transitive == 'False':
|
|
transitive = False
|
|
else:
|
|
raise CreoleLoaderError(_('unknown transitive boolean {}').format(transitive))
|
|
value = [param.text, transitive]
|
|
elif param.attrib['type'] == 'number':
|
|
value = int(param.text)
|
|
else:
|
|
raise CreoleLoaderError(_('unknown param type {}').format(param.attrib['type']))
|
|
return(name, value)
|
|
|
|
def add_information(self, key, value):
|
|
if key in self.informations:
|
|
raise CreoleLoaderError(_('key already exists in information {}').format(key))
|
|
self.informations[key] = value
|
|
|
|
def build_calculator(self, key):
|
|
if key in self.attrib:
|
|
values = self.attrib[key]
|
|
if isinstance(values, list):
|
|
is_list = True
|
|
else:
|
|
is_list = False
|
|
values = [values]
|
|
ret = []
|
|
for value in values:
|
|
if isinstance(value, tuple):
|
|
if key == 'validators':
|
|
args = [ParamSelfOption()]
|
|
else:
|
|
args = []
|
|
kwargs = {}
|
|
if len(value) == 3:
|
|
for param in value[1]:
|
|
if isinstance(param[1], list):
|
|
if value[0] in FUNC_TO_DICT:
|
|
param_value = ParamOption(self.storage.get(param[1][0]).get(), notraisepropertyerror=param[1][1], todict=True)
|
|
else:
|
|
param_value = ParamOption(self.storage.get(param[1][0]).get(), notraisepropertyerror=param[1][1])
|
|
else:
|
|
param_value = ParamValue(param[1])
|
|
if not param[0]:
|
|
args.append(param_value)
|
|
else:
|
|
kwargs[param[0]] = param_value
|
|
|
|
ret.append(Calculation(getattr(self.eosfunc, value[0]),
|
|
Params(tuple(args),
|
|
kwargs=kwargs)))
|
|
else:
|
|
ret.append(value)
|
|
if not is_list:
|
|
self.attrib[key] = ret[0]
|
|
else:
|
|
self.attrib[key] = ret
|
|
|
|
|
|
def get(self):
|
|
if self.option is None:
|
|
if self.object_type is SymLinkOption:
|
|
self.attrib['opt'] = self.attrib['opt'].get()
|
|
else:
|
|
self.build_properties()
|
|
self.build_calculator('default')
|
|
self.build_calculator('validators')
|
|
if not self.attrib['validators']:
|
|
del self.attrib['validators']
|
|
try:
|
|
option = self.object_type(**self.attrib)
|
|
except Exception as err:
|
|
import traceback
|
|
traceback.print_exc()
|
|
name = self.attrib['name']
|
|
raise CreoleLoaderError(_('cannot create option {}: {}').format(name, err))
|
|
for key, value in self.informations.items():
|
|
option.impl_set_information(key, value)
|
|
self.option = option
|
|
return self.option
|
|
|
|
|
|
class Family(Common):
|
|
def __init__(self, elt, booleans, storage, eosfunc, force_icon=False):
|
|
self.option = None
|
|
self.attrib = {}
|
|
self.is_leader = False
|
|
if force_icon:
|
|
self.informations = {'icon': None}
|
|
else:
|
|
self.informations = {}
|
|
self.children = []
|
|
self.storage = storage
|
|
self.eosfunc = eosfunc
|
|
self.attrib['properties'] = []
|
|
for key, value in elt.attrib.items():
|
|
if key in booleans:
|
|
if value == 'True':
|
|
value = True
|
|
elif value == 'False':
|
|
value = False
|
|
else:
|
|
raise CreoleLoaderError(_('unknown value {} for {}').format(value, key))
|
|
if key == 'icon':
|
|
self.add_information('icon', value)
|
|
continue
|
|
elif key == 'hidden':
|
|
if value:
|
|
self.attrib['properties'].append(key)
|
|
elif key == 'mode':
|
|
self.attrib['properties'].append(value)
|
|
elif key == 'help':
|
|
self.add_information(key, value)
|
|
elif key == 'type':
|
|
pass
|
|
else:
|
|
self.attrib[key] = value
|
|
if 'doc' not in self.attrib:
|
|
self.attrib['doc'] = u''
|
|
|
|
def add(self, child):
|
|
self.children.append(child)
|
|
|
|
def add_information(self, key, value):
|
|
if key in self.informations and not (key == 'icon' and self.informations[key] is None):
|
|
raise CreoleLoaderError(_('key already exists in information {}').format(key))
|
|
self.informations[key] = value
|
|
|
|
def set_leader(self):
|
|
self.is_leader = True
|
|
|
|
def get(self):
|
|
if self.option is None:
|
|
self.attrib['children'] = []
|
|
for child in self.children:
|
|
self.attrib['children'].append(child.get())
|
|
self.build_properties()
|
|
try:
|
|
if 'dynamic' in self.attrib:
|
|
dynamic = self.storage.get(self.attrib['dynamic']).get()
|
|
del self.attrib['dynamic']
|
|
self.attrib['suffixes'] = Calculation(self.eosfunc.calc_value,
|
|
Params((ParamOption(dynamic),)))
|
|
option = ConvertDynOptionDescription(**self.attrib)
|
|
elif not self.is_leader:
|
|
option = OptionDescription(**self.attrib)
|
|
else:
|
|
option = Leadership(**self.attrib)
|
|
except Exception as err:
|
|
raise CreoleLoaderError(_('cannot create optiondescription {}: {}').format(self.attrib['name'], err))
|
|
for key, value in self.informations.items():
|
|
option.impl_set_information(key, value)
|
|
self.option = option
|
|
#if self.is_leader:
|
|
# self.option.impl_set_group_type(groups.leader)
|
|
|
|
return self.option
|
|
|
|
|
|
def load(xmlroot: str,
|
|
dtd_path: str,
|
|
funcs_path: str):
|
|
tiramisu_objects = PopulateTiramisuObjects()
|
|
tiramisu_objects.parse_dtd(dtd_path)
|
|
tiramisu_objects.make_tiramisu_objects(xmlroot,
|
|
funcs_path)
|
|
return tiramisu_objects.storage.paths['.'].get()
|