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

Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4048
Reviewed-by: lassulus <clanlol@lassul.us>

formatter.nix

@@ -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

pkgs/clan-cli/clan_cli/__init__.py

@@ -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 str(flake_dir)
-    return Flake(arg)
+    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:

pkgs/clan-cli/clan_cli/flash/flash_cmd.py

@@ -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,
     )
 
 

pkgs/clan-cli/clan_cli/machines/install.py

@@ -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,

pkgs/clan-cli/clan_cli/machines/update.py

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

pkgs/clan-cli/clan_cli/tests/helpers/cli.py

@@ -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)

pkgs/clan-cli/clan_cli/vars/generate.py

@@ -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(

pkgs/clan-cli/clan_cli/vms/run.py

@@ -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(
+def build_vm(machine: Machine, tmpdir: Path) -> dict[str, str]:
-    machine: Machine, tmpdir: Path, nix_options: list[str] | None = None
-) -> 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",
+        "config.system.clan.vm.create", extra_config=facts_to_nixos_config(public_facts)
-        extra_config=facts_to_nixos_config(public_facts),
-        nix_options=nix_options,
     )
     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,

pkgs/clan-cli/clan_lib/flake/flake.py

@@ -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":
+    def from_json(
-        return cls(data["identifier"])
+        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 = self.nix_options if self.nix_options is not None else []
-            nix_options = []
 
         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(
+    def precache(self, selectors: list[str]) -> None:
-        self,
-        selectors: list[str],
-        nix_options: list[str] | None = None,
-    ) -> 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

pkgs/clan-cli/clan_lib/flake/flake_cache_test.py

@@ -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(
+        my_flake.select("testfile")
-            "testfile", nix_options=["--sandbox-build-dir", str(tmp_path / "build")]
-        )
     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)

pkgs/clan-cli/clan_lib/machines/install.py

@@ -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:

pkgs/clan-cli/clan_lib/machines/list.py

@@ -17,9 +17,7 @@ from clan_lib.nix_models.clan import InventoryMachine
 log = logging.getLogger(__name__)
 
 
-def list_full_machines(
+def list_full_machines(flake: Flake) -> dict[str, Machine]:
-    flake: Flake, nix_options: list[str] | None = None
-) -> 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(
+        machine = Machine(name=name, flake=flake)
-            name=name,
-            flake=flake,
-            nix_options=nix_options,
-        )
         res[machine.name] = machine
 
     return res

pkgs/clan-cli/clan_lib/machines/machines.py

@@ -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}',
+            f'clanInternals.machines."{system}"."{self.name}".{attr}'
-            nix_options=nix_options,
         )
 
-    def eval_nix(
+    def eval_nix(self, attr: str, extra_config: None | dict = None) -> Any:
-        self,
-        attr: str,
-        extra_config: None | dict = None,
-        nix_options: list[str] | None = 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:
+        return self.nix(attr)
-            nix_options = []
 
-        return self.nix(attr, nix_options)
+    def build_nix(self, attr: str, extra_config: None | dict = None) -> Path:
-
-    def build_nix(
-        self,
-        attr: str,
-        extra_config: None | dict = None,
-        nix_options: list[str] | None = 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:
+        output = self.nix(attr)
-            nix_options = []
-
-        output = self.nix(attr, nix_options)
         output = Path(output)
         if tmp_store := nix_test_store():
             output = tmp_store.joinpath(*output.parts[1:])

pkgs/clan-cli/clan_lib/machines/update.py

@@ -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}",
         ]

pkgs/clan-cli/clan_lib/persist/inventory_store.py

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

pkgs/generate-test-vars/generate_test_vars/cli.py

@@ -64,17 +64,11 @@ class TestMachine(Machine):
         return self.test_dir
 
     @override
-    def nix(
+    def nix(self, attr: str) -> Any:
-        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"]
@@ -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}',
+            f'checks."{test_system}".{self.check_attr}.machinesCross.{system}.{self.name}.{attr}'
-            nix_options=nix_options,
         )