Merge pull request 'cli/templates: init apply disk' (#4306) from templates-cli into main

Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4306

pkgs/clan-cli/clan_cli/templates/__init__.py

@@ -1,6 +1,7 @@
 # !/usr/bin/env python3
 import argparse
 
+from .apply import register_apply_parser
 from .list import register_list_parser
 
 
@@ -13,4 +14,8 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
         required=True,
     )
     list_parser = subparser.add_parser("list", help="List avilable templates")
+    apply_parser = subparser.add_parser(
+        "apply", help="Apply a template of the specified type"
+    )
     register_list_parser(list_parser)
+    register_apply_parser(apply_parser)

pkgs/clan-cli/clan_cli/templates/apply.py

@@ -0,0 +1,15 @@
+import argparse
+
+from .apply_disk import register_apply_disk_template_parser
+
+
+def register_apply_parser(parser: argparse.ArgumentParser) -> None:
+    subparser = parser.add_subparsers(
+        title="template_type",
+        description="the template type to apply",
+        help="the template type to apply",
+        required=True,
+    )
+    disk_parser = subparser.add_parser("disk", help="Apply a disk template")
+
+    register_apply_disk_template_parser(disk_parser)

pkgs/clan-cli/clan_cli/templates/apply_disk.py

@@ -0,0 +1,78 @@
+import argparse
+import logging
+from collections.abc import Sequence
+from typing import Any
+
+from clan_lib.api.disk import set_machine_disk_schema
+from clan_lib.machines.machines import Machine
+
+log = logging.getLogger(__name__)
+
+
+class AppendSetAction(argparse.Action):
+    def __init__(self, option_strings: str, dest: str, **kwargs: Any) -> None:
+        super().__init__(option_strings, dest, **kwargs)
+
+    def __call__(
+        self,
+        parser: argparse.ArgumentParser,
+        namespace: argparse.Namespace,
+        values: str | Sequence[str] | None,
+        option_string: str | None = None,
+    ) -> None:
+        lst = getattr(namespace, self.dest)
+        assert isinstance(values, list), "values must be a list"
+        lst.append((values[0], values[1]))
+
+
+def apply_command(args: argparse.Namespace) -> None:
+    """Apply a disk template to a machine."""
+    set_tuples: list[tuple[str, str]] = args.set
+
+    placeholders = dict(set_tuples)
+
+    set_machine_disk_schema(
+        Machine(args.to_machine, args.flake),
+        args.template,
+        placeholders,
+        force=args.force,
+        check_hw=not args.skip_hardware_check,
+    )
+    log.info(f"Applied disk template '{args.template}' to machine '{args.to_machine}' ")
+
+
+def register_apply_disk_template_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--to-machine",
+        type=str,
+        required=True,
+        help="The machine to apply the template to",
+    )
+    parser.add_argument(
+        "--template",
+        type=str,
+        required=True,
+        help="The name of the disk template to apply",
+    )
+    parser.add_argument(
+        "--set",
+        help="Set a placeholder in the template to a value",
+        nargs=2,
+        metavar=("placeholder", "value"),
+        action=AppendSetAction,
+        default=[],
+    )
+    parser.add_argument(
+        "--force",
+        help="Force apply the template even if the machine already has a disk schema",
+        action="store_true",
+        default=False,
+    )
+    parser.add_argument(
+        "--skip-hardware-check",
+        help="Disables hardware checking. By default this command checks that the facter.json report exists and validates provided options",
+        action="store_true",
+        default=False,
+    )
+
+    parser.set_defaults(func=apply_command)

pkgs/clan-cli/clan_lib/api/disk.py

@@ -72,8 +72,18 @@ templates: dict[str, dict[str, Callable[[dict[str, Any]], Placeholder]]] = {
 }
 
 
+def get_empty_placeholder(label: str) -> Placeholder:
+    return Placeholder(
+        label,
+        options=None,
+        required=not label.endswith("*"),
+    )
+
+
 @API.register
