diff --git a/pkgs/clan-cli/clan_cli/templates/__init__.py b/pkgs/clan-cli/clan_cli/templates/__init__.py index 1696b625f..8f0f175dc 100644 --- a/pkgs/clan-cli/clan_cli/templates/__init__.py +++ b/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) diff --git a/pkgs/clan-cli/clan_cli/templates/apply.py b/pkgs/clan-cli/clan_cli/templates/apply.py new file mode 100644 index 000000000..d5113853c --- /dev/null +++ b/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) diff --git a/pkgs/clan-cli/clan_cli/templates/apply_disk.py b/pkgs/clan-cli/clan_cli/templates/apply_disk.py new file mode 100644 index 000000000..c460ea64d --- /dev/null +++ b/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) diff --git a/pkgs/clan-cli/clan_lib/api/disk.py b/pkgs/clan-cli/clan_lib/api/disk.py index 755e10a5e..8afad3ffc 100644 --- a/pkgs/clan-cli/clan_lib/api/disk.py +++ b/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(): - msg = "Hardware configuration must exist before applying disk schema" - raise ClanError(msg) + if check_hw: + if not hw_config_path.exists(): + 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: - msg = "Hardware configuration must use type FACTER for applying disk schema automatically" - raise ClanError(msg) + if hw_config != HardwareConfig.NIXOS_FACTER: + msg = "Hardware configuration must use type FACTER for applying disk schema automatically" + 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 - if ph.options and placeholder_value not in ph.options: + # Checking invalid value: if the value is one of the provided 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, diff --git a/pkgs/clan-cli/clan_lib/machines/actions.py b/pkgs/clan-cli/clan_lib/machines/actions.py index 93c599a1b..2f903108d 100644 --- a/pkgs/clan-cli/clan_lib/machines/actions.py +++ b/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) diff --git a/pkgs/clan-cli/clan_lib/machines/hardware.py b/pkgs/clan-cli/clan_lib/machines/hardware.py index 08c50e0ac..6bf051eae 100644 --- a/pkgs/clan-cli/clan_lib/machines/hardware.py +++ b/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) diff --git a/pkgs/clan-cli/clan_lib/templates/filesystem.py b/pkgs/clan-cli/clan_lib/templates/filesystem.py index 1e18b57c0..aa2779f21 100644 --- a/pkgs/clan-cli/clan_lib/templates/filesystem.py +++ b/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(["chmod", "-R", "u+w", str(dest)]) + run(["cp", "-r", str(src / "."), str(dest)]) # Copy contents of src to dest + run( + ["chmod", "-R", "u+w", str(dest)] + ) # Ensure the destination is writable by the user