import os from pytest import fixture, raises from pathlib import Path from rougail import Rougail, RougailConfig ######################### from dotenv import load_dotenv from rougail.user_data_environment import RougailUserDataEnvironment as RougailUserData from json import load, dump ######################### from rougail_tests.utils import get_structures_list, get_rougail_config, get_values_for_config, config_to_dict EXT = "env" ######################### #let's save the original environment save = os.environ.copy() ######################### excludes = ["60_0family_dynamic_source_hidden"] test_ok = get_structures_list(excludes) # test_ok = [Path('../rougail-tests/structures/60_0family_dynamic_1_0_empty')] def idfn(fixture_value): return fixture_value.name @fixture(scope="module", params=test_ok, ids=idfn) def test_dir(request): return request.param def _test_structural_files(test_dir, namespace, ext, *, level, need_exclude=False): rougailconfig = get_rougail_config(test_dir, namespace) if not rougailconfig: return # rougailconfig['tiramisu_cache'] = "cache.py" ################################## rougailconfig['step.user_data'] = ['environment'] rougailconfig['environment.custom_separator'] = 'o' ################################## dir_name = 'test' if namespace: dir_name += '_namespace' elif (test_dir / 'force_namespace').is_file(): return rougail = Rougail(rougailconfig) config = rougail.run() ################################## root_path = Path('tests') / 'results' / dir_name / test_dir.name makedict = root_path / 'makedict' / f'{level}.json' filename = root_path / 'file' if need_exclude: filename = filename / f'{level}_exclude.{EXT}' else: filename = filename / f'{level}.{EXT}' try: populate(filename, makedict, rougailconfig, level, need_exclude, namespace) except SubMulti: return load_dotenv(str(filename)) ################################## # loads variables in the tiramisu config generated_user_data = RougailUserData(config, rougailconfig=rougailconfig).run() errors = rougail.user_datas(generated_user_data) #expected output expected_filename = Path('tests') / 'results' / dir_name / test_dir.name / 'makedict' / f'{level}.json' with expected_filename.open() as json_file: expected = load(json_file) # here is the effective test errors_file = Path('tests') / 'results' / dir_name / test_dir.name / 'errors' / f'{level}.json' if not errors_file.is_file(): errors_file.parent.mkdir(parents=True, exist_ok=True) with open(errors_file, 'a') as json_file: dump(errors, json_file, indent=4) with open(errors_file) as json_file: expected_errors = load(json_file) # expected_errors = { # 'errors': [], # 'warnings': [], # } assert expected_errors == errors, errors_file # config.property.read_only() config_dict = dict(config_to_dict(config.value.get())) assert expected == config_dict, expected_filename ###################################### #teardown: set the original environment again os.environ = save.copy() ###################################### def populate(filename, makedict_file, rougailconfig, level, need_exclude, namespace): if filename.is_file() and makedict_file.is_file(): return config = Rougail(rougailconfig).run() values = get_values_for_config(config, not need_exclude, level) if not filename.is_file(): filename.parent.mkdir(parents=True, exist_ok=True) ############################################## val = values_to_env(values, namespace) for idx, v in enumerate(val): if idx / 2 != int(idx / 2): a, b = v.split('=') val[idx] = a.replace('.', 'o') + "=" + b ############################################## with filename.open('w') as fh: ############################################## fh.write('\n'.join(val) + '\n') ############################################## if not makedict_file.is_file(): makedict_file.parent.mkdir(parents=True, exist_ok=True) config.property.read_only() config_dict = dict(config_to_dict(config.value.get())) with makedict_file.open('w') as fh: dump(config_dict, fh, indent=4) fh.write('\n') class SubMulti(Exception): pass ############################################## def values_to_env(values, namespace): def parse(root_path, values_, level=-1): level += 1 for key, value in values_.items(): if level == 0: sub_root_path = key else: sub_root_path = root_path + '.' + key if isinstance(value, dict): yield from parse(sub_root_path, value, level) elif isinstance(value, list): if value and isinstance(value[0], dict): keys = [] for ls in value: for k in ls: if k not in keys: keys.append(k) for k in keys: for val in value: sub_val = val.get(k, "") if isinstance(sub_val, list): raise SubMulti("submulti is not allowed for environment user_datas") ns = (sub_root_path + '.' + k).upper() if not namespace: ns = "ROUGAIL_" + ns yield ns + '="' + ','.join([convert_str(val.get(k, "")) for val in value])+ '"' else: ns = sub_root_path.upper() if not namespace: ns = "ROUGAIL_" + ns yield ns + '="' + ','.join([convert_str(val) for val in value]) + '"' elif value is None or value == 'None': ns = sub_root_path.upper() if not namespace: ns = "ROUGAIL_" + ns yield ns + '=""' else: ns = sub_root_path.upper() if not namespace: ns = "ROUGAIL_" + ns yield ns + '="' + str(value) + '"' return list(parse(None, values)) def convert_str(val): if isinstance(val, bool): return {True: 'true', False: 'false'}.get(val) if val is None: return "" return str(val) ############################################## def test_structural_files_all(test_dir): "tests the output" _test_structural_files(test_dir, False, EXT, level='all') def test_structural_files_all_exclude(test_dir): "tests the output" _test_structural_files(test_dir, False, EXT, level='all', need_exclude=True) def test_structural_files_mandatories(test_dir): "tests the output" _test_structural_files(test_dir, False, EXT, level='mandatories') def test_structural_files_namespace_all(test_dir): "tests the output" _test_structural_files(test_dir, True, EXT, level='all') def test_structural_files_namespace_all_exclude(test_dir): "tests the output" _test_structural_files(test_dir, True, EXT, level='all', need_exclude=True) def test_structural_files_namespace_mandatories(test_dir): "tests the output" _test_structural_files(test_dir, True, EXT, level='mandatories') ####################################################################### error_env = list((Path(__file__).parent / 'errors' / 'environment').glob("*.env")) error_env.sort() @fixture(scope="module", params=error_env, ids=idfn) def test_file_error(request): return request.param def test_structural_files_error(test_file_error): rougailconfig = get_rougail_config(test_file_error.parent.parent / 'structure') ################################## rougailconfig['step.user_data'] = ['environment'] rougail = Rougail(rougailconfig) config = rougail.run() load_dotenv(str(test_file_error)) ################################## # loads variables in the tiramisu config generated_user_data = RougailUserData(config, rougailconfig=rougailconfig).run() errors = rougail.user_datas(generated_user_data) errors_file = test_file_error.parent.parent / "results" / test_file_error.name if not errors_file.is_file(): errors_file.parent.mkdir(parents=True, exist_ok=True) with open(errors_file, 'a') as json_file: dump(errors, json_file, indent=4) with open(errors_file) as json_file: expected_errors = load(json_file) ###################################### #teardown: set the original environment again os.environ = save.copy() ###################################### assert expected_errors == errors, errors_file def test_environment_name(): rougailconfig = RougailConfig.copy() rougailconfig["main_namespace"] = None rougailconfig['step.user_data'] = ['environment'] assert rougailconfig["environment.default_environment_name"] == "ROUGAIL" with raises(ValueError): rougailconfig["environment.default_environment_name"] = "lower" with raises(ValueError): rougailconfig["environment.default_environment_name"] = "LOWEr" rougailconfig["environment.default_environment_name"] = "UPPER"