118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
from lxml import etree
|
|
from os.path import isfile, join, isdir
|
|
from pytest import fixture, mark
|
|
from os import listdir, mkdir
|
|
from json import dump, load, dumps, loads
|
|
|
|
from tiramisu import Config
|
|
from tiramisu.error import PropertiesOptionError
|
|
|
|
|
|
dico_dirs = 'tests/dictionaries'
|
|
|
|
|
|
test_ok = set()
|
|
|
|
for test in listdir(dico_dirs):
|
|
if isdir(join(dico_dirs, test)):
|
|
if isdir(join(dico_dirs, test, 'tiramisu')):
|
|
test_ok.add(test)
|
|
|
|
debug = False
|
|
#debug = True
|
|
excludes = set([])
|
|
#excludes = set(['01base_file_utfchar'])
|
|
test_ok -= excludes
|
|
#test_ok = ['10check_valid_ipnetmask']
|
|
|
|
|
|
test_ok = list(test_ok)
|
|
test_ok.sort()
|
|
|
|
|
|
@fixture(scope="module", params=test_ok)
|
|
def test_dir(request):
|
|
return request.param
|
|
|
|
|
|
async def launch_flattener(test_dir):
|
|
makedict_dir = join(test_dir, 'makedict')
|
|
makedict_file = join(makedict_dir, 'base.json')
|
|
makedict_before = join(makedict_dir, 'before.json')
|
|
makedict_after = join(makedict_dir, 'after.json')
|
|
|
|
modulepath = test_dir.replace('/', '.') + '.tiramisu.base'
|
|
mod = __import__(modulepath)
|
|
for token in modulepath.split(".")[1:]:
|
|
mod = getattr(mod, token)
|
|
config = await Config(mod.option_0)
|
|
# change default rights
|
|
ro_origin = await config.property.getdefault('read_only', 'append')
|
|
ro_append = frozenset(ro_origin - {'force_store_value'})
|
|
rw_origin = await config.property.getdefault('read_write', 'append')
|
|
rw_append = frozenset(rw_origin - {'force_store_value'})
|
|
await config.property.setdefault(ro_append, 'read_only', 'append')
|
|
await config.property.setdefault(rw_append, 'read_write', 'append')
|
|
|
|
await config.information.set('test_information', 'value')
|
|
await config.property.read_only()
|
|
await config.property.pop('mandatory')
|
|
await config.information.set('info', 'value')
|
|
config_dict = await config.value.dict(leader_to_list=True)
|
|
if not isdir(makedict_dir):
|
|
mkdir(makedict_dir)
|
|
if not isfile(makedict_file) or debug:
|
|
with open(makedict_file, 'w') as fh:
|
|
dump(config_dict, fh, indent=4)
|
|
fh.write('\n')
|
|
if not isfile(makedict_file):
|
|
raise Exception('dict is not empty')
|
|
with open(makedict_file, 'r') as fh:
|
|
assert load(fh) == loads(dumps(config_dict))
|
|
#
|
|
await value_owner(makedict_before, config)
|
|
# deploy
|
|
ro = await config.property.getdefault('read_only', 'append')
|
|
ro = frozenset(list(ro) + ['force_store_value'])
|
|
await config.property.setdefault(ro, 'read_only', 'append')
|
|
rw = await config.property.getdefault('read_write', 'append')
|
|
rw = frozenset(list(rw) + ['force_store_value'])
|
|
await config.property.setdefault(rw, 'read_write', 'append')
|
|
await config.property.add('force_store_value')
|
|
#
|
|
await value_owner(makedict_after, config)
|
|
|
|
async def value_owner(makedict_value_owner, config):
|
|
ret = {}
|
|
for key in await config.option.list(recursive=True):
|
|
path = await key.option.path()
|
|
if await key.option.isfollower():
|
|
value = []
|
|
owner = []
|
|
for idx in range(0, await key.value.len()):
|
|
try:
|
|
option = config.option(path, idx)
|
|
value.append(await option.value.get())
|
|
owner.append(await option.owner.get())
|
|
except PropertiesOptionError as err:
|
|
value.append(str(err))
|
|
owner.append('error')
|
|
else:
|
|
value = await key.value.get()
|
|
owner = await key.owner.get()
|
|
ret[path] = {'owner': owner,
|
|
'value': value,
|
|
}
|
|
if not isfile(makedict_value_owner) or debug:
|
|
with open(makedict_value_owner, 'w') as fh:
|
|
dump(ret, fh, indent=4)
|
|
fh.write('\n')
|
|
with open(makedict_value_owner, 'r') as fh:
|
|
assert load(fh) == loads(dumps(ret))
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_dictionary(test_dir):
|
|
test_dir = join(dico_dirs, test_dir)
|
|
await launch_flattener(test_dir)
|