from typing import Any from pytest import fixture, mark from os import makedirs, listdir, link from os.path import isfile, isdir, join from yaml import load, SafeLoader from json import dump from shutil import rmtree from traceback import print_exc from tiramisu import Config from tiramisu.error import PropertiesOptionError, LeadershipError from rougail import RougailConvert, RougailSystemdTemplate, RougailConfig IDX = 0 GENERATED_DIR = 'generated' if isdir(GENERATED_DIR): rmtree(GENERATED_DIR) test_ok = [] for distrib in listdir('seed'): distrib_dir = join('seed', distrib, 'zone') if isdir(distrib_dir): for zone in listdir(distrib_dir): zone_dir = join(distrib_dir, zone) if isdir(join(zone_dir, 'dictionaries')): test_ok.append(zone_dir) @fixture(scope="module", params=test_ok) def test_dir(request): return request.param async def get_values(option: 'TiramisuOption', type: str, properties: list, test_value: Any, multi: bool, has_dependency: bool) -> list: if test_value is not None: values = test_value has_dependency = True elif type == 'choice': values = await option.value.list() elif type == 'domainname': values = ['test.com'] elif type == 'filename': values = ['/a'] elif type == 'ip': true_option = await option.option.get() if true_option.impl_get_extra('_cidr'): values = ['192.168.1.1/24', '192.168.0.1/32'] else: values = ['192.168.1.1'] elif type == 'network': true_option = await option.option.get() if true_option.impl_get_extra('_cidr'): values = ['192.168.1.0/24', '192.168.1.1/32'] else: values = ['192.168.1.0'] elif type in ['string', 'password']: values = ['string'] elif type == 'integer': values = [0, 1] elif type in 'port': values = ["80"] elif type == 'boolean': values = [True, False] else: raise Exception(f'Unknown type {type}') if not has_dependency: values = [values[0]] if multi: values = [values] if 'mandatory' not in properties: if not multi: values.insert(0, None) else: values.insert(0, []) return values async def build_values(paths: list, paths_values: dict, dependencies, ): path, option, properties, test = paths[0] for val in await get_values(option, await option.option.type(), properties, test, await option.option.ismulti(), path in dependencies): new_paths_values = paths_values.copy() new_paths_values[path] = val new_values = paths[1:] if new_values: async for value in build_values(new_values, new_paths_values, dependencies, ): yield value else: yield new_paths_values def calc_depends(test_root_dir, xmls, appname): zone = join(test_root_dir, appname, 'zone.yml') with open(zone) as yaml: app = load(yaml, Loader=SafeLoader) if 'depends' in app and app['depends']: for xml in app['depends']: xmls.insert(0, xml) calc_depends(test_root_dir, xmls, xml) @mark.asyncio async def test_template(test_dir): global IDX if not isdir(GENERATED_DIR): makedirs(GENERATED_DIR) root_dir = f'{GENERATED_DIR}/{IDX}' IDX += 1 if isdir(root_dir): rmtree(root_dir) tmp_dir = f'{root_dir}/tmp' funcs_dir = f'{root_dir}/funcs' for dirname in [root_dir, tmp_dir, funcs_dir]: makedirs(dirname) print() print('>', test_dir) test_root_dir, test_file = test_dir.rsplit('/', 1) xmls = [test_file] calc_depends(test_root_dir, xmls, test_file) print(' o dependencies: ' + ', '.join(xmls)) with open(f'{root_dir}/dependencies.txt', 'w') as fh: dump(xmls, fh) dictionaries = [join(test_root_dir, xml, 'dictionaries') for xml in xmls if isdir(join(test_root_dir, xml, 'dictionaries'))] print(' o dictionaries: ' + ', '.join(dictionaries)) RougailConfig['dictionaries_dir'] = dictionaries RougailConfig['extra_dictionaries'] = {} extras_dir = join(test_dir, 'extras') if isdir(extras_dir): for extra in listdir(extras_dir): extra_dir = join(extras_dir, extra) if isdir(extra_dir): RougailConfig['extra_dictionaries'][extra] = [extra_dir] if isdir('templates'): rmtree('templates') makedirs('templates') func = f'{funcs_dir}/funcs.py' with open(func, 'wb') as fh: fh.write(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" # ============================================================= """) for func_file in xmls: func_dir = join(test_root_dir, func_file, '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(test_root_dir, func_file, '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(None) optiondescription = {} try: exec(xml, None, optiondescription) except Exception as err: print(xml) print_exc() raise Exception(f'unknown error when load tiramisu object {err}') from err config = await Config(optiondescription['option_0']) has_services = False for option in await config.option.list('optiondescription'): if await option.option.name() == 'services': has_services = True break if not has_services: print(f' o number of tests: any template') rmtree(root_dir) return await config.property.read_write() paths = [] unrestraint = config.unrestraint unaccessible_paths = [] dependencies = set() for option in await unrestraint.option.list('optiondescription', recursive=True, ): path = await option.option.path() properties = await config.option(path).property.get(only_raises=True, uncalculated=True, ) if properties: unaccessible_paths.append(path) for dependency in await option.option.dependencies(): dependencies.add(await dependency.option.path()) for path in await unrestraint.value.dict(): properties = await config.option(path).property.get(only_raises=True, uncalculated=True, ) if properties: continue for dependency in await option.option.dependencies(): dependencies.add(await dependency.option.path()) for unaccessible_path in unaccessible_paths: if path.startswith(unaccessible_path + '.'): break else: option = unrestraint.option(path) paths.append((path, option, await option.property.get(), await option.information.get('test', None), )) print(' o variables number: {}'.format(len(paths))) idx = 0 async for new_paths_values in build_values(paths, {}, dependencies, ): build_dir = f'{root_dir}/{idx}' dest_dir = f'{build_dir}/dest' for dirname in [build_dir, dest_dir]: makedirs(dirname) try: await config.property.read_write() for path, value in new_paths_values.items(): try: if isinstance(value, tuple): value = list(value) if not await config.option(path).option.isfollower(): await config.option(path).value.set(value) else: await config.option(path, 0).value.set(value) except (PropertiesOptionError, LeadershipError): pass except: print_exc() print(xml) print(f'test with {new_paths_values}') raise Exception('unable to set value') mandatories = await config.value.mandatory() if mandatories: print(xml) dico = await config.value.dict() print(f'test with {dico}') raise Exception(f'Mandatory option not configured {mandatories}') RougailConfig['functions_file'] = func RougailConfig['tmp_dir'] = tmp_dir RougailConfig['templates_dir'] = 'templates' RougailConfig['destinations_dir'] = dest_dir try: await config.property.read_only() engine = RougailSystemdTemplate(config) await engine.instance_files() except Exception as err: print_exc() print(xml) print(f'test with {new_paths_values}') raise Exception(f'unknown error {err}') from err with open(f'{build_dir}/variables.txt', 'w') as fh: dump(await config.value.dict(), fh, indent=4, sort_keys=True) idx += 1 print(f' o number of tests: {idx}') if isdir('templates'): rmtree('templates') rmtree(tmp_dir) rmtree(funcs_dir)