Compare commits

...

2 Commits

Author SHA1 Message Date
Johannes Kirschbauer
fea4c2dc48 vm_manager: make timesink happy 2025-08-13 20:17:37 +02:00
Johannes Kirschbauer
80bbc6d7a3 cli/clan-inspect: remove unsused command 2025-08-13 19:47:38 +02:00
4 changed files with 9 additions and 183 deletions

View File

@@ -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)

View File

@@ -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)

View File

@@ -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

View File

@@ -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: