rougail/tests/test_2_makedict.py

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)