diff --git a/pkgs/clan-cli/clan_cli/clan_uri.py b/pkgs/clan-cli/clan_cli/clan_uri.py index a7cf0f358..e4ca368eb 100644 --- a/pkgs/clan-cli/clan_cli/clan_uri.py +++ b/pkgs/clan-cli/clan_cli/clan_uri.py @@ -4,8 +4,6 @@ import urllib.request from dataclasses import dataclass from pathlib import Path -from .errors import ClanError - @dataclass class FlakeId: @@ -54,47 +52,21 @@ class FlakeId: return not self.is_local() +def _parse_url(comps: urllib.parse.ParseResult) -> FlakeId: + if comps.scheme == "" or "file" in comps.scheme: + res_p = Path(comps.path).expanduser().resolve() + flake_id = FlakeId(str(res_p)) + else: + flake_id = FlakeId(comps.geturl()) + return flake_id + + # Define the ClanURI class @dataclass class ClanURI: flake: FlakeId machine_name: str - # Initialize the class with a clan:// URI - def __init__(self, uri: str) -> None: - # users might copy whitespace along with the uri - uri = uri.strip() - self._orig_uri = uri - - # Check if the URI starts with clan:// - # If it does, remove the clan:// prefix - if uri.startswith("clan://"): - nested_uri = uri[7:] - else: - msg = f"Invalid uri: expected clan://, got {uri}" - raise ClanError(msg) - - # Parse the URI into components - # url://netloc/path;parameters?query#fragment - components: urllib.parse.ParseResult = urllib.parse.urlparse(nested_uri) - - # Replace the query string in the components with the new query string - clean_comps = components._replace(query=components.query, fragment="") - - # Parse the URL into a ClanUrl object - self.flake = self._parse_url(clean_comps) - self.machine_name = "defaultVM" - if components.fragment: - self.machine_name = components.fragment - - def _parse_url(self, comps: urllib.parse.ParseResult) -> FlakeId: - if comps.scheme == "" or "file" in comps.scheme: - res_p = Path(comps.path).expanduser().resolve() - flake_id = FlakeId(str(res_p)) - else: - flake_id = FlakeId(comps.geturl()) - return flake_id - def get_url(self) -> str: return str(self.flake) @@ -104,13 +76,31 @@ class ClanURI: url: str, machine_name: str | None = None, ) -> "ClanURI": - clan_uri = "" - if not url.startswith("clan://"): - clan_uri += "clan://" - - clan_uri += url + uri = url if machine_name: - clan_uri += f"#{machine_name}" + uri += f"#{machine_name}" - return cls(clan_uri) + # users might copy whitespace along with the uri + uri = uri.strip() + + # Check if the URI starts with clan:// + # If it does, remove the clan:// prefix + prefix = "clan://" + if uri.startswith(prefix): + uri = uri[len(prefix) :] + + # Parse the URI into components + # url://netloc/path;parameters?query#fragment + components: urllib.parse.ParseResult = urllib.parse.urlparse(uri) + + # Replace the query string in the components with the new query string + clean_comps = components._replace(query=components.query, fragment="") + + # Parse the URL into a ClanUrl object + flake = _parse_url(clean_comps) + machine_name = "defaultVM" + if components.fragment: + machine_name = components.fragment + + return cls(flake, machine_name) diff --git a/pkgs/clan-cli/clan_cli/flash/cli.py b/pkgs/clan-cli/clan_cli/flash/cli.py index 3b292863a..193e0c491 100644 --- a/pkgs/clan-cli/clan_cli/flash/cli.py +++ b/pkgs/clan-cli/clan_cli/flash/cli.py @@ -1,7 +1,7 @@ # !/usr/bin/env python3 import argparse -from .flash_command import register_flash_apply_parser +from .flash_cmd import register_flash_apply_parser from .list import register_flash_list_parser @@ -17,13 +17,8 @@ def register_parser(parser: argparse.ArgumentParser) -> None: apply_parser = subparser.add_parser( "apply", help="Flash a machine", - formatter_class=argparse.RawTextHelpFormatter, ) register_flash_apply_parser(apply_parser) - list_parser = subparser.add_parser( - "list", - help="List options", - formatter_class=argparse.RawTextHelpFormatter, - ) + list_parser = subparser.add_parser("list", help="List options") register_flash_list_parser(list_parser) diff --git a/pkgs/clan-cli/clan_cli/flash/flash_command.py b/pkgs/clan-cli/clan_cli/flash/flash_cmd.py similarity index 100% rename from pkgs/clan-cli/clan_cli/flash/flash_command.py rename to pkgs/clan-cli/clan_cli/flash/flash_cmd.py diff --git a/pkgs/clan-cli/clan_cli/history/add.py b/pkgs/clan-cli/clan_cli/history/add.py index d0235485c..ea4f055b0 100644 --- a/pkgs/clan-cli/clan_cli/history/add.py +++ b/pkgs/clan-cli/clan_cli/history/add.py @@ -111,7 +111,9 @@ def add_history_command(args: argparse.Namespace) -> None: # takes a (sub)parser and configures it def register_add_parser(parser: argparse.ArgumentParser) -> None: - parser.add_argument("uri", type=ClanURI, help="Path to the flake", default=".") + parser.add_argument( + "uri", type=ClanURI.from_str, help="Path to the flake", default="." + ) parser.add_argument( "--all", help="Add all machines", default=False, action="store_true" ) diff --git a/pkgs/clan-cli/clan_cli/machines/cli.py b/pkgs/clan-cli/clan_cli/machines/cli.py index c6b0631e1..e371317e8 100644 --- a/pkgs/clan-cli/clan_cli/machines/cli.py +++ b/pkgs/clan-cli/clan_cli/machines/cli.py @@ -4,6 +4,7 @@ import argparse from .create import register_create_parser from .delete import register_delete_parser from .hardware import register_hw_generate +from .import_cmd import register_import_parser from .install import register_install_parser from .list import register_list_parser from .update import register_update_parser @@ -113,3 +114,22 @@ For more detailed information, visit: https://docs.clan.lol/getting-started/depl formatter_class=argparse.RawTextHelpFormatter, ) register_install_parser(install_parser) + + import_parser = subparser.add_parser( + "import", + help="Import a machine", + description="Import a template machine from a local or remote source.", + epilog=( + """ + +Examples: + $ clan machines import flash-installer + Will import a machine from the flash-installer template from clan-core. + + $ clan machines import flash-installer --src https://git.clan.lol/clan/clan-core + Will import a machine from the flash-installer template from clan-core but with the source set to the provided URL. + """ + ), + formatter_class=argparse.RawTextHelpFormatter, + ) + register_import_parser(import_parser) diff --git a/pkgs/clan-cli/clan_cli/machines/import_cmd.py b/pkgs/clan-cli/clan_cli/machines/import_cmd.py new file mode 100644 index 000000000..4c32771b2 --- /dev/null +++ b/pkgs/clan-cli/clan_cli/machines/import_cmd.py @@ -0,0 +1,160 @@ +import argparse +import logging +import shutil +from dataclasses import dataclass +from pathlib import Path +from tempfile import TemporaryDirectory + +from clan_cli.api import API +from clan_cli.clan.create import git_command +from clan_cli.clan_uri import FlakeId +from clan_cli.cmd import Log, run +from clan_cli.dirs import clan_templates, get_clan_flake_toplevel_or_env +from clan_cli.errors import ClanError +from clan_cli.machines.list import list_nixos_machines +from clan_cli.machines.machines import Machine +from clan_cli.nix import nix_command + +log = logging.getLogger(__name__) + + +def validate_directory(root_dir: Path, machine_name: str) -> None: + machines_dir = root_dir / "machines" / machine_name + for root, _, files in root_dir.walk(): + for file in files: + file_path = Path(root) / file + if not file_path.is_relative_to(machines_dir): + msg = f"File {file_path} is not in the 'machines/{machine_name}' directory." + description = "Template machines are only allowed to contain files in the 'machines/{machine_name}' directory." + raise ClanError(msg, description=description) + + +@dataclass +class ImportOptions: + target: FlakeId + src: Machine + rename: str | None = None + + +@API.register +def import_machine(opts: ImportOptions) -> None: + if not opts.target.is_local(): + msg = f"Clan {opts.target} is not a local clan." + description = "Import machine only works on local clans" + raise ClanError(msg, description=description) + clan_dir = opts.target.path + + log.debug(f"Importing machine '{opts.src.name}' from {opts.src.flake}") + + if opts.src.name == opts.rename: + msg = "Rename option must be different from the template machine name" + raise ClanError(msg) + + if opts.src.name in list_nixos_machines(clan_dir) and not opts.rename: + msg = f"{opts.src.name} is already defined in {clan_dir}" + description = ( + "Please add the --rename option to import the machine with a different name" + ) + raise ClanError(msg, description=description) + + machine_name = opts.src.name if not opts.rename else opts.rename + dst = clan_dir / "machines" / machine_name + + if dst.exists(): + msg = f"Machine {machine_name} already exists in {clan_dir}" + description = ( + "Please delete the existing machine or import with a different name" + ) + raise ClanError(msg, description=description) + + with TemporaryDirectory() as tmpdir: + tmpdirp = Path(tmpdir) + command = nix_command( + [ + "flake", + "init", + "-t", + opts.src.get_id(), + ] + ) + run(command, log=Log.NONE, cwd=tmpdirp) + + validate_directory(tmpdirp, opts.src.name) + src = tmpdirp / "machines" / opts.src.name + + if ( + not (src / "configuration.nix").exists() + and not (src / "inventory.json").exists() + ): + msg = f"Template machine {opts.src.name} does not contain a configuration.nix or inventory.json" + description = ( + "Template machine must contain a configuration.nix or inventory.json" + ) + raise ClanError(msg, description=description) + + def log_copy(src: str, dst: str) -> None: + relative_dst = dst.replace(f"{clan_dir}/", "") + log.info(f"Add file: {relative_dst}") + shutil.copy2(src, dst) + + shutil.copytree(src, dst, ignore_dangling_symlinks=True, copy_function=log_copy) + + run(git_command(clan_dir, "add", f"machines/{machine_name}"), cwd=clan_dir) + + if (dst / "inventory.json").exists(): + # TODO: Implement inventory import + msg = "Inventory import not implemented yet" + raise NotImplementedError(msg) + # inventory = load_inventory_json(clan_dir) + + # inventory.machines[machine_name] = Inventory_Machine( + # name=machine_name, + # deploy=MachineDeploy(targetHost=None), + # ) + # set_inventory(inventory, clan_dir, "Imported machine from template") + + +def import_command(args: argparse.Namespace) -> None: + if args.flake: + target = args.flake + else: + tmp = get_clan_flake_toplevel_or_env() + target = FlakeId(str(tmp)) if tmp else None + + if not target: + msg = "No clan found." + description = ( + "Run this command in a clan directory or specify the --flake option" + ) + raise ClanError(msg, description=description) + + src_uri = args.src + if not src_uri: + src_uri = FlakeId(str(clan_templates())) + + opts = ImportOptions( + target=target, + src=Machine(flake=src_uri, name=args.machine_name), + rename=args.rename, + ) + import_machine(opts) + + +def register_import_parser(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "machine_name", + type=str, + help="The name of the machine to import", + ) + parser.add_argument( + "--src", + type=FlakeId, + help="The source flake to import the machine from", + ) + parser.add_argument( + "--rename", + type=str, + help="Rename the imported machine", + ) + + parser.set_defaults(func=import_command) diff --git a/pkgs/clan-cli/tests/test_clan_uri.py b/pkgs/clan-cli/tests/test_clan_uri.py index e449ff7d2..7acb855d1 100644 --- a/pkgs/clan-cli/tests/test_clan_uri.py +++ b/pkgs/clan-cli/tests/test_clan_uri.py @@ -5,46 +5,46 @@ from clan_cli.clan_uri import ClanURI def test_get_url() -> None: # Create a ClanURI object from a remote URI with parameters - uri = ClanURI("clan://https://example.com?password=1234#myVM") + uri = ClanURI.from_str("clan://https://example.com?password=1234#myVM") assert uri.get_url() == "https://example.com?password=1234" - uri = ClanURI("clan://~/Downloads") + uri = ClanURI.from_str("clan://~/Downloads") assert uri.get_url().endswith("/Downloads") - uri = ClanURI("clan:///home/user/Downloads") + uri = ClanURI.from_str("clan:///home/user/Downloads") assert uri.get_url() == "/home/user/Downloads" - uri = ClanURI("clan://file:///home/user/Downloads") + uri = ClanURI.from_str("clan://file:///home/user/Downloads") assert uri.get_url() == "/home/user/Downloads" def test_local_uri() -> None: # Create a ClanURI object from a local URI - uri = ClanURI("clan://file:///home/user/Downloads") + uri = ClanURI.from_str("clan://file:///home/user/Downloads") assert uri.flake.path == Path("/home/user/Downloads") def test_is_remote() -> None: # Create a ClanURI object from a remote URI - uri = ClanURI("clan://https://example.com") + uri = ClanURI.from_str("clan://https://example.com") assert uri.flake.url == "https://example.com" def test_direct_local_path() -> None: # Create a ClanURI object from a remote URI - uri = ClanURI("clan://~/Downloads") + uri = ClanURI.from_str("clan://~/Downloads") assert uri.get_url().endswith("/Downloads") def test_direct_local_path2() -> None: # Create a ClanURI object from a remote URI - uri = ClanURI("clan:///home/user/Downloads") + uri = ClanURI.from_str("clan:///home/user/Downloads") assert uri.get_url() == "/home/user/Downloads" def test_remote_with_clanparams() -> None: # Create a ClanURI object from a remote URI with parameters - uri = ClanURI("clan://https://example.com") + uri = ClanURI.from_str("clan://https://example.com") assert uri.machine_name == "defaultVM" assert uri.flake.url == "https://example.com" diff --git a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py index fc5dafe00..e3a9c6ac8 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py @@ -65,7 +65,7 @@ class ClanStore: self._emitter.connect(signal, cb) def set_logging_vm(self, ident: str) -> VMObject | None: - vm = self.get_vm(ClanURI(f"clan://{ident}")) + vm = self.get_vm(ClanURI.from_str(f"clan://{ident}")) if vm is not None: self._logging_vm = vm diff --git a/pkgs/clan-vm-manager/clan_vm_manager/views/list.py b/pkgs/clan-vm-manager/clan_vm_manager/views/list.py index 9f3a2d9ef..762010fcc 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/views/list.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/views/list.py @@ -335,7 +335,7 @@ class ClanList(Gtk.Box): def on_join_request(self, source: Any, url: str) -> None: log.debug("Join request: %s", url) - clan_uri = ClanURI(url) + clan_uri = ClanURI.from_str(url) JoinList.use().push(clan_uri, self.on_after_join) def on_after_join(self, source: JoinValue) -> None: diff --git a/templates/flake.nix b/templates/flake.nix index 4ad730b8f..d6eff58d6 100644 --- a/templates/flake.nix +++ b/templates/flake.nix @@ -15,6 +15,10 @@ description = "Flake-parts"; path = ./flake-parts; }; + flash-installer = { + description = "Flash installer"; + path = ./flash-installer; + }; }; }; } diff --git a/templates/flash-installer/machines/flash-installer/configuration.nix b/templates/flash-installer/machines/flash-installer/configuration.nix new file mode 100644 index 000000000..0d8450348 --- /dev/null +++ b/templates/flash-installer/machines/flash-installer/configuration.nix @@ -0,0 +1,7 @@ +{ + imports = [ + + ]; + + # Hello World +}