stove/rougail
3
0
Fork 0
Code Issues 5 Pull requests 1 Projects Releases Wiki Activity

feat: user_data can load secret manager values

Browse source
This commit is contained in:
egarette@silique.fr 2025-10-05 21:41:26 +02:00
parent fb0f1e5f2c
commit 5759e4f86b
2 changed files with 76 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
src/rougail/config/__init__.py
View file

@ -159,6 +159,15 @@ class _RougailConfig:
return not ret
return ret
def __contains__(
self, key,
) -> None:
try:
self.__getitem__(key)
except AttributeError:
return False
return True
def get_leadership(self, option) -> dict:
leader = None
followers = []

91
src/rougail/user_datas.py
View file

@ -28,8 +28,9 @@ from tiramisu.error import (
LeadershipError,
ConfigError,
CancelParam,
display_list,
)
from .utils import undefined
from .utils import undefined, get_properties_to_string
from .tiramisu import (
normalize_family,
CONVERT_OPTION,
@ -63,6 +64,7 @@ class UserDatas:
self.values = user_datas
self._auto_configure_dynamics()
self._populate_config()
self.properties_to_string = get_properties_to_string()
if return_values_not_error:
return self.values
else:
@ -133,6 +135,8 @@ class UserDatas:
identifier = get_identifier_from_dynamic_family(
tconfig.name(), name
)
if identifier == "{{ identifier }}":
continue
if identifier != normalize_family(identifier):
msg = _(
'cannot load variable path "{0}", the identifier "{1}" is not valid in {2}'
@ -200,16 +204,21 @@ class UserDatas:
return value
def _populate_config(self):
dynamics_variable = []
while self.values:
value_is_set = False
for option in self._get_variable(self.config):
path = option.path()
values_path = path = option.path()
if path not in self.values:
continue
if path in dynamics_variable or not option.isdynamic():
continue
values_path = option.path(uncalculated=True)
if values_path not in self.values:
continue
if self.only_default and option.owner.get() != owners.default:
self.values.pop(path)
self.values.pop(values_path)
continue
options = self.values[path].get("options", {})
options = self.values[values_path].get("options", {})
if (
options.get("allow_secrets_variables", True) is False
and option.type() == "password"
@ -217,15 +226,19 @@ class UserDatas:
self.errors.append(
_(
'the variable "{0}" contains secrets and should not be defined in {1}'
).format(path, self.values[path]["source"])
).format(path, self.values[values_path]["source"])
)
continue
value = self.convert_value(
path, option, options, self.values[path]["values"]
path, option, options, self.values[values_path]["values"]
)
index = option.index()
if index is not None:
if isinstance(value, tuple):
self.values[values_path]["values"] = []
for i in range(len(option.parent().leader().value.get())):
self.values[values_path]["values"].append(value)
value = self.values[values_path]["values"]
if not isinstance(value, list) or index >= len(value):
continue
value = value[index]
@ -233,33 +246,40 @@ class UserDatas:
else:
option_without_index = option
if option.isleader():
# set value for a leader, it began to remove all values!
len_leader = len(option.value.get())
if len_leader:
for idx in range(len_leader - 1, -1, -1):
option.value.pop(idx)
try:
option.value.set(value)
self.set_value(option, value, options)
# option.value.set(value)
value_is_set = True
except Exception:
except Exception as err:
if path != option.path():
self.values[option.path()] = self.values.pop(path)
self.values[option.path()] = self.values.pop(values_path)
else:
if "source" in self.values[path]:
if "source" in self.values[values_path]:
option_without_index.information.set(
"loaded_from",
_("loaded from {0}").format(self.values[path]["source"]),
_("loaded from {0}").format(self.values[values_path]["source"]),
)
# value is correctly set, remove variable to the set
if index is not None:
# if it's a follower waiting for all followers are sets
self.values[path]["values"][index] = undefined
for tmp_value in self.values[path]["values"]:
self.values[values_path]["values"][index] = undefined
for tmp_value in self.values[values_path]["values"]:
if tmp_value != undefined:
break
else:
self.values.pop(path)
else:
if path in self.values:
self.values.pop(path)
else:
dynamics_variable.append(path)
elif path in self.values:
self.values.pop(path)
else:
dynamics_variable.append(path)
if not value_is_set:
break
@ -272,7 +292,11 @@ class UserDatas:
# we don't find variable, apply value just to get error or warning messages
for path, options in self.values.items():
value = options["values"]
option = self.config.option(path)
if options.get('secret_manager'):
option = self.config.forcepermissive.option(path)
else:
option = self.config.option(path)
try:
if option.isoptiondescription():
if value:
@ -320,12 +344,10 @@ class UserDatas:
continue
self.config.option(path, index).value.set(value)
else:
option.value.set(value)
self.set_value(option, value, options.get("options", {}))
except PropertiesOptionError as err:
if err.code == "property-error":
properties = err.display_properties(
force_property=True, add_quote=False
)
properties = display_list([_(prop) for prop in err.proptype], add_quote=False)
err_path = err._subconfig.path
display_name = option.description(with_quote=True)
if index is not None:
@ -381,11 +403,13 @@ class UserDatas:
except ValueError as err:
err.prefix = ""
if index is not None:
type_ = option.type(translation=True)
self.warnings.append(
_(
'the value "{0}" is invalid for {1} at index "{2}", {3}, it will be ignored when loading from {4}'
'the value "{0}" is an invalid {1} for {2} at index "{3}", {4}, it will be ignored when loading from {5}'
).format(
self._display_value(option, value),
type_,
option.description(with_quote=True),
index,
err,
@ -393,16 +417,35 @@ class UserDatas:
)
)
else:
type_ = option.type(translation=True)
self.warnings.append(
_(
'the value "{0}" is invalid for {1}, {2}, it will be ignored when loading from {3}'
'the value "{0}" is an invalid {1} for {2}, {3}, it will be ignored when loading from {4}'
).format(
self._display_value(option, value),
type_,
option.description(with_quote=True),
err,
options["source"],
)
)
except AttributeOptionError as err:
if err.code == "option-dynamic":
continue
raise err from err
def set_value(self, option, value, options):
is_secret_manager = options.get("secret_manager", False)
if is_secret_manager and isinstance(value, tuple):
# it's a function
value = value[0](*value[1:], option=option, warnings=self.warnings, errors=self.errors)
if is_secret_manager and "hidden" in option.property.get():
option.permissive.add("hidden")
option.permissive.add("frozen")
option.value.set(value)
if is_secret_manager and "hidden" in option.property.get():
option.permissive.remove("hidden")
option.permissive.remove("frozen")
def convert_value(option, value):