255 lines
9.3 KiB
Python
255 lines
9.3 KiB
Python
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"
|