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

vars: refactor - remove generate_vars() in favor of run_generators()

Browse Source 
The motivation is to have one shared entry point for the CLI as well as API/GUI
This commit is contained in:
DavHau
2025-08-19 16:10:08 +07:00
parent ba1e598a76
commit ab274ce932
10 changed files with 102 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pkgs/clan-app/ui/src/workflows/Install/steps/installSteps.tsx
View File

@@ -552,12 +552,14 @@ const InstallSummary = () => {
const runGenerators = client.fetch("run_generators", {
prompt_values: store.install.promptValues,
machine: {
machines: [
{
name: store.install.machineName,
flake: {
identifier: clanUri,
},
},
],
});
set("install", (s) => ({

46
pkgs/clan-cli/clan_cli/tests/test_vars.py
View File

@@ -9,11 +9,13 @@ from clan_cli.tests.fixtures_flakes import ClanFlake
from clan_cli.tests.helpers import cli
from clan_cli.vars.check import check_vars
from clan_cli.vars.generate import (
Generator,
GeneratorKey,
get_generators,
run_generators,
)
from clan_cli.vars.generator import (
Generator,
GeneratorKey,
)
from clan_cli.vars.get import get_machine_var
from clan_cli.vars.graph import all_missing_closure, requested_closure
from clan_cli.vars.list import stringify_all_vars
@@ -698,8 +700,8 @@ def test_api_set_prompts(
monkeypatch.chdir(flake.path)
run_generators(
machine=Machine(name="my_machine", flake=Flake(str(flake.path))),
generators=[GeneratorKey(machine="my_machine", name="my_generator")],
machines=[Machine(name="my_machine", flake=Flake(str(flake.path)))],
generators=["my_generator"],
prompt_values={
"my_generator": {
"prompt1": "input1",
@@ -712,8 +714,8 @@ def test_api_set_prompts(
assert store.exists(my_generator, "prompt1")
assert store.get(my_generator, "prompt1").decode() == "input1"
run_generators(
machine=Machine(name="my_machine", flake=Flake(str(flake.path))),
generators=[GeneratorKey(machine="my_machine", name="my_generator")],
machines=[Machine(name="my_machine", flake=Flake(str(flake.path)))],
generators=["my_generator"],
prompt_values={
"my_generator": {
"prompt1": "input2",
@@ -759,8 +761,8 @@ def test_stdout_of_generate(
# with capture_output as output:
with caplog.at_level(logging.INFO):
run_generators(
Machine(name="my_machine", flake=flake),
generators=[GeneratorKey(machine="my_machine", name="my_generator")],
machines=[Machine(name="my_machine", flake=flake)],
generators=["my_generator"],
)
assert "Updated var my_generator/my_value" in caplog.text
@@ -771,8 +773,8 @@ def test_stdout_of_generate(
set_var("my_machine", "my_generator/my_value", b"world", flake)
with caplog.at_level(logging.INFO):
run_generators(
Machine(name="my_machine", flake=flake),
generators=[GeneratorKey(machine="my_machine", name="my_generator")],
machines=[Machine(name="my_machine", flake=flake)],
generators=["my_generator"],
)
assert "Updated var my_generator/my_value" in caplog.text
assert "old: world" in caplog.text
@@ -781,16 +783,16 @@ def test_stdout_of_generate(
# check the output when nothing gets regenerated
with caplog.at_level(logging.INFO):
run_generators(
Machine(name="my_machine", flake=flake),
generators=[GeneratorKey(machine="my_machine", name="my_generator")],
machines=[Machine(name="my_machine", flake=flake)],
generators=["my_generator"],
)
assert "Updated var" not in caplog.text
assert "hello" in caplog.text
caplog.clear()
with caplog.at_level(logging.INFO):
run_generators(
Machine(name="my_machine", flake=flake),
generators=[GeneratorKey(machine="my_machine", name="my_secret_generator")],
machines=[Machine(name="my_machine", flake=flake)],
generators=["my_secret_generator"],
)
assert "Updated secret var my_secret_generator/my_secret" in caplog.text
assert "hello" not in caplog.text
@@ -803,8 +805,8 @@ def test_stdout_of_generate(
)
with caplog.at_level(logging.INFO):
run_generators(
Machine(name="my_machine", flake=flake),
generators=[GeneratorKey(machine="my_machine", name="my_secret_generator")],
machines=[Machine(name="my_machine", flake=flake)],
generators=["my_secret_generator"],
)
assert "Updated secret var my_secret_generator/my_secret" in caplog.text
assert "world" not in caplog.text
@@ -892,8 +894,8 @@ def test_fails_when_files_are_left_from_other_backend(
monkeypatch.chdir(flake.path)
for generator in ["my_secret_generator", "my_value_generator"]:
run_generators(
Machine(name="my_machine", flake=Flake(str(flake.path))),
generators=GeneratorKey(machine="my_machine", name=generator),
machines=[Machine(name="my_machine", flake=Flake(str(flake.path)))],
generators=generator,
)
# Will raise. It was secret before, but now it's not.
my_secret_generator["files"]["my_secret"]["secret"] = (
@@ -908,13 +910,13 @@ def test_fails_when_files_are_left_from_other_backend(
if generator == "my_secret_generator":
with pytest.raises(ClanError):
run_generators(
Machine(name="my_machine", flake=Flake(str(flake.path))),
generators=GeneratorKey(machine="my_machine", name=generator),
machines=[Machine(name="my_machine", flake=Flake(str(flake.path)))],
generators=generator,
)
else:
run_generators(
Machine(name="my_machine", flake=Flake(str(flake.path))),
generators=GeneratorKey(machine="my_machine", name=generator),
machines=[Machine(name="my_machine", flake=Flake(str(flake.path)))],
generators=generator,
)

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

@@ -1,7 +1,6 @@
import argparse
import logging
from collections.abc import Callable
from typing import Literal
from clan_cli.completions import (
add_dynamic_completer,
@@ -106,19 +105,17 @@ strategies (e.g., interactive CLI, GUI, or programmatic).
@API.register
def run_generators(
machine: Machine,
generators: GeneratorKey
| list[GeneratorKey]
| Literal["all", "minimal"] = "minimal",
machines: list[Machine],
generators: str | list[str] = "minimal",
prompt_values: dict[str, dict[str, str]] | PromptFunc = lambda g: g.ask_prompts(),
no_sandbox: bool = False,
) -> None:
"""Run the specified generators for a machine.
"""Run the specified generators for machines.
Args:
machine: The machine to run generators for.
machines: The machines to run generators for.
generators: Can be:
- GeneratorKey: Single generator to run (ensuring dependencies are met)
- list[GeneratorKey]: Specific generators to run exactly as provided.
- str: Single generator name to run (ensuring dependencies are met)
- list[str]: Specific generator names to run exactly as provided.
Dependency generators are not added automatically in this case.
The caller must ensure that all dependencies are included.
- "all": Run all generators (full closure)
@@ -130,31 +127,35 @@ def run_generators(
ClanError: If the machine or generator is not found, or if there are issues with
executing the generator.
"""
for machine in machines:
if generators == "all":
generator_objects = get_generators(machine, full_closure=True)
elif generators == "minimal":
generator_objects = get_generators(machine, full_closure=False)
elif isinstance(generators, GeneratorKey):
# Single generator - compute minimal closure for it
elif isinstance(generators, str) and generators not in ["all", "minimal"]:
# Single generator name - compute minimal closure for it
generator_objects = get_generators(
machine, full_closure=False, generator_name=generators.name
machine, full_closure=False, generator_name=generators
)
elif isinstance(generators, list):
if len(generators) == 0:
return
generator_keys = set(generators)
# Create GeneratorKeys for this specific machine
generator_keys = {
GeneratorKey(machine=machine.name, name=name) for name in generators
}
all_generators = get_generators(machine, full_closure=True)
generator_objects = [g for g in all_generators if g.key in generator_keys]
else:
msg = f"Invalid generators argument: {generators}. Must be 'all', 'minimal', GeneratorKey, or a list of GeneratorKey"
msg = f"Invalid generators argument: {generators}. Must be 'all', 'minimal', a generator name, or a list of generator names"
raise ValueError(msg)
# If prompt function provided, ask all prompts
# TODO: make this more lazy and ask for every generator on execution
if callable(prompt_values):
prompt_values = {
generator.name: prompt_values(generator) for generator in generator_objects
generator.name: prompt_values(generator)
for generator in generator_objects
}
# execute health check
_ensure_healthy(machine=machine, generators=generator_objects)
@@ -171,38 +172,6 @@ def run_generators(
)
def generate_vars(
machines: list["Machine"],
generator_name: str | None = None,
regenerate: bool = False,
no_sandbox: bool = False,
) -> None:
for machine in machines:
errors = []
try:
generators: GeneratorKey | Literal["all", "minimal"]
if generator_name:
generators = GeneratorKey(machine=machine.name, name=generator_name)
else:
generators = "all" if regenerate else "minimal"
run_generators(
machine,
generators=generators,
no_sandbox=no_sandbox,
)
machine.info("All vars are up to date")
except Exception as exc:
errors += [(machine, exc)]
if len(errors) == 1:
raise errors[0][1]
if len(errors) > 1:
msg = f"Failed to generate vars for {len(errors)} hosts:"
for machine, error in errors:
msg += f"\n{machine}: {error}"
raise ClanError(msg) from errors[0][1]
def generate_command(args: argparse.Namespace) -> None:
flake = require_flake(args.flake)
machines: list[Machine] = list(list_full_machines(flake).values())
@@ -225,10 +194,14 @@ def generate_command(args: argparse.Namespace) -> None:
f"clanInternals.machines.{system}.{{{','.join(machine_names)}}}.config.clan.core.vars.generators.*.validationHash",
]
)
generate_vars(
run_generators(
machines,
args.generator,
args.regenerate,
generators=args.generator
if args.generator
else "all"
if args.regenerate
else "minimal",
no_sandbox=args.no_sandbox,
)

4
pkgs/clan-cli/clan_cli/vms/run.py
View File

@@ -22,7 +22,7 @@ from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.facts.generate import generate_facts
from clan_cli.qemu.qga import QgaSession
from clan_cli.qemu.qmp import QEMUMonitorProtocol
from clan_cli.vars.generate import generate_vars
from clan_cli.vars.generate import run_generators
from clan_cli.vars.upload import populate_secret_vars
from .inspect import VmConfig, inspect_vm
@@ -85,7 +85,7 @@ def get_secrets(
secrets_dir.mkdir(parents=True, exist_ok=True)
generate_facts([machine])
generate_vars([machine])
run_generators([machine])
machine.secret_facts_store.upload(secrets_dir)
populate_secret_vars(machine, secrets_dir)

6
pkgs/clan-cli/clan_lib/flash/flash.py
View File

@@ -7,7 +7,7 @@ from tempfile import TemporaryDirectory
from typing import Any, Literal
from clan_cli.facts.generate import generate_facts
from clan_cli.vars.generate import generate_vars
from clan_cli.vars.generate import run_generators
from clan_cli.vars.upload import populate_secret_vars
from clan_lib.api import API
@@ -78,7 +78,7 @@ def run_machine_flash(
system_config_nix: dict[str, Any] = {}
generate_facts([machine])
generate_vars([machine])
run_generators([machine])
if system_config.language:
if system_config.language not in list_languages():
@@ -113,7 +113,7 @@ def run_machine_flash(
"users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}}
}
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
for generator in Generator.get_machine_generators(machine.name, machine.flake):
for file in generator.files:

4
pkgs/clan-cli/clan_lib/machines/install.py
View File

@@ -7,7 +7,7 @@ from typing import Literal
from clan_cli.facts.generate import generate_facts
from clan_cli.machines.hardware import HardwareConfig
from clan_cli.vars.generate import generate_vars
from clan_cli.vars.generate import run_generators
from clan_lib.api import API, message_queue
from clan_lib.cmd import Log, RunOpts, run
@@ -88,7 +88,7 @@ def run_machine_install(opts: InstallOptions, target_host: Remote) -> None:
# Notify the UI about what we are doing
notify_install_step("generators")
generate_facts([machine])
generate_vars([machine])
run_generators([machine])
with (
TemporaryDirectory(prefix="nixos-install-") as _base_directory,

4
pkgs/clan-cli/clan_lib/machines/morph.py
View File

@@ -7,7 +7,7 @@ from pathlib import Path
from tempfile import TemporaryDirectory
from clan_cli.machines.create import CreateOptions, create_machine
from clan_cli.vars.generate import generate_vars
from clan_cli.vars.generate import run_generators
from clan_lib.cmd import Log, RunOpts, run
from clan_lib.dirs import specific_machine_dir
@@ -78,7 +78,7 @@ def morph_machine(
machine = Machine(name=name, flake=Flake(str(flakedir)))
generate_vars([machine], generator_name=None, regenerate=False)
run_generators([machine], generators="minimal")
machine.secret_vars_store.populate_dir(
machine.name,

4
pkgs/clan-cli/clan_lib/machines/update.py
View File

@@ -7,7 +7,7 @@ from contextlib import ExitStack
from clan_cli.facts.generate import generate_facts
from clan_cli.facts.upload import upload_secrets
from clan_cli.vars.generate import generate_vars
from clan_cli.vars.generate import run_generators
from clan_cli.vars.upload import upload_secret_vars
from clan_lib.api import API
@@ -148,7 +148,7 @@ def run_machine_update(
target_host_root = stack.enter_context(_target_host.become_root())
generate_facts([machine], service=None, regenerate=False)
generate_vars([machine], generator_name=None, regenerate=False)
run_generators([machine], generators="minimal")
# Upload secrets to the target host using root
upload_secrets(machine, target_host_root)

4
pkgs/clan-cli/clan_lib/tests/test_create.py
View File

@@ -231,8 +231,8 @@ def test_clan_create_api(
collected_prompt_values[generator.name] = prompt_values
run_generators(
machine=machine,
generators=[gen.key for gen in generators],
machines=[machine],
generators=[gen.name for gen in generators],
prompt_values=collected_prompt_values,
)

11
pkgs/generate-test-vars/generate_test_vars/cli.py
View File

@@ -10,9 +10,9 @@ from dataclasses import dataclass
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any, override
from unittest.mock import patch
from clan_cli.vars.generate import Generator, generate_vars
from clan_cli.vars.generate import run_generators
from clan_cli.vars.generator import Generator
from clan_cli.vars.prompt import PromptType
from clan_lib.dirs import find_toplevel
from clan_lib.errors import ClanError
@@ -241,16 +241,13 @@ def main() -> None:
raise ClanError(msg)
return prompt_values
with (
patch("clan_cli.vars.generate._ask_prompts", new=mocked_prompts),
NamedTemporaryFile("w") as f,
):
with NamedTemporaryFile("w") as f:
f.write("# created: 2023-07-17T10:51:45+02:00\n")
f.write(f"# public key: {sops_pub_key}\n")
f.write(sops_priv_key)
f.seek(0)
os.environ["SOPS_AGE_KEY_FILE"] = f.name
generate_vars(list(machines))
run_generators(list(machines), prompt_values=mocked_prompts)
if __name__ == "__main__":