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

Merge pull request 'clan-lib: Move nix_options from Machine class to Flake class' (#4048) from Qubasa/clan-core:move_nix_options into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4048
Reviewed-by: lassulus <clanlol@lassul.us>
This commit is contained in:
Luis Hebendanz
2025-06-24 17:19:16 +00:00
parent 96cf48a8a8 60d2837ddd
commit 81b87a6437
16 changed files with 77 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
formatter.nix
View File

@@ -106,6 +106,13 @@
extraPythonPackages = (self'.packages.clan-app.devshellPyDeps pkgs.python3Packages);
extraPythonPaths = [ "../../clan-cli" ];
};
"generate-test-vars" = {
directory = "pkgs/generate-test-vars";
extraPythonPackages = [
(pkgs.python3.withPackages (ps: self'.packages.clan-cli.devshellPyDeps ps))
];
extraPythonPaths = [ "../clan-cli" ];
};
}
// (
if pkgs.stdenv.isLinux then

21
pkgs/clan-cli/clan_cli/__init__.py
View File

@@ -35,20 +35,27 @@ with contextlib.suppress(ImportError):
import argcomplete # type: ignore[no-redef]
def flake_path(arg: str) -> Flake:
def flake_path(arg: str) -> str:
flake_dir = Path(arg).resolve()
if flake_dir.exists() and flake_dir.is_dir():
return Flake(str(flake_dir))
return Flake(arg)
return str(flake_dir)
return arg
def default_flake() -> Flake | None:
def default_flake() -> str | None:
val = get_clan_flake_toplevel_or_env()
if val:
return Flake(str(val))
return str(val)
return None
def create_flake_from_args(args: argparse.Namespace) -> Flake:
"""Create a Flake object from parsed arguments, including nix_options."""
flake_path_str = args.flake
nix_options = getattr(args, "option", [])
return Flake(flake_path_str, nix_options=nix_options)
def add_common_flags(parser: argparse.ArgumentParser) -> None:
def argument_exists(parser: argparse.ArgumentParser, arg: str) -> bool:
"""
@@ -450,6 +457,10 @@ def main() -> None:
if not hasattr(args, "func"):
return
# Convert flake path to Flake object with nix_options if flake argument exists
if hasattr(args, "flake") and args.flake is not None:
args.flake = create_flake_from_args(args)
try:
args.func(args)
except ClanError as e:

3
pkgs/clan-cli/clan_cli/flash/flash_cmd.py
View File

@@ -26,7 +26,6 @@ class FlashOptions:
debug: bool
mode: str
write_efi_boot_entries: bool
nix_options: list[str]
system_config: SystemConfig
@@ -72,7 +71,6 @@ def flash_command(args: argparse.Namespace) -> None:
ssh_keys_path=args.ssh_pubkey,
),
write_efi_boot_entries=args.write_efi_boot_entries,
nix_options=args.option,
)
machine = Machine(opts.machine, flake=opts.flake)
@@ -94,7 +92,6 @@ def flash_command(args: argparse.Namespace) -> None:
dry_run=opts.dry_run,
debug=opts.debug,
write_efi_boot_entries=opts.write_efi_boot_entries,
extra_args=opts.nix_options,
)

3
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -42,7 +42,7 @@ def install_command(args: argparse.Namespace) -> None:
else:
password = None
machine = Machine(name=args.machine, flake=args.flake, nix_options=args.option)
machine = Machine(name=args.machine, flake=args.flake)
host_key_check = args.host_key_check
if target_host_str is not None:
@@ -72,7 +72,6 @@ def install_command(args: argparse.Namespace) -> None:
phases=args.phases,
debug=args.debug,
no_reboot=args.no_reboot,
nix_options=args.option,
build_on=BuildOn(args.build_on) if args.build_on is not None else None,
update_hardware_config=HardwareConfig(args.update_hardware_config),
password=password,

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

@@ -46,9 +46,7 @@ def update_command(args: argparse.Namespace) -> None:
raise ClanError(msg)
for machine_name in selected_machines:
machine = Machine(
name=machine_name, flake=args.flake, nix_options=args.option
)
machine = Machine(name=machine_name, flake=args.flake)
machines.append(machine)
if args.target_host is not None and len(machines) > 1:

6
pkgs/clan-cli/clan_cli/tests/helpers/cli.py
View File

@@ -2,7 +2,7 @@ import argparse
import logging
import shlex
from clan_cli import create_parser
from clan_cli import create_flake_from_args, create_parser
from clan_lib.custom_logger import print_trace
log = logging.getLogger(__name__)
@@ -13,6 +13,10 @@ def run(args: list[str]) -> argparse.Namespace:
parsed = parser.parse_args(args)
cmd = shlex.join(["clan", *args])
# Convert flake path to Flake object with nix_options if flake argument exists
if hasattr(parsed, "flake") and parsed.flake is not None:
parsed.flake = create_flake_from_args(parsed)
print_trace(f"$ {cmd}", log, "localhost")
if hasattr(parsed, "func"):
parsed.func(parsed)

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

@@ -511,7 +511,7 @@ def generate_command(args: argparse.Namespace) -> None:
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
machines: list[Machine] = list(list_full_machines(args.flake, args.option).values())
machines: list[Machine] = list(list_full_machines(args.flake).values())
if len(args.machines) > 0:
machines = list(

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

@@ -49,20 +49,15 @@ def facts_to_nixos_config(facts: dict[str, dict[str, bytes]]) -> dict:
# TODO move this to the Machines class
def build_vm(
machine: Machine, tmpdir: Path, nix_options: list[str] | None = None
) -> dict[str, str]:
def build_vm(machine: Machine, tmpdir: Path) -> dict[str, str]:
# TODO pass prompt here for the GTK gui
if nix_options is None:
nix_options = []
secrets_dir = get_secrets(machine, tmpdir)
public_facts = machine.public_facts_store.get_all()
nixos_config_file = machine.build_nix(
"config.system.clan.vm.create",
extra_config=facts_to_nixos_config(public_facts),
nix_options=nix_options,
"config.system.clan.vm.create", extra_config=facts_to_nixos_config(public_facts)
)
try:
vm_data = json.loads(Path(nixos_config_file).read_text())
@@ -204,7 +199,6 @@ def spawn_vm(
*,
cachedir: Path | None = None,
socketdir: Path | None = None,
nix_options: list[str] | None = None,
portmap: dict[int, int] | None = None,
stdout: int | None = None,
stderr: int | None = None,
@@ -212,8 +206,7 @@ def spawn_vm(
) -> Iterator[QemuVm]:
if portmap is None:
portmap = {}
if nix_options is None:
nix_options = []
with ExitStack() as stack:
machine = Machine(name=vm.machine_name, flake=vm.flake_url)
machine.debug(f"Creating VM for {machine}")
@@ -234,7 +227,7 @@ def spawn_vm(
socketdir = Path(socket_tmp)
# TODO: We should get this from the vm argument
nixos_config = build_vm(machine, cachedir, nix_options)
nixos_config = build_vm(machine, cachedir)
state_dir = vm_state_dir(vm.flake_url.identifier, machine.name)
state_dir.mkdir(parents=True, exist_ok=True)
@@ -321,7 +314,6 @@ def spawn_vm(
class RuntimeConfig:
cachedir: Path | None = None
socketdir: Path | None = None
nix_options: list[str] | None = None
portmap: dict[int, int] | None = None
command: list[str] | None = None
no_block: bool = False
@@ -339,7 +331,6 @@ def run_vm(
vm_config,
cachedir=runtime_config.cachedir,
socketdir=runtime_config.socketdir,
nix_options=runtime_config.nix_options,
portmap=runtime_config.portmap,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
@@ -395,7 +386,6 @@ def run_command(
portmap = dict(p.split(":") for p in args.publish)
runtime_config = RuntimeConfig(
nix_options=args.option,
portmap=portmap,
command=args.command,
no_block=args.no_block,

37
pkgs/clan-cli/clan_lib/flake/flake.py
View File

@@ -576,6 +576,7 @@ class Flake:
identifier: str
hash: str | None = None
store_path: str | None = None
nix_options: list[str] | None = None
_flake_cache_path: Path | None = field(init=False, default=None)
_cache: FlakeCache | None = field(init=False, default=None)
@@ -583,8 +584,13 @@ class Flake:
_is_local: bool | None = field(init=False, default=None)
@classmethod
def from_json(cls: type["Flake"], data: dict[str, Any]) -> "Flake":
return cls(data["identifier"])
def from_json(
cls: type["Flake"],
data: dict[str, Any],
*,
nix_options: list[str] | None = None,
) -> "Flake":
return cls(data["identifier"], nix_options=nix_options)
def __str__(self) -> str:
return self.identifier
@@ -632,10 +638,14 @@ class Flake:
nix_command,
)
if self.nix_options is None:
self.nix_options = []
cmd = [
"flake",
"prefetch",
"--json",
*self.nix_options,
"--option",
"flake-registry",
"",
@@ -690,7 +700,6 @@ class Flake:
def get_from_nix(
self,
selectors: list[str],
nix_options: list[str] | None = None,
apply: str = "v: v",
) -> None:
"""
@@ -722,8 +731,7 @@ class Flake:
self.invalidate_cache()
assert self._cache is not None
if nix_options is None:
nix_options = []
nix_options = self.nix_options if self.nix_options is not None else []
str_selectors: list[str] = []
for selector in selectors:
@@ -736,7 +744,9 @@ class Flake:
# method to getting the NAR hash
fallback_nixpkgs_hash = "@fallback_nixpkgs_hash@"
if not fallback_nixpkgs_hash.startswith("sha256-"):
fallback_nixpkgs = Flake(str(nixpkgs_source()))
fallback_nixpkgs = Flake(
str(nixpkgs_source()), nix_options=self.nix_options
)
fallback_nixpkgs.invalidate_cache()
assert fallback_nixpkgs.hash is not None, (
"this should be impossible as invalidate_cache() should always set `hash`"
@@ -745,7 +755,7 @@ class Flake:
select_hash = "@select_hash@"
if not select_hash.startswith("sha256-"):
select_flake = Flake(str(select_source()))
select_flake = Flake(str(select_source()), nix_options=self.nix_options)
select_flake.invalidate_cache()
assert select_flake.hash is not None, (
"this should be impossible as invalidate_cache() should always set `hash`"
@@ -808,11 +818,7 @@ class Flake:
if self.flake_cache_path:
self._cache.save_to_file(self.flake_cache_path)
def precache(
self,
selectors: list[str],
nix_options: list[str] | None = None,
) -> None:
def precache(self, selectors: list[str]) -> None:
"""
Ensures that the specified selectors are cached locally.
@@ -822,7 +828,6 @@ class Flake:
Args:
selectors (list[str]): A list of attribute selectors to check and cache.
nix_options (list[str] | None): Optional additional options to pass to the Nix build command.
"""
if self._cache is None:
self.invalidate_cache()
@@ -833,12 +838,11 @@ class Flake:
if not self._cache.is_cached(selector):
not_fetched_selectors.append(selector)
if not_fetched_selectors:
self.get_from_nix(not_fetched_selectors, nix_options)
self.get_from_nix(not_fetched_selectors)
def select(
self,
selector: str,
nix_options: list[str] | None = None,
apply: str = "v: v",
) -> Any:
"""
@@ -847,7 +851,6 @@ class Flake:
Args:
selector (str): The attribute selector string to fetch the value for.
nix_options (list[str] | None): Optional additional options to pass to the Nix build command.
"""
if self._cache is None:
self.invalidate_cache()
@@ -856,6 +859,6 @@ class Flake:
if not self._cache.is_cached(selector):
log.debug(f"Cache miss for {selector}")
self.get_from_nix([selector], nix_options, apply=apply)
self.get_from_nix([selector], apply=apply)
value = self._cache.select(selector)
return value

9
pkgs/clan-cli/clan_lib/flake/flake_cache_test.py
View File

@@ -284,13 +284,14 @@ def test_cache_gc(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
}
""")
my_flake = Flake(str(tmp_path / "flake"))
my_flake = Flake(
str(tmp_path / "flake"),
nix_options=["--sandbox-build-dir", str(tmp_path / "build")],
)
if platform == "darwin":
my_flake.select("testfile")
else:
my_flake.select(
"testfile", nix_options=["--sandbox-build-dir", str(tmp_path / "build")]
)
my_flake.select("testfile")
assert my_flake._cache is not None # noqa: SLF001
assert my_flake._cache.is_cached("testfile") # noqa: SLF001
subprocess.run(["nix-collect-garbage"], check=True)

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

@@ -1,6 +1,6 @@
import logging
import os
from dataclasses import dataclass, field
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from tempfile import TemporaryDirectory
@@ -32,7 +32,6 @@ class InstallOptions:
no_reboot: bool = False
phases: str | None = None
build_on: BuildOn | None = None
nix_options: list[str] = field(default_factory=list)
update_hardware_config: HardwareConfig = HardwareConfig.NONE
password: str | None = None
identity_file: Path | None = None
@@ -127,7 +126,7 @@ def install_machine(opts: InstallOptions, target_host: Remote) -> None:
cmd.append("--debug")
# Add nix options to nixos-anywhere
cmd.extend(opts.nix_options)
cmd.extend(opts.machine.flake.nix_options or [])
cmd.append(target_host.target)
if opts.use_tor:

13
pkgs/clan-cli/clan_lib/machines/list.py
View File

@@ -17,9 +17,7 @@ from clan_lib.nix_models.clan import InventoryMachine
log = logging.getLogger(__name__)
def list_full_machines(
flake: Flake, nix_options: list[str] | None = None
) -> dict[str, Machine]:
def list_full_machines(flake: Flake) -> dict[str, Machine]:
"""
Like `list_machines`, but returns a full 'machine' instance for each machine.
"""
@@ -27,9 +25,6 @@ def list_full_machines(
res: dict[str, Machine] = {}
if nix_options is None:
nix_options = []
for inv_machine in machines.values():
name = inv_machine.get("name")
# Technically, this should not happen, but we are defensive here.
@@ -37,11 +32,7 @@ def list_full_machines(
msg = "InternalError: Machine name is required. But got a machine without a name."
raise ClanError(msg)
machine = Machine(
name=name,
flake=flake,
nix_options=nix_options,
)
machine = Machine(name=name, flake=flake)
res[machine.name] = machine
return res

34
pkgs/clan-cli/clan_lib/machines/machines.py
View File

@@ -2,7 +2,7 @@ import importlib
import json
import logging
import re
from dataclasses import dataclass, field
from dataclasses import dataclass
from functools import cached_property
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal
@@ -30,8 +30,6 @@ class Machine:
name: str
flake: Flake
nix_options: list[str] = field(default_factory=list)
def get_inv_machine(self) -> "InventoryMachine":
return get_machine(self.flake, self.name)
@@ -164,29 +162,20 @@ class Machine:
def nix(
self,
attr: str,
nix_options: list[str] | None = None,
) -> Any:
"""
Build the machine and return the path to the result
accepts a secret store and a facts store # TODO
"""
if nix_options is None:
nix_options = []
config = nix_config()
system = config["system"]
return self.flake.select(
f'clanInternals.machines."{system}"."{self.name}".{attr}',
nix_options=nix_options,
f'clanInternals.machines."{system}"."{self.name}".{attr}'
)
def eval_nix(
self,
attr: str,
extra_config: None | dict = None,
nix_options: list[str] | None = None,
) -> Any:
def eval_nix(self, attr: str, extra_config: None | dict = None) -> Any:
"""
eval a nix attribute of the machine
@attr: the attribute to get
@@ -195,17 +184,9 @@ class Machine:
if extra_config:
log.warning("extra_config in eval_nix is no longer supported")
if nix_options is None:
nix_options = []
return self.nix(attr)
return self.nix(attr, nix_options)
def build_nix(
self,
attr: str,
extra_config: None | dict = None,
nix_options: list[str] | None = None,
) -> Path:
def build_nix(self, attr: str, extra_config: None | dict = None) -> Path:
"""
build a nix attribute of the machine
@attr: the attribute to get
@@ -214,10 +195,7 @@ class Machine:
if extra_config:
log.warning("extra_config in build_nix is no longer supported")
if nix_options is None:
nix_options = []
output = self.nix(attr, nix_options)
output = self.nix(attr)
output = Path(output)
if tmp_store := nix_test_store():
output = tmp_store.joinpath(*output.parts[1:])

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

@@ -124,6 +124,8 @@ def deploy_machine(
path = upload_sources(machine, sudo_host)
nix_options = machine.flake.nix_options if machine.flake.nix_options else []
nix_options = [
"--show-trace",
"--option",
@@ -133,7 +135,7 @@ def deploy_machine(
"accept-flake-config",
"true",
"-L",
*machine.nix_options,
*nix_options,
"--flake",
f"{path}#{machine.name}",
]

6
pkgs/clan-cli/clan_lib/persist/inventory_store.py
View File

@@ -86,11 +86,7 @@ class WriteInfo:
class FlakeInterface(Protocol):
def select(
self,
selector: str,
nix_options: list[str] | None = None,
) -> Any: ...
def select(self, selector: str) -> Any: ...
def invalidate_cache(self) -> None: ...

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

@@ -64,17 +64,11 @@ class TestMachine(Machine):
return self.test_dir
@override
def nix(
self,
attr: str,
nix_options: list[str] | None = None,
) -> Any:
def nix(self, attr: str) -> Any:
"""
Build the machine and return the path to the result
accepts a secret store and a facts store # TODO
"""
if nix_options is None:
nix_options = []
config = nix_config()
system = config["system"]
@@ -83,8 +77,7 @@ class TestMachine(Machine):
test_system = system.rstrip("darwin") + "linux"
return self.flake.select(
f'checks."{test_system}".{self.check_attr}.machinesCross.{system}.{self.name}.{attr}',
nix_options=nix_options,
f'checks."{test_system}".{self.check_attr}.machinesCross.{system}.{self.name}.{attr}'
)