rougail/tests/test_3_template.py

from os import listdir, mkdir, readlink
from os.path import join, isdir, isfile, islink
from shutil import rmtree
from pytest import fixture, mark, raises
from lxml.etree import parse
from tiramisu import Config

from rougail import RougailConfig, RougailBaseTemplate, RougailSystemdTemplate
from rougail.error import TemplateError
from jinja2.exceptions import TemplateNotFound


template_dirs = 'tests/dictionaries'
excludes = set([])
#excludes = set(['01base_file_utfchar'])
test_ok = {f for f in listdir(template_dirs) if not f.startswith('_') and isdir(join(template_dirs, f, 'tmpl'))}
test_ok -= excludes
test_ok = list(test_ok)
test_ok.sort()
#test_ok = ['70container_files']


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


def find_files(dirname: str,
               root: list,
               files: set,
               ) -> None:
    for filename in listdir(dirname):
        abs_dirname = join(dirname, filename)
        root_file = root + [filename]
        if isdir(join(dirname, filename)):
            find_files(abs_dirname,
                       root_file,
                       files,
                       )
        else:
            files.add(join(*root_file))


async def templates(test_dir, filename, root, engine_name, only_one=False):
    test_dir = join(template_dirs, test_dir)
    tmp_dir = join(test_dir, '..', 'tmp')

    modulepath = test_dir.replace('/', '.') + f'.tiramisu.{filename}'
    mod = __import__(modulepath)
    for token in modulepath.split(".")[1:]:
        mod = getattr(mod, token)
    config = await Config(mod.option_0)
    await config.property.read_only()
    if filename == 'multi':
        sconfig = config
        config = config.option(root)
    
    funcs_file = join(template_dirs, '../eosfunc/test.py')
    distrib_dir = join(test_dir, 'tmpl')
    dest_dir = join(test_dir, 'dest')
    RougailConfig['patches_dir'] = join(test_dir, 'patches')
    RougailConfig['templates_dir'] = distrib_dir
    RougailConfig['tmp_dir'] = tmp_dir
    RougailConfig['functions_file'] = funcs_file
    RougailConfig['destinations_dir'] = dest_dir
    if isfile(join(test_dir, 'file_tmpfiles_factory_dir')):
        RougailConfig['systemd_tmpfile_factory_dir'] = '/test/new/file'
    else:
        RougailConfig['systemd_tmpfile_factory_dir'] = '/usr/local/lib'
    if isfile(join(test_dir, 'file_tmpfiles_delete_before_create')):
        RougailConfig['systemd_tmpfile_delete_before_create'] = True
    else:
        RougailConfig['systemd_tmpfile_delete_before_create'] = False
    if engine_name == 'base':
        engine = RougailBaseTemplate(config)
    else:
        engine = RougailSystemdTemplate(config)
    if not only_one:
        list_templates = [None]
    else:
        list_templates = set()
        find_files(distrib_dir,
                   [],
                   list_templates,
                   )
    for template in list_templates:
        if isdir(tmp_dir):
            rmtree(tmp_dir)
        mkdir(tmp_dir)
        if isdir(dest_dir):
            rmtree(dest_dir)
        mkdir(dest_dir)
        if template is None:
            await engine.instance_files()
            list_results = set()
            if isdir(join(test_dir, 'result')):
                find_files(join(test_dir, 'result'),
                           [],
                           list_results,
                           )
        else:
            list_results = set()
            included = False
            activate = True
            for service in await config.option('services').option.list('all'):
                names = [await o.option.name() for o in await service.option.list('optiondescription')]
                if 'files' in names:
                    for file in await service.option('files').list('optiondescription'):
                        if not await file.option('source').value.get() == template:
                            continue
                        included = await file.information.get('included', 'no') != 'no'
                        values = await file.option('name').value.get()
                        if isinstance(values, list):
                            for value in values:
                                list_results.add(value[1:])
                        else:
                            list_results.add(values[1:])
                        if not await file.option('activate').value.get():
                            activate = False
                # if template is an overrides
                if not list_results and 'overrides' in names:
                    for override in await service.option('overrides').list('optiondescription'):
                        if await override.option.description() == template:
                            return
                # if template is a service
                if not list_results and template == await service.option.description() and await service.information.get('engine', None):
                    return
            #this file is include and not declared
            if template.startswith('inc') and not list_results:
                return
            try:
                await engine.instance_file(template)
            except TemplateError as err:

                if 'No such file or directory' in str(err):
                    continue
                if included or not activate:
                    continue
                raise err from err
            except TemplateNotFound:
                continue
            assert activate, 'template should be desactivate'
            assert not included, 'include file should not be templated'
#            else:
#                assert list_results, "should find file"

        list_files = set()
        if isdir(dest_dir):
            find_files(dest_dir,
                       [],
                       list_files,
                       )
        if engine_name == 'base' and 'tmpfiles.d/0rougail.conf' in list_results:
            list_results.remove('tmpfiles.d/0rougail.conf')
        assert list_files == list_results, f'error with {test_dir}:'
        for result in list_results:
            template_file = join(dest_dir, result)
            assert islink(template_file) == islink(join(test_dir, 'result', result))
            if islink(template_file):
                assert readlink(template_file) == readlink(join(test_dir, 'result', result))
                continue
            if not isfile(template_file):
                raise Exception(f'{template_file} is not generated')
            with open(join(test_dir, 'result', result), 'r') as fh:
                result_file = fh.read()
            with open(template_file, 'r') as fh:
                generated_file = fh.read()
            assert result_file == generated_file, f'{template_file} content : \n{generated_file}\nexpected: \nresult_file\n'
        if isdir(dest_dir):
            rmtree(dest_dir)
        if isdir(tmp_dir):
            rmtree(tmp_dir)


@mark.asyncio
async def test_templates(test_dir):
    for engine in ['base', 'systemd']:
        not_file = join(template_dirs, test_dir, 'no_' + engine)
        if isfile(not_file):
            with raises(Exception) as err:
                await templates(test_dir, 'base', '1', engine)
        else:
            await templates(test_dir, 'base', '1', engine)


@mark.asyncio
async def test_template(test_dir):
    for engine in ['base', 'systemd']:
        not_file = join(template_dirs, test_dir, 'no_' + engine)
        if isfile(not_file):
            continue
        await templates(test_dir, 'base', '1', engine, only_one=True)


@mark.asyncio
async def test_template_multi_1(test_dir):
    for engine in ['base', 'systemd']:
        not_file = join(template_dirs, test_dir, 'no_' + engine)
        if isfile(not_file):
            with raises(Exception) as err:
                await templates(test_dir, 'multi', '1', engine)
        else:
            await templates(test_dir, 'multi', '1', engine)


@mark.asyncio
async def test_template_multi_2(test_dir):
    for engine in ['base', 'systemd']:
        not_file = join(template_dirs, test_dir, 'no_' + engine)
        if isfile(not_file):
            with raises(Exception) as err:
                await templates(test_dir, 'multi', '1', engine)
        else:
            await templates(test_dir, 'multi', '1', engine)