dataset/tests/test_adictionaries.py

from pytest import fixture, mark
from os import makedirs, listdir, link
from os.path import isfile, isdir, join, abspath, basename
from yaml import load, SafeLoader
from json import dump
from shutil import rmtree
from traceback import print_exc

from tiramisu import Config
from rougail import RougailConvert, RougailConfig


FUNCTIONS = b"""from tiramisu import valid_network_netmask, valid_ip_netmask, valid_broadcast, valid_in_network, valid_not_equal as valid_differ, valid_not_equal, calc_value
# =============================================================
# fork of risotto-setting/src/risotto_setting/config/config.py
def get_password(**kwargs):
    return 'password'


def get_ip(**kwargs):
    return '1.1.1.1'


def get_chain(**kwargs):
    return 'chain'


def get_certificate(**kwargs):
    return 'certificate'


def get_private_key(**kwargs):
    return 'private_key'


def get_linked_configuration(**kwargs):
    if 'test' in kwargs and kwargs['test']:
        return kwargs['test'][0]
    if kwargs.get('multi', False) is True:
        return ['configuration']
    return 'configuration'


def get_certificates(**kwargs):
    return ["XXXX", "YYYY", "ZZZZ"]


def zone_information(*args, **kwargs):
    return "192.168.0.0/24"
# =============================================================

"""


def tiramisu_display_name(kls,
                          dyn_name: 'Base'=None,
                          suffix: str=None,
                          ) -> str:
    if dyn_name is not None:
        name = dyn_name
    else:
        name = kls.impl_getname()
    return name


GENERATED_DIR = 'generated'
if isdir(GENERATED_DIR):
    rmtree(GENERATED_DIR)


applications = {}


test_ok = []
for distrib in listdir('seed'):
    distrib_dir = join('seed', distrib, 'applicationservice')
    if isdir(distrib_dir):
        for release in listdir(distrib_dir):
            release_dir = join(distrib_dir, release)
            if isdir(release_dir):
                for applicationservice in listdir(release_dir):
                    applicationservice_dir = join(release_dir, applicationservice)
                    if isdir(applicationservice_dir):
                        if applicationservice in applications:
                            raise Exception('multi applicationservice')
                        applications[applicationservice] = applicationservice_dir
                        if isdir(join(applicationservice_dir, 'dictionaries')):
                            test_ok.append(applicationservice_dir)



@fixture(scope="module", params=test_ok)
def test_dir(request):
    return request.param


def calc_depends(xmls, appname):
    application = join(applications[appname], 'applicationservice.yml')
    with open(application) as yaml:
        app = load(yaml, Loader=SafeLoader)

    if 'depends' in app and app['depends']:
        for xml in app['depends']:
            if xml not in xmls:
                xmls.insert(0, xml)
            calc_depends(xmls, xml)


@mark.asyncio
async def test_template(test_dir):
#    if test_dir != 'seed/eole/applicationservice/2020.1.1/eole-lemonldap-ng':
#        return
#    if test_dir in [
#                    'seed/eole/applicationservice/2020.1.1/eole-proxy',
#                    'seed/eole/applicationservice/2020.1.1/eole-bareos',
#                    ]:
#        return
    if not isdir(GENERATED_DIR):
        makedirs(GENERATED_DIR)
    test_name = basename(test_dir)
    root_dir = f'{GENERATED_DIR}/{test_name}'
    tmp_dir = f'{root_dir}/tmp'
    funcs_dir = f'{root_dir}/funcs'
    for dirname in [root_dir, tmp_dir, funcs_dir]:
        if isdir(dirname):
            rmtree(dirname)
        makedirs(dirname)
    print()
    print('>', test_dir)
    test_file = test_dir.rsplit('/', 1)[-1]
    xmls = [test_file]
    calc_depends(xmls, test_file)
    print('  o dependencies: ' + ', '.join(xmls))
    with open(f'{root_dir}/dependencies.txt', 'w') as fh:
        dump(xmls, fh)
    dictionaries = [join(applications[xml], 'dictionaries') for xml in xmls if isdir(join(applications[xml], 'dictionaries'))]
    if 'risotto_setting.rougail' not in RougailConfig['extra_annotators']:
        RougailConfig['extra_annotators'].append('risotto_setting.rougail')
    RougailConfig['dictionaries_dir'] = dictionaries
    print('  o dictionaries: ' + ', '.join(dictionaries))
    RougailConfig['extra_dictionaries'] = {}
    extra_dictionaries = []
    for xml in xmls:
        extras_dir = join(applications[xml], 'extras')
        if isdir(extras_dir):
            for extra in listdir(extras_dir):
                extra_dir = join(extras_dir, extra)
                if isdir(extra_dir):
                    RougailConfig['extra_dictionaries'].setdefault(extra, []).append(extra_dir)
                    extra_dictionaries.append(extra_dir)
    if RougailConfig['extra_dictionaries']:
        print('  o extra dictionaries: ' + ', '.join(extra_dictionaries))
    if isdir('templates'):
        rmtree('templates')
    makedirs('templates')
    func = f'{funcs_dir}/funcs.py'
    with open(func, 'wb') as fh:
        fh.write(FUNCTIONS)
        for xml in xmls:
            func_dir = join(applications[xml], 'funcs')
            if isdir(func_dir):
                for f in listdir(func_dir):
                    if not f.startswith('__'):
                        with open(join(func_dir, f), 'rb') as rfh:
                            fh.write(rfh.read())
            templates_dir = join(applications[xml], 'templates')
            if isdir(templates_dir):
                for f in listdir(templates_dir):
                    template_file = join(templates_dir, f)
                    dest_template_file = join('templates', f)
                    if isfile(dest_template_file):
                        raise Exception(f'duplicate template {f}')
                    link(template_file, dest_template_file)

    RougailConfig['functions_file'] = func
    eolobj = RougailConvert()
    xml = eolobj.save(join(root_dir, 'dictionary.py'))
    optiondescription = {}
    try:
        exec(xml, None, optiondescription)
    except Exception as err:
        print('Tiramisu file ' + join(root_dir, 'dictionary.py'))
        print_exc()
        raise Exception(f'unknown error when load tiramisu object {err}') from err
    config = await Config(optiondescription['option_0'], display_name=tiramisu_display_name)
    rmtree(tmp_dir)
    rmtree(funcs_dir)