yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

vars/generate: improve output when vars are updated

Browse Source 
fixes #2076

- print old and new value if possible
- also inform the user if something hasn't changed
This commit is contained in:
DavHau
2024-09-18 16:56:10 +02:00
parent 78df6cf757
commit 89d39186ee
4 changed files with 134 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
pkgs/clan-cli/clan_cli/vars/_types.py
View File

@@ -1,3 +1,4 @@
import logging
import shutil
from abc import ABC, abstractmethod
from dataclasses import dataclass
@@ -5,6 +6,15 @@ from pathlib import Path
from clan_cli.machines import machines
log = logging.getLogger(__name__)
def string_repr(value: bytes) -> str:
try:
return value.decode()
except UnicodeDecodeError:
return "<binary blob>"
@dataclass
class Prompt:
@@ -50,10 +60,7 @@ class Var:
@property
def printable_value(self) -> str:
try:
return self.value.decode()
except UnicodeDecodeError:
return "<binary blob>"
return string_repr(self.value)
def set(self, value: bytes) -> None:
self._store.set(self.generator, self.name, value, self.shared, self.deployed)
@@ -128,6 +135,16 @@ class StoreBase(ABC):
shared: bool = False,
deployed: bool = True,
) -> Path | None:
if self.exists(generator_name, var_name, shared):
if self.is_secret_store:
old_val = None
old_val_str = "********"
else:
old_val = self.get(generator_name, var_name, shared)
old_val_str = string_repr(old_val)
else:
old_val = None
old_val_str = "<not set>"
directory = self.directory(generator_name, var_name, shared)
# delete directory
if directory.exists():
@@ -135,6 +152,19 @@ class StoreBase(ABC):
# re-create directory
directory.mkdir(parents=True, exist_ok=True)
new_file = self._set(generator_name, var_name, value, shared, deployed)
if self.is_secret_store:
print(f"Updated secret var {generator_name}/{var_name}\n")
else:
if value != old_val:
print(
f"Updated var {generator_name}/{var_name}\n"
f" old: {old_val_str}\n"
f" new: {string_repr(value)}"
)
else:
print(
f"Var {generator_name}/{var_name} remains unchanged: {old_val_str}"
)
return new_file
def get_all(self) -> list[Var]:

4
pkgs/clan-cli/clan_cli/vars/generate.py
View File

@@ -223,7 +223,7 @@ def get_closure(
return minimal_closure([generator_name], generators)
def _generate_vars_for_machine(
def generate_vars_for_machine(
machine: Machine,
generator_name: str | None,
regenerate: bool,
@@ -254,7 +254,7 @@ def generate_vars(
for machine in machines:
errors = []
try:
was_regenerated |= _generate_vars_for_machine(
was_regenerated |= generate_vars_for_machine(
machine, generator_name, regenerate
)
machine.flush_caches()

21
pkgs/clan-cli/clan_cli/vars/set.py
View File

@@ -7,23 +7,38 @@ from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.machines.machines import Machine
from clan_cli.vars.get import get_var
from ._types import Var
from .prompt import ask
log = logging.getLogger(__name__)
def set_command(machine: str, var_id: str, flake: FlakeId) -> None:
def set_var(
machine: str | Machine, var: str | Var, value: bytes, flake: FlakeId
) -> None:
if isinstance(machine, str):
_machine = Machine(name=machine, flake=flake)
else:
_machine = machine
if isinstance(var, str):
_var = get_var(_machine, var)
else:
_var = var
_var.set(value)
def set_via_stdin(machine: str, var_id: str, flake: FlakeId) -> None:
_machine = Machine(name=machine, flake=flake)
var = get_var(_machine, var_id)
if sys.stdin.isatty():
new_value = ask(var.id, "hidden").encode("utf-8")
else:
new_value = sys.stdin.buffer.read()
var.set(new_value)
set_var(_machine, var, new_value, flake)
def _set_command(args: argparse.Namespace) -> None:
set_command(args.machine, args.var_id, args.flake)
set_via_stdin(args.machine, args.var_id, args.flake)
def register_set_parser(parser: argparse.ArgumentParser) -> None:

80
pkgs/clan-cli/tests/test_vars.py
View File

@@ -14,10 +14,12 @@ from clan_cli.vars.check import check_vars
from clan_cli.vars.list import stringify_all_vars
from clan_cli.vars.public_modules import in_repo
from clan_cli.vars.secret_modules import password_store, sops
from clan_cli.vars.set import set_var
from fixtures_flakes import generate_flake
from helpers import cli
from helpers.nixos_config import nested_dict
from root import CLAN_CORE
from stdout import CaptureOutput
def test_dependencies_as_files() -> None:
@@ -636,3 +638,81 @@ def test_default_value(
)
).stdout.strip()
assert json.loads(value_eval) == "hello"
@pytest.mark.impure
def test_stdout_of_generate(
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
capture_output: CaptureOutput,
) -> None:
config = nested_dict()
my_generator = config["clan"]["core"]["vars"]["generators"]["my_generator"]
my_generator["files"]["my_value"]["secret"] = False
my_generator["script"] = "echo -n hello > $out/my_value"
my_secret_generator = config["clan"]["core"]["vars"]["generators"][
"my_secret_generator"
]
my_secret_generator["files"]["my_secret"]["secret"] = True
my_secret_generator["script"] = "echo -n hello > $out/my_secret"
flake = generate_flake(
temporary_home,
flake_template=CLAN_CORE / "templates" / "minimal",
machine_configs={"my_machine": config},
monkeypatch=monkeypatch,
)
monkeypatch.chdir(flake.path)
from clan_cli.vars.generate import generate_vars_for_machine
with capture_output as output:
generate_vars_for_machine(
Machine(name="my_machine", flake=FlakeId(str(flake.path))),
"my_generator",
regenerate=False,
)
assert "Updated var my_generator/my_value" in output.out
assert "old: <not set>" in output.out
assert "new: hello" in output.out
set_var("my_machine", "my_generator/my_value", b"world", FlakeId(str(flake.path)))
with capture_output as output:
generate_vars_for_machine(
Machine(name="my_machine", flake=FlakeId(str(flake.path))),
"my_generator",
regenerate=True,
)
assert "Updated var my_generator/my_value" in output.out
assert "old: world" in output.out
assert "new: hello" in output.out
# check the output when nothing gets regenerated
with capture_output as output:
generate_vars_for_machine(
Machine(name="my_machine", flake=FlakeId(str(flake.path))),
"my_generator",
regenerate=True,
)
assert "Updated" not in output.out
assert "hello" in output.out
with capture_output as output:
generate_vars_for_machine(
Machine(name="my_machine", flake=FlakeId(str(flake.path))),
"my_secret_generator",
regenerate=False,
)
assert "Updated secret var my_secret_generator/my_secret" in output.out
assert "hello" not in output.out
set_var(
"my_machine",
"my_secret_generator/my_secret",
b"world",
FlakeId(str(flake.path)),
)
with capture_output as output:
generate_vars_for_machine(
Machine(name="my_machine", flake=FlakeId(str(flake.path))),
"my_secret_generator",
regenerate=True,
)
assert "Updated secret var my_secret_generator/my_secret" in output.out
assert "world" not in output.out
assert "hello" not in output.out