Merge pull request 'pkgs/clan: Further unify clan flake validation' (#4358) from kenji/ke-non-clan-commands into main

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

pkgs/clan-cli/clan_cli/backups/create.py

@@ -2,7 +2,7 @@ import argparse
 import logging
 
 from clan_lib.backups.create import create_backup
-from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.machines.machines import Machine
 
 from clan_cli.completions import (
@@ -15,10 +15,8 @@ log = logging.getLogger(__name__)
 
 
 def create_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    machine = Machine(name=args.machine, flake=args.flake)
+    flake = require_flake(args.flake)
+    machine = Machine(name=args.machine, flake=flake)
     create_backup(machine=machine, provider=args.provider)
 
 

pkgs/clan-cli/clan_cli/backups/create_test.py

@@ -0,0 +1,15 @@
+from pathlib import Path
+
+import pytest
+from clan_lib.errors import ClanError
+
+from clan_cli.tests.helpers import cli
+
+
+def test_create_command_no_flake(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    with pytest.raises(ClanError):
+        cli.run(["backups", "create", "machine1"])

pkgs/clan-cli/clan_cli/backups/list.py

@@ -1,7 +1,7 @@
 import argparse
 
 from clan_lib.backups.list import list_backups
-from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.machines.machines import Machine
 
 from clan_cli.completions import (
@@ -12,11 +12,8 @@ from clan_cli.completions import (
 
 
 def list_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-
-    machine = Machine(name=args.machine, flake=args.flake)
+    flake = require_flake(args.flake)
+    machine = Machine(name=args.machine, flake=flake)
     backups = list_backups(machine=machine, provider=args.provider)
     for backup in backups:
         print(backup.name)

pkgs/clan-cli/clan_cli/backups/list_test.py

@@ -0,0 +1,13 @@
+from pathlib import Path
+
+import pytest
+from clan_lib.errors import ClanError
+
+from clan_cli.tests.helpers import cli
+
+
+def test_list_command_no_flake(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    with pytest.raises(ClanError):
+        cli.run(["backups", "list", "machine1"])

pkgs/clan-cli/clan_cli/backups/restore.py

@@ -1,7 +1,7 @@
 import argparse
 
 from clan_lib.backups.restore import restore_backup
-from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.machines.machines import Machine
 
 from clan_cli.completions import (
@@ -12,10 +12,8 @@ from clan_cli.completions import (
 
 
 def restore_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    machine = Machine(name=args.machine, flake=args.flake)
+    flake = require_flake(args.flake)
+    machine = Machine(name=args.machine, flake=flake)
     restore_backup(
         machine=machine,
         provider=args.provider,

pkgs/clan-cli/clan_cli/backups/restore_test.py

@@ -0,0 +1,15 @@
+from pathlib import Path
+
+import pytest
+from clan_lib.errors import ClanError
+
+from clan_cli.tests.helpers import cli
+
+
+def test_restore_command_no_flake(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    with pytest.raises(ClanError):
+        cli.run(["backups", "restore", "machine1", "provider1", "backup1"])

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

@@ -9,6 +9,7 @@ from tempfile import TemporaryDirectory
 
 from clan_lib.cmd import RunOpts, run
 from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.git import commit_files
 from clan_lib.machines.list import list_full_machines
 from clan_lib.machines.machines import Machine
@@ -223,11 +224,8 @@ def generate_facts(
 
 
 def generate_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-
-    machines: list[Machine] = list(list_full_machines(args.flake).values())
+    flake = require_flake(args.flake)
+    machines: list[Machine] = list(list_full_machines(flake).values())
     if len(args.machines) > 0:
         machines = list(
             filter(

pkgs/clan-cli/clan_cli/facts/generate_test.py

@@ -0,0 +1,15 @@
+from pathlib import Path
+
+import pytest
+from clan_lib.errors import ClanError
+
+from clan_cli.tests.helpers import cli
+
+
+def test_generate_command_no_flake(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    with pytest.raises(ClanError):
+        cli.run(["facts", "generate"])

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

@@ -4,6 +4,7 @@ import sys
 from pathlib import Path
 
 from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.machines.install import BuildOn, InstallOptions, run_machine_install
 from clan_lib.machines.machines import Machine
 from clan_lib.ssh.remote import Remote
@@ -21,6 +22,7 @@ log = logging.getLogger(__name__)
 
 def install_command(args: argparse.Namespace) -> None:
     try:
+        flake = require_flake(args.flake)
         # Only if the caller did not specify a target_host via args.target_host
         # Find a suitable target_host that is reachable
         target_host_str = args.target_host
@@ -44,7 +46,7 @@ def install_command(args: argparse.Namespace) -> None:
         else:
             password = None
 
-        machine = Machine(name=args.machine, flake=args.flake)
+        machine = Machine(name=args.machine, flake=flake)
         host_key_check = args.host_key_check
 
         if target_host_str is not None:
@@ -58,10 +60,6 @@ def install_command(args: argparse.Namespace) -> None:
             msg = "Installing macOS machines is not yet supported"
             raise ClanError(msg)
 
-        if args.flake is None:
-            msg = "Could not find clan flake toplevel directory"
-            raise ClanError(msg)
-
         if not args.yes:
             ask = input(f"Install {args.machine} to {target_host.target}? [y/N] ")
             if ask != "y":

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

@@ -4,6 +4,7 @@ import sys
 
 from clan_lib.async_run import AsyncContext, AsyncOpts, AsyncRuntime
 from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.flake.flake import Flake
 from clan_lib.machines.actions import list_machines
 from clan_lib.machines.list import instantiate_inventory_to_machines
@@ -95,13 +96,8 @@ def get_machines_for_update(
 
 def update_command(args: argparse.Namespace) -> None:
     try:
-        if args.flake is None:
-            msg = "Could not find clan flake toplevel directory"
-            raise ClanError(msg)
-
-        machines_to_update = get_machines_for_update(
-            args.flake, args.machines, args.tags
-        )
+        flake = require_flake(args.flake)
+        machines_to_update = get_machines_for_update(flake, args.machines, args.tags)
 
         if args.target_host is not None and len(machines_to_update) > 1:
             msg = "Target Host can only be set for one machines"
@@ -111,7 +107,7 @@ def update_command(args: argparse.Namespace) -> None:
         config = nix_config()
         system = config["system"]
         machine_names = [machine.name for machine in machines_to_update]
-        args.flake.precache(
+        flake.precache(
             [
                 f"clanInternals.machines.{system}.{{{','.join(machine_names)}}}.config.clan.core.vars.generators.*.validationHash",
                 f"clanInternals.machines.{system}.{{{','.join(machine_names)}}}.config.clan.deployment.requireExplicitUpdate",

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

@@ -1,10 +1,12 @@
+from pathlib import Path
+
 import pytest
+from clan_lib.errors import ClanError
 from clan_lib.flake import Flake
 
 from clan_cli.machines.update import get_machines_for_update
-
-# Functions to test
 from clan_cli.tests.fixtures_flakes import FlakeForTest
+from clan_cli.tests.helpers import cli
 
 
 @pytest.mark.parametrize(
@@ -159,4 +161,13 @@ def test_get_machines_for_update_implicit_all(
     assert names == expected_names
 
 
+def test_update_command_no_flake(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    with pytest.raises(ClanError):
+        cli.run(["machines", "update", "machine1"])
+
+
 # TODO: Add more tests for requireExplicitUpdate

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

@@ -1,7 +1,7 @@
 import argparse
 from pathlib import Path
 
-from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.git import commit_files
 
 from clan_cli.completions import (
@@ -108,56 +108,44 @@ def remove_secret(
 
 
 def list_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    lst = list_sops_machines(args.flake.path)
+    flake = require_flake(args.flake)
+    lst = list_sops_machines(flake.path)
     if len(lst) > 0:
         print("\n".join(lst))
 
 
 def add_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    add_machine(args.flake.path, args.machine, args.key, args.force)
+    flake = require_flake(args.flake)
+    add_machine(flake.path, args.machine, args.key, args.force)
 
 
 def get_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    print(get_machine_pubkey(args.flake.path, args.machine))
+    flake = require_flake(args.flake)
+    print(get_machine_pubkey(flake.path, args.machine))
 
 
 def remove_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    remove_machine(args.flake.path, args.machine)
+    flake = require_flake(args.flake)
+    remove_machine(flake.path, args.machine)
 
 
 def add_secret_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
+    flake = require_flake(args.flake)
     add_secret(
-        args.flake.path,
+        flake.path,
         args.machine,
-        sops_secrets_folder(args.flake.path) / args.secret,
-        age_plugins=load_age_plugins(args.flake),
+        sops_secrets_folder(flake.path) / args.secret,
+        age_plugins=load_age_plugins(flake),
     )
 
 
 def remove_secret_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
+    flake = require_flake(args.flake)
     remove_secret(
-        args.flake.path,
+        flake.path,
         args.machine,
         args.secret,
-        age_plugins=load_age_plugins(args.flake),
+        age_plugins=load_age_plugins(flake),
     )
 
 

pkgs/clan-cli/clan_cli/secrets/users.py

@@ -6,6 +6,7 @@ from collections.abc import Iterable
 from pathlib import Path
 
 from clan_lib.errors import ClanError
+from clan_lib.flake import require_flake
 from clan_lib.git import commit_files
 
 from clan_cli.completions import add_dynamic_completer, complete_secrets, complete_users
@@ -122,10 +123,8 @@ def remove_secret(
 
 
 def list_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    lst = list_users(args.flake.path)
+    flake = require_flake(args.flake)
+    lst = list_users(flake.path)
     if len(lst) > 0:
         print("\n".join(lst))
 
@@ -193,66 +192,52 @@ def _key_args(args: argparse.Namespace) -> Iterable[sops.SopsKey]:
 
 
 def add_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
+    flake = require_flake(args.flake)
 
-    add_user(args.flake.path, args.user, _key_args(args), args.force)
+    add_user(flake.path, args.user, _key_args(args), args.force)
 
 
 def get_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    keys = get_user(args.flake.path, args.user)
+    flake = require_flake(args.flake)
+    keys = get_user(flake.path, args.user)
     json.dump([key.as_dict() for key in keys], sys.stdout, indent=2, sort_keys=True)
 
 
 def remove_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-    remove_user(args.flake.path, args.user)
+    flake = require_flake(args.flake)
+    remove_user(flake.path, args.user)
 
 
 def add_secret_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
+    flake = require_flake(args.flake)
     add_secret(
-        args.flake.path,
+        flake.path,
         args.user,
         args.secret,
-        age_plugins=load_age_plugins(args.flake),
+        age_plugins=load_age_plugins(flake),
     )
 
 
 def remove_secret_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
+    flake = require_flake(args.flake)
     remove_secret(
-        args.flake.path,
+        flake.path,
         args.user,
         args.secret,
-        age_plugins=load_age_plugins(args.flake),
+        age_plugins=load_age_plugins(flake),
     )
 
 
 def add_key_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
+    flake = require_flake(args.flake)
 
-    add_user_key(args.flake.path, args.user, _key_args(args))
+    add_user_key(flake.path, args.user, _key_args(args))
 
 
 def remove_key_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
+    flake = require_flake(args.flake)
 
-    remove_user_key(args.flake.path, args.user, _key_args(args))
+    remove_user_key(flake.path, args.user, _key_args(args))
 
 
 def register_users_parser(parser: argparse.ArgumentParser) -> None:

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

@@ -20,7 +20,7 @@ from clan_cli.vars.migration import check_can_migrate, migrate_files
 from clan_lib.api import API
 from clan_lib.cmd import RunOpts, run
 from clan_lib.errors import ClanError
-from clan_lib.flake import Flake
+from clan_lib.flake import Flake, require_flake
 from clan_lib.git import commit_files
 from clan_lib.machines.list import list_full_machines
 from clan_lib.nix import nix_config, nix_shell, nix_test_store
@@ -603,11 +603,8 @@ def generate_vars(
 
 
 def generate_command(args: argparse.Namespace) -> None:
-    if args.flake is None:
-        msg = "Could not find clan flake toplevel directory"
-        raise ClanError(msg)
-
-    machines: list[Machine] = list(list_full_machines(args.flake).values())
+    flake = require_flake(args.flake)
+    machines: list[Machine] = list(list_full_machines(flake).values())
 
     if len(args.machines) > 0:
         machines = list(
@@ -622,7 +619,7 @@ def generate_command(args: argparse.Namespace) -> None:
     system = config["system"]
     machine_names = [machine.name for machine in machines]
     # test
-    args.flake.precache(
+    flake.precache(
         [
             f"clanInternals.machines.{system}.{{{','.join(machine_names)}}}.config.clan.core.vars.generators.*.validationHash",
         ]
@@ -635,7 +632,7 @@ def generate_command(args: argparse.Namespace) -> None:
         fake_prompts=args.fake_prompts,
     )
     if has_changed:
-        args.flake.invalidate_cache()
+        flake.invalidate_cache()
 
 
 def register_generate_parser(parser: argparse.ArgumentParser) -> None:

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

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+import pytest
+from clan_cli.tests.helpers import cli
+from clan_lib.errors import ClanError
+
+
+def test_generate_command_no_flake(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    with pytest.raises(ClanError):
+        cli.run(["vars", "generate"])