515 lines
19 KiB
Python
515 lines
19 KiB
Python
"""loader
|
|
flattened XML specific
|
|
"""
|
|
from os.path import isfile
|
|
from lxml.etree import DTD
|
|
|
|
from .config import dtdfilename, variable_namespace
|
|
from .i18n import _
|
|
from .error import LoaderError
|
|
from .annotator import ERASED_ATTRIBUTES
|
|
|
|
|
|
FUNC_TO_DICT = ['valid_not_equal']
|
|
FORCE_INFORMATIONS = ['help', 'test', 'separator', 'manage']
|
|
ATTRIBUTES_ORDER = ('name', 'doc', 'default', 'multi')
|
|
|
|
|
|
CONVERT_OPTION = {'number': dict(opttype="IntOption", func=int),
|
|
'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': False}),
|
|
'hostname': 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 TiramisuReflector:
|
|
def __init__(self,
|
|
xmlroot,
|
|
funcs_path,
|
|
):
|
|
self.storage = ElementStorage()
|
|
self.storage.text = ["from tiramisu import *",
|
|
"from rougail.tiramisu import ConvertDynOptionDescription",
|
|
"import imp",
|
|
f"func = imp.load_source('func', '{funcs_path}')",
|
|
]
|
|
self.make_tiramisu_objects(xmlroot)
|
|
# parse object
|
|
self.storage.get('.').get()
|
|
|
|
def make_tiramisu_objects(self,
|
|
xmlroot,
|
|
):
|
|
family = self.get_root_family()
|
|
for xmlelt in self.reorder_family(xmlroot):
|
|
self.iter_family(xmlelt,
|
|
family,
|
|
None,
|
|
)
|
|
|
|
def get_root_family(self):
|
|
family = Family(BaseElt('baseoption',
|
|
'baseoption',
|
|
),
|
|
self.storage,
|
|
False,
|
|
'.',
|
|
)
|
|
return family
|
|
|
|
def reorder_family(self, xmlroot):
|
|
# variable_namespace family has to be loaded before any other family
|
|
# because `extra` family could use `variable_namespace` variables.
|
|
if hasattr(xmlroot, 'variables'):
|
|
if variable_namespace in xmlroot.variables:
|
|
yield xmlroot.variables[variable_namespace]
|
|
for xmlelt, value in xmlroot.variables.items():
|
|
if xmlelt != variable_namespace:
|
|
yield value
|
|
if hasattr(xmlroot, 'services'):
|
|
yield xmlroot.services
|
|
|
|
def get_attributes(self, space): # pylint: disable=R0201
|
|
for attr in dir(space):
|
|
if not attr.startswith('_') and attr not in ERASED_ATTRIBUTES:
|
|
yield attr
|
|
|
|
def get_children(self,
|
|
space,
|
|
):
|
|
for tag in self.get_attributes(space):
|
|
children = getattr(space, tag)
|
|
if children.__class__.__name__ == 'Family':
|
|
children = [children]
|
|
if isinstance(children, dict):
|
|
children = list(children.values())
|
|
if isinstance(children, list):
|
|
yield tag, children
|
|
|
|
def iter_family(self,
|
|
child,
|
|
family,
|
|
subpath,
|
|
):
|
|
tag = child.__class__.__name__
|
|
if tag == 'Variable':
|
|
function = self.populate_variable
|
|
elif tag == 'Property':
|
|
# property already imported with family
|
|
return
|
|
else:
|
|
function = self.populate_family
|
|
#else:
|
|
# raise Exception('unknown tag {}'.format(child.tag))
|
|
function(family,
|
|
child,
|
|
subpath,
|
|
)
|
|
|
|
def populate_family(self,
|
|
parent_family,
|
|
elt,
|
|
subpath,
|
|
):
|
|
path = self.build_path(subpath,
|
|
elt,
|
|
)
|
|
tag = elt.__class__.__name__
|
|
family = Family(elt,
|
|
self.storage,
|
|
tag == 'Leadership',
|
|
path,
|
|
)
|
|
parent_family.add(family)
|
|
for tag, children in self.get_children(elt):
|
|
for child in children:
|
|
self.iter_family(child,
|
|
family,
|
|
path,
|
|
)
|
|
|
|
def populate_variable(self,
|
|
family,
|
|
elt,
|
|
subpath,
|
|
):
|
|
is_follower = False
|
|
is_leader = False
|
|
if family.is_leader:
|
|
if elt.name != family.elt.name:
|
|
is_follower = True
|
|
else:
|
|
is_leader = True
|
|
family.add(Variable(elt,
|
|
self.storage,
|
|
is_follower,
|
|
is_leader,
|
|
self.build_path(subpath,
|
|
elt,
|
|
)
|
|
))
|
|
|
|
def build_path(self,
|
|
subpath,
|
|
elt,
|
|
):
|
|
if subpath is None:
|
|
return elt.name
|
|
return subpath + '.' + elt.name
|
|
|
|
def get_text(self):
|
|
return '\n'.join(self.storage.get('.').get_text())
|
|
|
|
|
|
class BaseElt:
|
|
def __init__(self,
|
|
name,
|
|
doc,
|
|
):
|
|
self.name = name
|
|
self.doc = doc
|
|
|
|
def __iter__(self):
|
|
return iter([])
|
|
|
|
|
|
class ElementStorage:
|
|
def __init__(self,
|
|
):
|
|
self.paths = {}
|
|
self.text = []
|
|
self.index = 0
|
|
|
|
def add(self, path, elt):
|
|
self.paths[path] = (elt, self.index)
|
|
self.index += 1
|
|
|
|
def get(self, path):
|
|
if path not in self.paths or self.paths[path][0] is None:
|
|
raise LoaderError(_('there is no element for path {}').format(path))
|
|
return self.paths[path][0]
|
|
|
|
def get_name(self, path):
|
|
if path not in self.paths:
|
|
raise LoaderError(_('there is no element for index {}').format(path))
|
|
return f'option_{self.paths[path][1]}'
|
|
|
|
|
|
class Common:
|
|
def __init__(self,
|
|
elt,
|
|
storage,
|
|
is_leader,
|
|
path,
|
|
):
|
|
self.option_name = None
|
|
self.path = path
|
|
self.attrib = {}
|
|
self.informations = {}
|
|
self.storage = storage
|
|
self.is_leader = is_leader
|
|
self.storage.add(self.path, self)
|
|
|
|
def populate_properties(self, child):
|
|
if child.type == 'calculation':
|
|
action = f"ParamValue('{child.name}')"
|
|
option_name = self.storage.get(child.source).get()
|
|
kwargs = f"'condition': ParamOption({option_name}, todict=True), 'expected': ParamValue('{child.expected}')"
|
|
if child.inverse:
|
|
kwargs += ", 'reverse_condition': ParamValue(True)"
|
|
prop = 'Calculation(calc_value, Params(' + action + ', kwargs={' + kwargs + '}))'
|
|
else:
|
|
prop = "'" + child.text + "'"
|
|
if self.attrib['properties']:
|
|
self.attrib['properties'] += ', '
|
|
self.attrib['properties'] += prop
|
|
|
|
def get_attrib(self, attrib):
|
|
ret_list = []
|
|
for key, value in self.attrib.items():
|
|
if value is None:
|
|
continue
|
|
if key == 'properties':
|
|
if not self.attrib[key]:
|
|
continue
|
|
value = "frozenset({" + self.attrib[key] + "})"
|
|
elif key in ['default', 'multi', 'suffixes', 'validators', 'values']:
|
|
value = self.attrib[key]
|
|
elif isinstance(value, str) and key != 'opt' and not value.startswith('['):
|
|
value = "'" + value.replace("'", "\\\'") + "'"
|
|
ret_list.append(f'{key}={value}')
|
|
return ', '.join(ret_list)
|
|
|
|
def populate_informations(self):
|
|
for key, value in self.informations.items():
|
|
if isinstance(value, str):
|
|
value = '"' + value.replace('"', '\"') + '"'
|
|
self.storage.text.append(f'{self.option_name}.impl_set_information("{key}", {value})')
|
|
|
|
def get_text(self,
|
|
):
|
|
return self.storage.text
|
|
|
|
def get_attributes(self, space): # pylint: disable=R0201
|
|
attributes = dir(space)
|
|
for attr in ATTRIBUTES_ORDER:
|
|
if attr in attributes:
|
|
yield attr
|
|
for attr in dir(space):
|
|
if attr not in ATTRIBUTES_ORDER:
|
|
if not attr.startswith('_') and attr not in ERASED_ATTRIBUTES:
|
|
value = getattr(space, attr)
|
|
if not isinstance(value, (list, dict)) and not value.__class__.__name__ == 'Family':
|
|
yield attr
|
|
|
|
def get_children(self,
|
|
space,
|
|
):
|
|
for attr in dir(space):
|
|
if not attr.startswith('_') and attr not in ERASED_ATTRIBUTES:
|
|
value = getattr(space, attr)
|
|
if isinstance(value, (list, dict)):
|
|
children = getattr(space, attr)
|
|
if children.__class__.__name__ == 'Family':
|
|
children = [children]
|
|
if isinstance(children, dict):
|
|
children = list(children.values())
|
|
if children and isinstance(children, list):
|
|
yield attr, children
|
|
|
|
|
|
class Variable(Common):
|
|
def __init__(self,
|
|
elt,
|
|
storage,
|
|
is_follower,
|
|
is_leader,
|
|
path,
|
|
):
|
|
super().__init__(elt,
|
|
storage,
|
|
is_leader,
|
|
path,
|
|
)
|
|
self.is_follower = is_follower
|
|
convert_option = CONVERT_OPTION[elt.type]
|
|
del elt.type
|
|
self.object_type = convert_option['opttype']
|
|
self.attrib.update(convert_option.get('initkwargs', {}))
|
|
if self.object_type != 'SymLinkOption':
|
|
self.attrib['properties'] = []
|
|
self.attrib['validators'] = []
|
|
self.elt = elt
|
|
|
|
def get(self):
|
|
if self.option_name is None:
|
|
self.populate_attrib()
|
|
if self.object_type == 'SymLinkOption':
|
|
self.attrib['opt'] = self.storage.get(self.attrib['opt']).get()
|
|
else:
|
|
self.parse_children()
|
|
attrib = self.get_attrib(self.attrib)
|
|
self.option_name = self.storage.get_name(self.path)
|
|
self.storage.text.append(f'{self.option_name} = {self.object_type}({attrib})')
|
|
self.populate_informations()
|
|
return self.option_name
|
|
|
|
def populate_attrib(self):
|
|
for key in self.get_attributes(self.elt):
|
|
value = getattr(self.elt, key)
|
|
if key in FORCE_INFORMATIONS:
|
|
if key == 'test':
|
|
value = value.split(',')
|
|
if self.object_type == 'IntOption':
|
|
value = [int(v) for v in value]
|
|
self.informations[key] = value
|
|
else:
|
|
self.attrib[key] = value
|
|
|
|
def parse_children(self):
|
|
if not 'default' in self.attrib or self.attrib['multi']:
|
|
self.attrib['default'] = []
|
|
if self.attrib['multi'] == 'submulti' and self.is_follower:
|
|
self.attrib['default_multi'] = []
|
|
choices = []
|
|
if 'properties' in self.attrib:
|
|
if self.attrib['properties']:
|
|
self.attrib['properties'] = "'" + "', '".join(sorted(list(self.attrib['properties']))) + "'"
|
|
else:
|
|
self.attrib['properties'] = ''
|
|
for tag, children in self.get_children(self.elt):
|
|
for child in children:
|
|
if tag == 'property':
|
|
self.populate_properties(child)
|
|
elif tag == 'value':
|
|
if child.type == 'calculation':
|
|
self.attrib['default'] = self.calculation_value(child, [])
|
|
else:
|
|
self.populate_value(child)
|
|
elif tag == 'check':
|
|
self.attrib['validators'].append(self.calculation_value(child, ['ParamSelfOption()']))
|
|
elif tag == 'choice':
|
|
if child.type == 'calculation':
|
|
value = self.storage.get(child.name).get()
|
|
choices = f"Calculation(func.calc_value, Params((ParamOption({value}))))"
|
|
else:
|
|
choices.append(child.name)
|
|
if choices:
|
|
if isinstance(choices, list):
|
|
self.attrib['values'] = str(tuple(choices))
|
|
else:
|
|
self.attrib['values'] = choices
|
|
if self.attrib['default'] == []:
|
|
del self.attrib['default']
|
|
elif not self.attrib['multi'] and isinstance(self.attrib['default'], list):
|
|
self.attrib['default'] = self.attrib['default'][-1]
|
|
if self.attrib['validators'] == []:
|
|
del self.attrib['validators']
|
|
else:
|
|
self.attrib['validators'] = '[' + ', '.join(self.attrib['validators']) + ']'
|
|
|
|
def calculation_value(self, child, args):
|
|
kwargs = []
|
|
if hasattr(child, 'name'):
|
|
# has parameters
|
|
function = child.name
|
|
if hasattr(child, 'param'):
|
|
for param in child.param:
|
|
value = self.populate_param(param)
|
|
if not hasattr(param, 'name'):
|
|
args.append(str(value))
|
|
else:
|
|
kwargs.append(f"'{param.name}': " + value)
|
|
else:
|
|
# function without any parameter
|
|
function = child.text.strip()
|
|
ret = f"Calculation(func.{function}, Params((" + ', '.join(args) + "), kwargs=" + "{" + ', '.join(kwargs) + "})"
|
|
if hasattr(child, 'warnings_only'):
|
|
ret += f', warnings_only={child.warnings_only}'
|
|
return ret + ')'
|
|
|
|
def populate_param(self,
|
|
param,
|
|
):
|
|
if param.type == 'string':
|
|
return f'ParamValue("{param.text}")'
|
|
elif param.type == 'number':
|
|
return f'ParamValue({param.text})'
|
|
elif param.type == 'variable':
|
|
value = {'option': param.text,
|
|
'notraisepropertyerror': param.notraisepropertyerror,
|
|
'todict': param.text in FUNC_TO_DICT,
|
|
}
|
|
if hasattr(param, 'suffix'):
|
|
value['suffix'] = param.suffix
|
|
return self.build_param(value)
|
|
elif param.type == 'information':
|
|
return f'ParamInformation("{param.text}", None)'
|
|
elif param.type == 'suffix':
|
|
return f'ParamSuffix()'
|
|
raise LoaderError(_('unknown param type {}').format(param.type))
|
|
|
|
def populate_value(self,
|
|
child,
|
|
):
|
|
value = child.name
|
|
if self.attrib['multi'] == 'submulti':
|
|
self.attrib['default_multi'].append(value)
|
|
elif self.is_follower:
|
|
self.attrib['default_multi'] = value
|
|
elif self.attrib['multi']:
|
|
self.attrib['default'].append(value)
|
|
if not self.is_leader:
|
|
self.attrib['default_multi'] = value
|
|
elif isinstance(value, int):
|
|
self.attrib['default'].append(value)
|
|
else:
|
|
self.attrib['default'].append("'" + value + "'")
|
|
|
|
def build_param(self,
|
|
param,
|
|
):
|
|
option_name = self.storage.get(param['option']).get()
|
|
if 'suffix' in param:
|
|
family = '.'.join(param['option'].split('.')[:-1])
|
|
family_option = self.storage.get_name(family)
|
|
return f"ParamDynOption({option_name}, '{param['suffix']}', {family_option}, notraisepropertyerror={param['notraisepropertyerror']}, todict={param['todict']})"
|
|
return f"ParamOption({option_name}, notraisepropertyerror={param['notraisepropertyerror']}, todict={param['todict']})"
|
|
|
|
|
|
class Family(Common):
|
|
def __init__(self,
|
|
elt,
|
|
storage,
|
|
is_leader,
|
|
path,
|
|
):
|
|
super().__init__(elt,
|
|
storage,
|
|
is_leader,
|
|
path,
|
|
)
|
|
self.children = []
|
|
self.elt = elt
|
|
|
|
def add(self, child):
|
|
self.children.append(child)
|
|
|
|
def get(self):
|
|
if not self.option_name:
|
|
self.populate_attrib()
|
|
self.parse_children()
|
|
self.option_name = self.storage.get_name(self.path)
|
|
object_name = self.get_object_name()
|
|
attrib = self.get_attrib(self.attrib) + ', children=[' + ', '.join([child.get() for child in self.children]) + ']'
|
|
self.storage.text.append(f'{self.option_name} = {object_name}({attrib})')
|
|
self.populate_informations()
|
|
return self.option_name
|
|
|
|
def populate_attrib(self):
|
|
for key in self.get_attributes(self.elt):
|
|
value = getattr(self.elt, key)
|
|
if key in FORCE_INFORMATIONS:
|
|
self.informations[key] = value
|
|
elif key == 'dynamic':
|
|
dynamic = self.storage.get(value).get()
|
|
self.attrib['suffixes'] = f"Calculation(func.calc_value, Params((ParamOption({dynamic}))))"
|
|
else:
|
|
self.attrib[key] = value
|
|
|
|
def parse_children(self):
|
|
if 'properties' in self.attrib:
|
|
self.attrib['properties'] = "'" + "', '".join(sorted(list(self.attrib['properties']))) + "'"
|
|
if hasattr(self.elt, 'property'):
|
|
#self.attrib['properties'] = ''
|
|
for child in self.elt.property:
|
|
self.populate_properties(child)
|
|
if not self.attrib['properties']:
|
|
del self.attrib['properties']
|
|
|
|
def get_object_name(self):
|
|
if 'suffixes' in self.attrib:
|
|
return 'ConvertDynOptionDescription'
|
|
elif not self.is_leader:
|
|
return 'OptionDescription'
|
|
return 'Leadership'
|