214 lines
8.2 KiB
Python
214 lines
8.2 KiB
Python
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 = ['70service_servicelist_file']
|
|
|
|
|
|
@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)
|