Compare commits
2 Commits
simplify_h
...
cli-clan-i
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fea4c2dc48 | ||
|
|
80bbc6d7a3 |
@@ -1,8 +1,6 @@
|
||||
# !/usr/bin/env python3
|
||||
import argparse
|
||||
|
||||
from clan_cli.clan.inspect import register_inspect_parser
|
||||
|
||||
from .create import register_create_parser
|
||||
|
||||
|
||||
@@ -16,5 +14,3 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
|
||||
)
|
||||
create_parser = subparser.add_parser("create", help="Create a clan")
|
||||
register_create_parser(create_parser)
|
||||
inspect_parser = subparser.add_parser("inspect", help="Inspect a clan ")
|
||||
register_inspect_parser(inspect_parser)
|
||||
|
||||
@@ -1,143 +0,0 @@
|
||||
import argparse
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from clan_lib.cmd import run
|
||||
from clan_lib.dirs import machine_gcroot
|
||||
from clan_lib.errors import ClanError
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.machines.actions import list_machines
|
||||
from clan_lib.machines.machines import Machine
|
||||
from clan_lib.nix import (
|
||||
nix_add_to_gcroots,
|
||||
nix_build,
|
||||
nix_config,
|
||||
nix_eval,
|
||||
nix_metadata,
|
||||
)
|
||||
|
||||
from clan_cli.vms.inspect import VmConfig, inspect_vm
|
||||
|
||||
|
||||
@dataclass
|
||||
class FlakeConfig:
|
||||
flake_url: Flake
|
||||
flake_attr: str
|
||||
|
||||
clan_name: str
|
||||
nar_hash: str
|
||||
icon: str | None
|
||||
description: str | None
|
||||
last_updated: str
|
||||
revision: str | None
|
||||
vm: VmConfig
|
||||
|
||||
@classmethod
|
||||
def from_json(cls: type["FlakeConfig"], data: dict[str, Any]) -> "FlakeConfig":
|
||||
return cls(
|
||||
flake_url=Flake.from_json(data["flake_url"]),
|
||||
flake_attr=data["flake_attr"],
|
||||
clan_name=data["clan_name"],
|
||||
nar_hash=data["nar_hash"],
|
||||
icon=data.get("icon"),
|
||||
description=data.get("description"),
|
||||
last_updated=data["last_updated"],
|
||||
revision=data.get("revision"),
|
||||
vm=VmConfig.from_json(data["vm"]),
|
||||
)
|
||||
|
||||
|
||||
def run_cmd(cmd: list[str]) -> str:
|
||||
proc = run(cmd)
|
||||
return proc.stdout.strip()
|
||||
|
||||
|
||||
def inspect_flake(flake_url: str | Path, machine_name: str) -> FlakeConfig:
|
||||
config = nix_config()
|
||||
system = config["system"]
|
||||
|
||||
# Check if the machine exists
|
||||
machines = list_machines(Flake(str(flake_url)))
|
||||
if machine_name not in machines:
|
||||
msg = f"Machine {machine_name} not found in {flake_url}. Available machines: {', '.join(machines)}"
|
||||
raise ClanError(msg)
|
||||
|
||||
machine = Machine(machine_name, Flake(str(flake_url)))
|
||||
vm = inspect_vm(machine)
|
||||
|
||||
# Make symlink to gcroots from vm.machine_icon
|
||||
if vm.machine_icon:
|
||||
gcroot_icon: Path = machine_gcroot(flake_url=str(flake_url)) / vm.machine_name
|
||||
nix_add_to_gcroots(vm.machine_icon, gcroot_icon)
|
||||
|
||||
# Get the Clan name
|
||||
cmd = nix_eval(
|
||||
[
|
||||
f'{flake_url}#clanInternals.machines."{system}"."{machine_name}".config.clan.core.name'
|
||||
]
|
||||
)
|
||||
res = run_cmd(cmd)
|
||||
clan_name = res.strip('"')
|
||||
|
||||
# Get the clan icon path
|
||||
cmd = nix_eval(
|
||||
[
|
||||
f'{flake_url}#clanInternals.machines."{system}"."{machine_name}".config.clan.core.icon'
|
||||
]
|
||||
)
|
||||
res = run_cmd(cmd)
|
||||
|
||||
# If the icon is null, no icon is set for this Clan
|
||||
if res == "null":
|
||||
icon_path = None
|
||||
else:
|
||||
icon_path = res.strip('"')
|
||||
|
||||
cmd = nix_build(
|
||||
[
|
||||
f'{flake_url}#clanInternals.machines."{system}"."{machine_name}".config.clan.core.icon'
|
||||
],
|
||||
machine_gcroot(flake_url=str(flake_url)) / "icon",
|
||||
)
|
||||
run_cmd(cmd)
|
||||
|
||||
# Get the flake metadata
|
||||
meta = nix_metadata(flake_url)
|
||||
return FlakeConfig(
|
||||
vm=vm,
|
||||
flake_url=Flake(str(flake_url)),
|
||||
clan_name=clan_name,
|
||||
flake_attr=machine_name,
|
||||
nar_hash=meta["locked"]["narHash"],
|
||||
icon=icon_path,
|
||||
description=meta.get("description"),
|
||||
last_updated=meta["lastModified"],
|
||||
revision=meta.get("revision"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class InspectOptions:
|
||||
machine: str
|
||||
flake: Flake
|
||||
|
||||
|
||||
def inspect_command(args: argparse.Namespace) -> None:
|
||||
inspect_options = InspectOptions(
|
||||
machine=args.machine,
|
||||
flake=args.flake or Flake(str(Path.cwd())),
|
||||
)
|
||||
res = inspect_flake(
|
||||
flake_url=str(inspect_options.flake), machine_name=inspect_options.machine
|
||||
)
|
||||
print("Clan name:", res.clan_name)
|
||||
print("Icon:", res.icon)
|
||||
print("Description:", res.description)
|
||||
print("Last updated:", res.last_updated)
|
||||
print("Revision:", res.revision)
|
||||
|
||||
|
||||
def register_inspect_parser(parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument("--machine", type=str, default="defaultVM")
|
||||
parser.set_defaults(func=inspect_command)
|
||||
@@ -1,27 +0,0 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from clan_cli.tests.fixtures_flakes import FlakeForTest
|
||||
from clan_cli.tests.helpers import cli
|
||||
from clan_cli.tests.stdout import CaptureOutput
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.impure
|
||||
def test_flakes_inspect(
|
||||
test_flake_with_core: FlakeForTest, capture_output: CaptureOutput
|
||||
) -> None:
|
||||
with capture_output as output:
|
||||
cli.run(
|
||||
[
|
||||
"flakes",
|
||||
"inspect",
|
||||
"--flake",
|
||||
str(test_flake_with_core.path),
|
||||
"--machine",
|
||||
"vm1",
|
||||
]
|
||||
)
|
||||
assert "Icon" in output.out
|
||||
@@ -5,10 +5,10 @@ import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from clan_cli.clan.inspect import FlakeConfig, inspect_flake
|
||||
from clan_lib.dirs import user_history_file
|
||||
from clan_lib.errors import ClanError
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.machines.machines import Machine
|
||||
from clan_lib.locked_open import read_history_file, write_history_file
|
||||
from clan_lib.machines.list import list_machines
|
||||
|
||||
@@ -20,14 +20,15 @@ log = logging.getLogger(__name__)
|
||||
@dataclasses.dataclass
|
||||
class HistoryEntry:
|
||||
last_used: str
|
||||
flake: FlakeConfig
|
||||
machine: Machine
|
||||
settings: dict[str, Any] = dataclasses.field(default_factory=dict)
|
||||
|
||||
@classmethod
|
||||
def from_json(cls: type["HistoryEntry"], data: dict[str, Any]) -> "HistoryEntry":
|
||||
return cls(
|
||||
last_used=data["last_used"],
|
||||
flake=FlakeConfig.from_json(data["flake"]),
|
||||
# TODO: This needs to be fixed, if we every spin up the vm Manager again
|
||||
machine=Machine(data["name"], Flake(data["flake"])),
|
||||
settings=data.get("settings", {}),
|
||||
)
|
||||
|
||||
@@ -66,9 +67,8 @@ def list_history() -> list[HistoryEntry]:
|
||||
|
||||
|
||||
def new_history_entry(url: str, machine: str) -> HistoryEntry:
|
||||
flake = inspect_flake(url, machine)
|
||||
return HistoryEntry(
|
||||
flake=flake,
|
||||
machine=Machine(machine, Flake(url)),
|
||||
last_used=datetime.datetime.now(tz=datetime.UTC).isoformat(),
|
||||
)
|
||||
|
||||
@@ -96,8 +96,8 @@ def _add_maschine_to_history_list(
|
||||
) -> HistoryEntry:
|
||||
for new_entry in entries:
|
||||
if (
|
||||
new_entry.flake.flake_url == str(uri_path)
|
||||
and new_entry.flake.flake_attr == uri_machine
|
||||
new_entry.machine.flake.path == str(uri_path)
|
||||
and new_entry.machine.flake.path == uri_machine
|
||||
):
|
||||
new_entry.last_used = datetime.datetime.now(tz=datetime.UTC).isoformat()
|
||||
return new_entry
|
||||
@@ -117,7 +117,7 @@ def add_history_command(args: argparse.Namespace) -> None:
|
||||
def list_history_command(args: argparse.Namespace) -> None:
|
||||
res: dict[str, list[HistoryEntry]] = {}
|
||||
for history_entry in list_history():
|
||||
url = str(history_entry.flake.flake_url)
|
||||
url = str(history_entry.machine.flake.path)
|
||||
if res.get(url) is None:
|
||||
res[url] = []
|
||||
res[url].append(history_entry)
|
||||
@@ -127,7 +127,7 @@ def list_history_command(args: argparse.Namespace) -> None:
|
||||
for entry in entries:
|
||||
d = datetime.datetime.fromisoformat(entry.last_used)
|
||||
last_used = d.strftime("%d/%m/%Y %H:%M:%S")
|
||||
print(f" {entry.flake.flake_attr} ({last_used})")
|
||||
print(f" {entry.machine} ({last_used})")
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
|
||||
Reference in New Issue
Block a user