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


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 = ['01base_file_include_content']


@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 template(test_dir, filename, root, engine_name):
    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')
    if isdir(tmp_dir):
        rmtree(tmp_dir)
    mkdir(tmp_dir)
    dest_dir = join(test_dir, 'dest')
    if isdir(dest_dir):
        rmtree(dest_dir)
    mkdir(dest_dir)
    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)
    await engine.instance_files()
    list_templates = set()
    if isdir(dest_dir):
        find_files(dest_dir,
                   [],
                   list_templates,
                   )
    list_results = set()
    if isdir(join(test_dir, 'result')):
        find_files(join(test_dir, 'result'),
                   [],
                   list_results,
                   )
    if engine_name == 'base' and 'tmpfiles.d/0rougail.conf' in list_results:
        list_results.remove('tmpfiles.d/0rougail.conf')
    assert list_templates == 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_template(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 template(test_dir, 'base', '1', engine)
        else:
            await template(test_dir, 'base', '1', engine)


@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 template(test_dir, 'multi', '1', engine)
        else:
            await template(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 template(test_dir, 'multi', '1', engine)
        else:
            await template(test_dir, 'multi', '1', engine)