Compare commits
2 Commits
push-qxwmr
...
cli-clan-i
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fea4c2dc48 | ||
|
|
80bbc6d7a3 |
@@ -1,8 +1,6 @@
|
|||||||
# !/usr/bin/env python3
|
# !/usr/bin/env python3
|
||||||
import argparse
|
import argparse
|
||||||
|
|
||||||
from clan_cli.clan.inspect import register_inspect_parser
|
|
||||||
|
|
||||||
from .create import register_create_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")
|
create_parser = subparser.add_parser("create", help="Create a clan")
|
||||||
register_create_parser(create_parser)
|
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
|
import logging
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from clan_cli.clan.inspect import FlakeConfig, inspect_flake
|
|
||||||
from clan_lib.dirs import user_history_file
|
from clan_lib.dirs import user_history_file
|
||||||
from clan_lib.errors import ClanError
|
from clan_lib.errors import ClanError
|
||||||
from clan_lib.flake import Flake
|
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.locked_open import read_history_file, write_history_file
|
||||||
from clan_lib.machines.list import list_machines
|
from clan_lib.machines.list import list_machines
|
||||||
|
|
||||||
@@ -20,14 +20,15 @@ log = logging.getLogger(__name__)
|
|||||||
@dataclasses.dataclass
|
@dataclasses.dataclass
|
||||||
class HistoryEntry:
|
class HistoryEntry:
|
||||||
last_used: str
|
last_used: str
|
||||||
flake: FlakeConfig
|
machine: Machine
|
||||||
settings: dict[str, Any] = dataclasses.field(default_factory=dict)
|
settings: dict[str, Any] = dataclasses.field(default_factory=dict)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_json(cls: type["HistoryEntry"], data: dict[str, Any]) -> "HistoryEntry":
|
def from_json(cls: type["HistoryEntry"], data: dict[str, Any]) -> "HistoryEntry":
|
||||||
return cls(
|
return cls(
|
||||||
last_used=data["last_used"],
|
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", {}),
|
settings=data.get("settings", {}),
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -66,9 +67,8 @@ def list_history() -> list[HistoryEntry]:
|
|||||||
|
|
||||||
|
|
||||||
def new_history_entry(url: str, machine: str) -> HistoryEntry:
|
def new_history_entry(url: str, machine: str) -> HistoryEntry:
|
||||||
flake = inspect_flake(url, machine)
|
|
||||||
return HistoryEntry(
|
return HistoryEntry(
|
||||||
flake=flake,
|
machine=Machine(machine, Flake(url)),
|
||||||
last_used=datetime.datetime.now(tz=datetime.UTC).isoformat(),
|
last_used=datetime.datetime.now(tz=datetime.UTC).isoformat(),
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -96,8 +96,8 @@ def _add_maschine_to_history_list(
|
|||||||
) -> HistoryEntry:
|
) -> HistoryEntry:
|
||||||
for new_entry in entries:
|
for new_entry in entries:
|
||||||
if (
|
if (
|
||||||
new_entry.flake.flake_url == str(uri_path)
|
new_entry.machine.flake.path == str(uri_path)
|
||||||
and new_entry.flake.flake_attr == uri_machine
|
and new_entry.machine.flake.path == uri_machine
|
||||||
):
|
):
|
||||||
new_entry.last_used = datetime.datetime.now(tz=datetime.UTC).isoformat()
|
new_entry.last_used = datetime.datetime.now(tz=datetime.UTC).isoformat()
|
||||||
return new_entry
|
return new_entry
|
||||||
@@ -117,7 +117,7 @@ def add_history_command(args: argparse.Namespace) -> None:
|
|||||||
def list_history_command(args: argparse.Namespace) -> None:
|
def list_history_command(args: argparse.Namespace) -> None:
|
||||||
res: dict[str, list[HistoryEntry]] = {}
|
res: dict[str, list[HistoryEntry]] = {}
|
||||||
for history_entry in list_history():
|
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:
|
if res.get(url) is None:
|
||||||
res[url] = []
|
res[url] = []
|
||||||
res[url].append(history_entry)
|
res[url].append(history_entry)
|
||||||
@@ -127,7 +127,7 @@ def list_history_command(args: argparse.Namespace) -> None:
|
|||||||
for entry in entries:
|
for entry in entries:
|
||||||
d = datetime.datetime.fromisoformat(entry.last_used)
|
d = datetime.datetime.fromisoformat(entry.last_used)
|
||||||
last_used = d.strftime("%d/%m/%Y %H:%M:%S")
|
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:
|
def parse_args() -> argparse.Namespace:
|
||||||
|
|||||||
Reference in New Issue
Block a user