DavHau/clan-core
1
0
Fork 0
forked from clan/clan-core
Code Pull Requests Activity

make machine class now a dataclass

Browse Source
This commit is contained in:
Jörg Thalheim 2024-07-02 14:06:31 +02:00
parent cc583dd79e
commit b4698528ef
23 changed files with 125 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
pkgs/clan-cli/clan_cli/backups/create.py
View File

@ -2,6 +2,7 @@ import argparse
import json
import logging
from ..clan_uri import FlakeId
from ..completions import (
add_dynamic_completer,
complete_backup_providers_for_machine,
@ -40,7 +41,7 @@ def create_backup(machine: Machine, provider: str | None = None) -> None:
def create_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
machine = Machine(name=args.machine, flake=args.flake)
machine = Machine(name=args.machine, flake=FlakeId(args.flake))
create_backup(machine=machine, provider=args.provider)

3
pkgs/clan-cli/clan_cli/backups/list.py
View File

@ -3,6 +3,7 @@ import json
import subprocess
from dataclasses import dataclass
from ..clan_uri import FlakeId
from ..completions import (
add_dynamic_completer,
complete_backup_providers_for_machine,
@ -55,7 +56,7 @@ def list_backups(machine: Machine, provider: str | None = None) -> list[Backup]:
def list_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
machine = Machine(name=args.machine, flake=args.flake)
machine = Machine(name=args.machine, flake=FlakeId(args.flake))
backups = list_backups(machine=machine, provider=args.provider)
for backup in backups:
print(backup.name)

3
pkgs/clan-cli/clan_cli/backups/restore.py
View File

@ -2,6 +2,7 @@ import argparse
import json
import subprocess
from ..clan_uri import FlakeId
from ..completions import (
add_dynamic_completer,
complete_backup_providers_for_machine,
@ -86,7 +87,7 @@ def restore_backup(
def restore_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
machine = Machine(name=args.machine, flake=args.flake)
machine = Machine(name=args.machine, flake=FlakeId(args.flake))
restore_backup(
machine=machine,
provider=args.provider,

9
pkgs/clan-cli/clan_cli/clan/inspect.py
View File

@ -2,6 +2,7 @@ import argparse
from dataclasses import dataclass
from pathlib import Path
from ..clan_uri import FlakeId
from ..cmd import run
from ..dirs import machine_gcroot
from ..errors import ClanError
@ -45,7 +46,7 @@ def inspect_flake(flake_url: str | Path, machine_name: str) -> FlakeConfig:
f"Machine {machine_name} not found in {flake_url}. Available machines: {', '.join(machines)}"
)
machine = Machine(machine_name, flake_url)
machine = Machine(machine_name, FlakeId(flake_url))
vm = inspect_vm(machine)
# Make symlink to gcroots from vm.machine_icon
@ -102,16 +103,16 @@ def inspect_flake(flake_url: str | Path, machine_name: str) -> FlakeConfig:
@dataclass
class InspectOptions:
machine: str
flake: Path
flake: FlakeId
def inspect_command(args: argparse.Namespace) -> None:
inspect_options = InspectOptions(
machine=args.machine,
flake=args.flake or Path.cwd(),
flake=FlakeId(args.flake or Path.cwd()),
)
res = inspect_flake(
flake_url=inspect_options.flake, machine_name=inspect_options.machine
flake_url=str(inspect_options.flake), machine_name=inspect_options.machine
)
print("Clan name:", res.clan_name)
print("Icon:", res.icon)

14
pkgs/clan-cli/clan_cli/clan_uri.py
View File

@ -9,6 +9,7 @@ from .errors import ClanError
@dataclass
class FlakeId:
# FIXME: this is such a footgun if you accidnetally pass a string
_value: str | Path
def __str__(self) -> str:
@ -26,9 +27,6 @@ class FlakeId:
assert isinstance(self._value, str)
return self._value
def __repr__(self) -> str:
return f"ClanUrl({self._value})"
def is_local(self) -> bool:
return isinstance(self._value, Path)
@ -37,8 +35,9 @@ class FlakeId:
# Define the ClanURI class
@dataclass
class ClanURI:
flake_id: FlakeId
flake: FlakeId
machine_name: str
# Initialize the class with a clan:// URI
@ -62,7 +61,7 @@ class ClanURI:
clean_comps = components._replace(query=components.query, fragment="")
# Parse the URL into a ClanUrl object
self.flake_id = self._parse_url(clean_comps)
self.flake = self._parse_url(clean_comps)
self.machine_name = "defaultVM"
if components.fragment:
self.machine_name = components.fragment
@ -85,7 +84,7 @@ class ClanURI:
return flake_id
def get_url(self) -> str:
return str(self.flake_id)
return str(self.flake)
@classmethod
def from_str(
@ -103,6 +102,3 @@ class ClanURI:
clan_uri += f"#{machine_name}"
return cls(clan_uri)
def __repr__(self) -> str:
return f"ClanURI({self})"

7
pkgs/clan-cli/clan_cli/completions.py
View File

@ -208,8 +208,7 @@ def complete_secrets(
"""
Provides completion functionality for clan secrets
"""
from pathlib import Path
from .clan_uri import FlakeId
from .secrets.secrets import ListSecretsOptions, list_secrets
if (clan_dir_result := clan_dir(None)) is not None:
@ -218,11 +217,11 @@ def complete_secrets(
flake = "."
options = ListSecretsOptions(
flake=Path(flake),
flake=FlakeId(flake),
pattern=None,
)
secrets = list_secrets(options.flake, options.pattern)
secrets = list_secrets(options.flake.path, options.pattern)
secrets_dict = {name: "secret" for name in secrets}
return secrets_dict

9
pkgs/clan-cli/clan_cli/errors.py
View File

@ -57,10 +57,13 @@ class ClanError(Exception):
self.description = description
self.location = location or "Unknown location"
self.msg = msg or ""
exception_msg = ""
if location:
exception_msg += f"{location}: \n"
exception_msg += self.msg
if self.description:
exception_msg = f"{self.location}: \n{self.msg} - {self.description}"
else:
exception_msg = f"{self.location}: \n{self.msg}"
exception_msg = f" - {self.description}"
super().__init__(exception_msg)

3
pkgs/clan-cli/clan_cli/facts/check.py
View File

@ -2,6 +2,7 @@ import argparse
import importlib
import logging
from ..clan_uri import FlakeId
from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine
@ -49,7 +50,7 @@ def check_secrets(machine: Machine, service: None | str = None) -> bool:
def check_command(args: argparse.Namespace) -> None:
machine = Machine(
name=args.machine,
flake=args.flake,
flake=FlakeId(args.flake),
)
check_secrets(machine, service=args.service)

3
pkgs/clan-cli/clan_cli/facts/list.py
View File

@ -3,6 +3,7 @@ import importlib
import json
import logging
from ..clan_uri import FlakeId
from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine
@ -26,7 +27,7 @@ def get_all_facts(machine: Machine) -> dict:
def get_command(args: argparse.Namespace) -> None:
machine = Machine(name=args.machine, flake=args.flake)
machine = Machine(name=args.machine, flake=FlakeId(args.flake))
# the raw_facts are bytestrings making them not json serializable
raw_facts = get_all_facts(machine)

8
pkgs/clan-cli/clan_cli/facts/public_modules/in_repo.py
View File

@ -12,9 +12,13 @@ class FactStore(FactStoreBase):
self.works_remotely = False
def set(self, service: str, name: str, value: bytes) -> Path | None:
if isinstance(self.machine.flake, Path):
if self.machine.flake.is_local():
fact_path = (
self.machine.flake / "machines" / self.machine.name / "facts" / name
self.machine.flake.path
/ "machines"
/ self.machine.name
/ "facts"
/ name
)
fact_path.parent.mkdir(parents=True, exist_ok=True)
fact_path.touch()

3
pkgs/clan-cli/clan_cli/facts/upload.py
View File

@ -4,6 +4,7 @@ import logging
from pathlib import Path
from tempfile import TemporaryDirectory
from ..clan_uri import FlakeId
from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine
@ -44,7 +45,7 @@ def upload_secrets(machine: Machine) -> None:
def upload_command(args: argparse.Namespace) -> None:
machine = Machine(name=args.machine, flake=args.flake)
machine = Machine(name=args.machine, flake=FlakeId(args.flake))
upload_secrets(machine)

5
pkgs/clan-cli/clan_cli/flash.py
View File

@ -11,6 +11,7 @@ from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any
from .clan_uri import FlakeId
from .cmd import Log, run
from .completions import add_dynamic_completer, complete_machines
from .errors import ClanError
@ -87,7 +88,7 @@ def flash_machine(
@dataclass
class FlashOptions:
flake: Path
flake: FlakeId
machine: str
disks: dict[str, str]
ssh_keys_path: list[Path]
@ -119,7 +120,7 @@ class AppendDiskAction(argparse.Action):
def flash_command(args: argparse.Namespace) -> None:
opts = FlashOptions(
flake=args.flake,
flake=FlakeId(args.flake),
machine=args.machine,
disks=args.disk,
ssh_keys_path=args.ssh_pubkey,

5
pkgs/clan-cli/clan_cli/machines/install.py
View File

@ -7,6 +7,7 @@ from dataclasses import dataclass
from pathlib import Path
from tempfile import TemporaryDirectory
from ..clan_uri import FlakeId
from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines
from ..facts.generate import generate_facts
@ -90,7 +91,7 @@ def install_nixos(
@dataclass
class InstallOptions:
flake: Path
flake: FlakeId
machine: str
target_host: str
kexec: str | None
@ -123,7 +124,7 @@ def install_command(args: argparse.Namespace) -> None:
password = None
opts = InstallOptions(
flake=args.flake,
flake=FlakeId(args.flake),
machine=args.machine,
target_host=target_host,
kexec=args.kexec,

9
pkgs/clan-cli/clan_cli/machines/inventory.py
View File

@ -1,6 +1,7 @@
import json
from pathlib import Path
from ..clan_uri import FlakeId
from ..cmd import run
from ..nix import nix_build, nix_config
from .machines import Machine
@ -21,8 +22,8 @@ def get_all_machines(flake_dir: Path, nix_options: list[str]) -> list[Machine]:
machines.append(
Machine(
name=name,
flake=flake_dir,
deployment_info=machine_data,
flake=FlakeId(flake_dir),
cached_deployment=machine_data,
nix_options=nix_options,
)
)
@ -34,5 +35,7 @@ def get_selected_machines(
) -> list[Machine]:
machines = []
for name in machine_names:
machines.append(Machine(name=name, flake=flake_dir, nix_options=nix_options))
machines.append(
Machine(name=name, flake=FlakeId(flake_dir), nix_options=nix_options)
)
return machines

99
pkgs/clan-cli/clan_cli/machines/machines.py
View File

@ -1,10 +1,11 @@
import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any
from clan_cli.clan_uri import ClanURI, FlakeId
from clan_cli.clan_uri import FlakeId
from ..cmd import run_no_stdout
from ..errors import ClanError
@ -14,63 +15,41 @@ from ..ssh import Host, parse_deployment_address
log = logging.getLogger(__name__)
@dataclass
class Machine:
name: str
flake: FlakeId
nix_options: list[str]
eval_cache: dict[str, str]
build_cache: dict[str, Path]
_flake_path: Path | None
_deployment_info: None | dict
nix_options: list[str] = field(default_factory=list)
cached_deployment: None | dict = None
def __init__(
self,
name: str,
flake: Path | str,
deployment_info: dict | None = None,
nix_options: list[str] = [],
) -> None:
"""
Creates a Machine
@name: the name of the machine
@clan_dir: the directory of the clan, optional, if not set it will be determined from the current working directory
@machine_json: can be optionally used to skip evaluation of the machine, location of the json file with machine data
"""
uri = ClanURI.from_str(str(flake), name)
self.flake_id = uri.flake_id
self.name: str = uri.machine_name
self.eval_cache: dict[str, str] = {}
self.build_cache: dict[str, Path] = {}
self._flake_path: Path | None = None
self._deployment_info: None | dict = deployment_info
self.nix_options = nix_options
_eval_cache: dict[str, str] = field(default_factory=dict)
_build_cache: dict[str, Path] = field(default_factory=dict)
def flush_caches(self) -> None:
self._deployment_info = None
self._flake_path = None
self.build_cache.clear()
self.eval_cache.clear()
self.cached_deployment = None
self._build_cache.clear()
self._eval_cache.clear()
def __str__(self) -> str:
return f"Machine(name={self.name}, flake={self.flake_id})"
return f"Machine(name={self.name}, flake={self.flake})"
def __repr__(self) -> str:
return str(self)
@property
def deployment_info(self) -> dict:
if self._deployment_info is not None:
return self._deployment_info
self._deployment_info = json.loads(
def deployment(self) -> dict:
if self.cached_deployment is not None:
return self.cached_deployment
deployment = json.loads(
self.build_nix("config.system.clan.deployment.file").read_text()
)
return self._deployment_info
self.cached_deployment = deployment
return deployment
@property
def target_host_address(self) -> str:
# deploymentAddress is deprecated.
val = self.deployment_info.get("targetHost") or self.deployment_info.get(
val = self.deployment.get("targetHost") or self.deployment.get(
"deploymentAddress"
)
if val is None:
@ -80,40 +59,34 @@ class Machine:
@target_host_address.setter
def target_host_address(self, value: str) -> None:
self.deployment_info["targetHost"] = value
self.deployment["targetHost"] = value
@property
def secret_facts_module(self) -> str:
return self.deployment_info["facts"]["secretModule"]
return self.deployment["facts"]["secretModule"]
@property
def public_facts_module(self) -> str:
return self.deployment_info["facts"]["publicModule"]
return self.deployment["facts"]["publicModule"]
@property
def facts_data(self) -> dict[str, dict[str, Any]]:
if self.deployment_info["facts"]["services"]:
return self.deployment_info["facts"]["services"]
if self.deployment["facts"]["services"]:
return self.deployment["facts"]["services"]
return {}
@property
def secrets_upload_directory(self) -> str:
return self.deployment_info["facts"]["secretUploadDirectory"]
return self.deployment["facts"]["secretUploadDirectory"]
@property
def flake_dir(self) -> Path:
if self._flake_path:
return self._flake_path
if self.flake_id.is_local():
self._flake_path = self.flake_id.path
elif self.flake_id.is_remote():
self._flake_path = Path(nix_metadata(self.flake_id.url)["path"])
if self.flake.is_local():
return self.flake.path
elif self.flake.is_remote():
return Path(nix_metadata(self.flake.url)["path"])
else:
raise ClanError(f"Unsupported flake url: {self.flake_id}")
assert self._flake_path is not None
return self._flake_path
raise ClanError(f"Unsupported flake url: {self.flake}")
@property
def target_host(self) -> Host:
@ -127,7 +100,7 @@ class Machine:
The host where the machine is built and deployed from.
Can be the same as the target host.
"""
build_host = self.deployment_info.get("buildHost")
build_host = self.deployment.get("buildHost")
if build_host is None:
return self.target_host
# enable ssh agent forwarding to allow the build host to access the target host
@ -230,12 +203,12 @@ class Machine:
eval a nix attribute of the machine
@attr: the attribute to get
"""
if attr in self.eval_cache and not refresh and extra_config is None:
return self.eval_cache[attr]
if attr in self._eval_cache and not refresh and extra_config is None:
return self._eval_cache[attr]
output = self.nix("eval", attr, extra_config, impure, nix_options)
if isinstance(output, str):
self.eval_cache[attr] = output
self._eval_cache[attr] = output
return output
else:
raise ClanError("eval_nix returned not a string")
@ -253,12 +226,12 @@ class Machine:
@attr: the attribute to get
"""
if attr in self.build_cache and not refresh and extra_config is None:
return self.build_cache[attr]
if attr in self._build_cache and not refresh and extra_config is None:
return self._build_cache[attr]
output = self.nix("build", attr, extra_config, impure, nix_options)
if isinstance(output, Path):
self.build_cache[attr] = output
self._build_cache[attr] = output
return output
else:
raise ClanError("build_nix returned not a Path")

14
pkgs/clan-cli/clan_cli/machines/update.py
View File

@ -5,6 +5,7 @@ import os
import shlex
import sys
from ..clan_uri import FlakeId
from ..cmd import run
from ..completions import add_dynamic_completer, complete_machines
from ..errors import ClanError
@ -80,7 +81,7 @@ def upload_sources(
)
def deploy_nixos(machines: MachineGroup) -> None:
def deploy_machine(machines: MachineGroup) -> None:
"""
Deploy to all hosts in parallel
"""
@ -137,7 +138,7 @@ def update(args: argparse.Namespace) -> None:
machines = []
if len(args.machines) == 1 and args.target_host is not None:
machine = Machine(
name=args.machines[0], flake=args.flake, nix_options=args.option
name=args.machines[0], flake=FlakeId(args.flake), nix_options=args.option
)
machine.target_host_address = args.target_host
machines.append(machine)
@ -149,7 +150,7 @@ def update(args: argparse.Namespace) -> None:
if len(args.machines) == 0:
ignored_machines = []
for machine in get_all_machines(args.flake, args.option):
if machine.deployment_info.get("requireExplicitUpdate", False):
if machine.deployment.get("requireExplicitUpdate", False):
continue
try:
machine.build_host
@ -170,7 +171,7 @@ def update(args: argparse.Namespace) -> None:
else:
machines = get_selected_machines(args.flake, args.option, args.machines)
deploy_nixos(MachineGroup(machines))
deploy_machine(MachineGroup(machines))
def register_update_parser(parser: argparse.ArgumentParser) -> None:
@ -190,4 +191,9 @@ def register_update_parser(parser: argparse.ArgumentParser) -> None:
type=str,
help="address of the machine to update, in the format of user@host:1234",
)
parser.add_argument(
"--darwin",
type=str,
help="Hack to deploy darwin machines. This will be removed in the future when we have full darwin integration.",
)
parser.set_defaults(func=update)

7
pkgs/clan-cli/clan_cli/secrets/secrets.py
View File

@ -9,6 +9,7 @@ from pathlib import Path
from typing import IO
from .. import tty
from ..clan_uri import FlakeId
from ..completions import (
add_dynamic_completer,
complete_groups,
@ -263,16 +264,16 @@ def list_secrets(flake_dir: Path, pattern: str | None = None) -> list[str]:
@dataclass
class ListSecretsOptions:
flake: Path
flake: FlakeId
pattern: str | None
def list_command(args: argparse.Namespace) -> None:
options = ListSecretsOptions(
flake=args.flake,
flake=FlakeId(args.flake),
pattern=args.pattern,
)
lst = list_secrets(options.flake, options.pattern)
lst = list_secrets(options.flake.path, options.pattern)
if len(lst) > 0:
print("\n".join(lst))

9
pkgs/clan-cli/clan_cli/vms/inspect.py
View File

@ -3,6 +3,7 @@ import json
from dataclasses import dataclass
from pathlib import Path
from ..clan_uri import FlakeId
from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine
@ -12,7 +13,7 @@ class VmConfig:
machine_name: str
machine_icon: Path
machine_description: str
flake_url: str | Path
flake_url: FlakeId
clan_name: str
cores: int
@ -23,19 +24,19 @@ class VmConfig:
def inspect_vm(machine: Machine) -> VmConfig:
data = json.loads(machine.eval_nix("config.clan.core.vm.inspect"))
return VmConfig(flake_url=str(machine.flake), **data)
return VmConfig(flake_url=machine.flake, **data)
@dataclass
class InspectOptions:
machine: str
flake: Path
flake: FlakeId
def inspect_command(args: argparse.Namespace) -> None:
inspect_options = InspectOptions(
machine=args.machine,
flake=args.flake or Path.cwd(),
flake=FlakeId(args.flake or Path.cwd()),
)
machine = Machine(inspect_options.machine, inspect_options.flake)

3
pkgs/clan-cli/clan_cli/vms/run.py
View File

@ -7,6 +7,7 @@ from contextlib import ExitStack
from pathlib import Path
from tempfile import TemporaryDirectory
from ..clan_uri import FlakeId
from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines
from ..dirs import module_root, user_cache_dir, vm_state_dir
@ -198,7 +199,7 @@ def run_command(
option: list[str] = [],
**kwargs: dict[str, str],
) -> None:
machine_obj: Machine = Machine(machine, flake)
machine_obj: Machine = Machine(machine, FlakeId(flake))
vm: VmConfig = inspect_vm(machine=machine_obj)

20
pkgs/clan-cli/tests/test_clan_uri.py
View File

@ -21,13 +21,13 @@ def test_get_url() -> None:
def test_local_uri() -> None:
# Create a ClanURI object from a local URI
uri = ClanURI("clan://file:///home/user/Downloads")
assert uri.flake_id.path == Path("/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")
assert uri.flake_id.url == "https://example.com"
assert uri.flake.url == "https://example.com"
def test_direct_local_path() -> None:
@ -47,35 +47,35 @@ def test_remote_with_clanparams() -> None:
uri = ClanURI("clan://https://example.com")
assert uri.machine_name == "defaultVM"
assert uri.flake_id.url == "https://example.com"
assert uri.flake.url == "https://example.com"
def test_from_str_remote() -> None:
uri = ClanURI.from_str(url="https://example.com", machine_name="myVM")
assert uri.get_url() == "https://example.com"
assert uri.machine_name == "myVM"
assert uri.flake_id.url == "https://example.com"
assert uri.flake.url == "https://example.com"
def test_from_str_local() -> None:
uri = ClanURI.from_str(url="~/Projects/democlan", machine_name="myVM")
assert uri.get_url().endswith("/Projects/democlan")
assert uri.machine_name == "myVM"
assert uri.flake_id.is_local()
assert str(uri.flake_id).endswith("/Projects/democlan") # type: ignore
assert uri.flake.is_local()
assert str(uri.flake).endswith("/Projects/democlan") # type: ignore
def test_from_str_local_no_machine() -> None:
uri = ClanURI.from_str("~/Projects/democlan")
assert uri.get_url().endswith("/Projects/democlan")
assert uri.machine_name == "defaultVM"
assert uri.flake_id.is_local()
assert str(uri.flake_id).endswith("/Projects/democlan") # type: ignore
assert uri.flake.is_local()
assert str(uri.flake).endswith("/Projects/democlan") # type: ignore
def test_from_str_local_no_machine2() -> None:
uri = ClanURI.from_str("~/Projects/democlan#syncthing-peer1")
assert uri.get_url().endswith("/Projects/democlan")
assert uri.machine_name == "syncthing-peer1"
assert uri.flake_id.is_local()
assert str(uri.flake_id).endswith("/Projects/democlan") # type: ignore
assert uri.flake.is_local()
assert str(uri.flake).endswith("/Projects/democlan") # type: ignore

7
pkgs/clan-cli/tests/test_history_cli.py
View File

@ -6,7 +6,6 @@ from cli import Cli
from fixtures_flakes import FlakeForTest
from pytest import CaptureFixture
from clan_cli.clan_uri import ClanURI
from clan_cli.dirs import user_history_file
from clan_cli.history.add import HistoryEntry
@ -19,11 +18,10 @@ def test_history_add(
test_flake_with_core: FlakeForTest,
) -> None:
cli = Cli()
uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1")
cmd = [
"history",
"add",
str(uri),
f"clan://{test_flake_with_core.path}#vm1",
]
cli.run(cmd)
@ -39,7 +37,6 @@ def test_history_list(
test_flake_with_core: FlakeForTest,
) -> None:
cli = Cli()
uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1")
cmd = [
"history",
"list",
@ -48,6 +45,6 @@ def test_history_list(
cli.run(cmd)
assert str(test_flake_with_core.path) not in capsys.readouterr().out
cli.run(["history", "add", str(uri)])
cli.run(["history", "add", f"clan://{test_flake_with_core.path}#vm1"])
cli.run(cmd)
assert str(test_flake_with_core.path) in capsys.readouterr().out

5
pkgs/clan-cli/tests/test_secrets_generate.py
View File

@ -6,6 +6,7 @@ from cli import Cli
from fixtures_flakes import FlakeForTest
from validator import is_valid_age_key, is_valid_ssh_key
from clan_cli.clan_uri import FlakeId
from clan_cli.facts.secret_modules.sops import SecretStore
from clan_cli.machines.facts import machine_get_fact
from clan_cli.machines.machines import Machine
@ -48,7 +49,7 @@ def test_generate_secret(
)
cmd = ["facts", "generate", "--flake", str(test_flake_with_core.path), "vm1"]
cli.run(cmd)
store1 = SecretStore(Machine(name="vm1", flake=test_flake_with_core.path))
store1 = SecretStore(Machine(name="vm1", flake=FlakeId(test_flake_with_core.path)))
assert store1.exists("", "age.key")
assert store1.exists("", "zerotier-identity-secret")
@ -76,7 +77,7 @@ def test_generate_secret(
secrets_folder / "vm1-zerotier-identity-secret" / "machines" / "vm1"
).exists()
store2 = SecretStore(Machine(name="vm2", flake=test_flake_with_core.path))
store2 = SecretStore(Machine(name="vm2", flake=FlakeId(test_flake_with_core.path)))
assert store2.exists("", "password")
assert store2.exists("", "password-hash")

5
pkgs/clan-cli/tests/test_secrets_password_store.py
View File

@ -6,6 +6,7 @@ from cli import Cli
from fixtures_flakes import FlakeForTest
from validator import is_valid_ssh_key
from clan_cli.clan_uri import FlakeId
from clan_cli.facts.secret_modules.password_store import SecretStore
from clan_cli.machines.facts import machine_get_fact
from clan_cli.machines.machines import Machine
@ -48,7 +49,9 @@ def test_upload_secret(
)
cli.run(["facts", "generate", "vm1"])
store = SecretStore(Machine(name="vm1", flake=test_flake_with_core_and_pass.path))
store = SecretStore(
Machine(name="vm1", flake=FlakeId(test_flake_with_core_and_pass.path))
)
network_id = machine_get_fact(
test_flake_with_core_and_pass.path, "vm1", "zerotier-network-id"