-def get_machine_disk_schemas(machine: Machine) -> dict[str, DiskSchema]:
+def get_machine_disk_schemas(
+    machine: Machine, check_hw: bool = True
+) -> dict[str, DiskSchema]:
     """
     Get the available disk schemas.
     This function reads the disk schemas from the templates directory and returns them as a dictionary.
@@ -89,11 +99,13 @@ def get_machine_disk_schemas(machine: Machine) -> dict[str, DiskSchema]:
     hw_report = {}
 
     hw_report_path = HardwareConfig.NIXOS_FACTER.config_path(machine)
-    if not hw_report_path.exists():
+    if check_hw and not hw_report_path.exists():
         msg = "Hardware configuration missing"
         raise ClanError(msg)
-    with hw_report_path.open("r") as hw_report_file:
+
-        hw_report = json.load(hw_report_file)
+    if hw_report_path.exists():
+        with hw_report_path.open("r") as hw_report_file:
+            hw_report = json.load(hw_report_file)
 
     for disk_template in disk_templates.iterdir():
         if disk_template.is_dir():
@@ -109,7 +121,10 @@ def get_machine_disk_schemas(machine: Machine) -> dict[str, DiskSchema]:
             placeholders = {}
 
             if placeholder_getters:
-                placeholders = {k: v(hw_report) for k, v in placeholder_getters.items()}
+                placeholders = {
+                    k: v(hw_report) if hw_report else get_empty_placeholder(k)
+                    for k, v in placeholder_getters.items()
+                }
 
             raw_readme = (disk_template / "README.md").read_text()
             frontmatter, readme = extract_frontmatter(
@@ -139,30 +154,37 @@ def set_machine_disk_schema(
     # Use get disk schemas to get the placeholders and their options
     placeholders: dict[str, str],
     force: bool = False,
+    check_hw: bool = True,
 ) -> None:
     """
     Set the disk placeholders of the template
     """
+    # Ensure the machine exists
+    machine.get_inv_machine()
+
     # Assert the hw-config must exist before setting the disk
     hw_config = get_machine_hardware_config(machine)
     hw_config_path = hw_config.config_path(machine)
 
-    if not hw_config_path.exists():
+    if check_hw:
-        msg = "Hardware configuration must exist before applying disk schema"
+        if not hw_config_path.exists():
-        raise ClanError(msg)
+            msg = "Hardware configuration must exist for checking."
+            msg += f"\nrun 'clan machines update-hardware-config {machine.name}' to generate a hardware report. Alternatively disable hardware checking to skip this check"
+            raise ClanError(msg)
 
-    if hw_config != HardwareConfig.NIXOS_FACTER:
+        if hw_config != HardwareConfig.NIXOS_FACTER:
-        msg = "Hardware configuration must use type FACTER for applying disk schema automatically"
+            msg = "Hardware configuration must use type FACTER for applying disk schema automatically"
-        raise ClanError(msg)
+            raise ClanError(msg)
 
     disk_schema_path = clan_templates(TemplateType.DISK) / f"{schema_name}/default.nix"
 
     if not disk_schema_path.exists():
-        msg = f"Disk schema not found at {disk_schema_path}"
+        msg = f"Disk schema '{schema_name}' not found at {disk_schema_path}"
+        msg += f"\nAvailable schemas: {', '.join([p.name for p in clan_templates(TemplateType.DISK).iterdir()])}"
         raise ClanError(msg)
 
     # Check that the placeholders are valid
-    disk_schema = get_machine_disk_schemas(machine)[schema_name]
+    disk_schema = get_machine_disk_schemas(machine, check_hw)[schema_name]
     # check that all required placeholders are present
     for placeholder_name, schema_placeholder in disk_schema.placeholders.items():
         if schema_placeholder.required and placeholder_name not in placeholders:
@@ -183,12 +205,15 @@ def set_machine_disk_schema(
                 description=f"Available placeholders: {disk_schema.placeholders.keys()}",
             )
 
-        # Invalid value. Check if the value is one of the provided options
+        # Checking invalid value: if the value is one of the provided options
-        if ph.options and placeholder_value not in ph.options:
+        if check_hw and ph.options and placeholder_value not in ph.options:
             msg = (
                 f"Invalid value {placeholder_value} for placeholder {placeholder_name}"
             )
-            raise ClanError(msg, description=f"Valid options: {ph.options}")
+            raise ClanError(
+                msg,
+                description=f"Valid options: \n{'\n'.join(ph.options)}",
+            )
 
     placeholders_toml = "\n".join(
         [f"""# {k} = "{v}" """ for k, v in placeholders.items() if v is not None]
@@ -221,6 +246,9 @@ def set_machine_disk_schema(
             disk_config.write(header)
             disk_config.write(config_str)
 
+        # TODO: return files to commit
+        # Don't commit here
+        # The top level command will usually collect files and commit them in batches
         commit_file(
             disko_file_path,
             machine.flake.path,

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

@@ -65,14 +65,14 @@ def get_machine(flake: Flake, name: str) -> InventoryMachine:
         InventoryMachine: An instance representing the machine's inventory details.
 
     Raises:
-        ClanError: If the machine with the specified name is not found in the inventory.
+        ClanError: If the machine with the specified name is not found in the clan
     """
     inventory_store = InventoryStore(flake=flake)
     inventory = inventory_store.read()
 
     machine_inv = inventory.get("machines", {}).get(name)
     if machine_inv is None:
-        msg = f"Machine {name} not found in inventory"
+        msg = f"Machine {name} does not exist"
         raise ClanError(msg)
 
     return InventoryMachine(**machine_inv)

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

@@ -118,7 +118,7 @@ def run_machine_hardware_info(
     commit_file(
         hw_file,
         opts.machine.flake.path,
-        f"machines/{opts.machine}/{hw_file.name}: update hardware configuration",
+        f"machines/{opts.machine.name}/{hw_file.name}: update hardware configuration",
     )
     try:
         get_machine_target_platform(opts.machine)

pkgs/clan-cli/clan_lib/templates/filesystem.py

@@ -35,5 +35,7 @@ def copy_from_nixstore(src: Path, dest: Path) -> None:
     Uses `cp -r` to recursively copy the directory.
     Ensures the destination directory is writable by the user.
     """
-    run(["cp", "-r", str(src), str(dest)])
+    run(["cp", "-r", str(src / "."), str(dest)])  # Copy contents of src to dest
-    run(["chmod", "-R", "u+w", str(dest)])
+    run(
+        ["chmod", "-R", "u+w", str(dest)]
+    )  # Ensure the destination is writable by the user