migrate clan history to clan-vm-manager
this functionality is not really useful or used in clan-vm-manager and therefore should live in the clan-vm-manager. Not porting the test for now because we probably get rid of the clan-vm-manager soon in favour of the UI.
This commit is contained in:
@@ -16,7 +16,6 @@ __all__ = ["admin", "directory", "disk", "iwd", "mdns_discovery", "modules", "up
|
||||
from . import (
|
||||
backups,
|
||||
clan,
|
||||
history,
|
||||
secrets,
|
||||
select,
|
||||
state,
|
||||
@@ -372,12 +371,6 @@ For more detailed information, visit: {help_hyperlink("deploy", "https://docs.cl
|
||||
)
|
||||
vms.register_parser(parser_vms)
|
||||
|
||||
parser_history = subparsers.add_parser(
|
||||
"history",
|
||||
description="manage history",
|
||||
)
|
||||
history.register_parser(parser_history)
|
||||
|
||||
parser_select = subparsers.add_parser(
|
||||
"select",
|
||||
aliases=["se"],
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
# !/usr/bin/env python3
|
||||
import argparse
|
||||
|
||||
from .add import register_add_parser
|
||||
from .list import register_list_parser
|
||||
|
||||
|
||||
# takes a (sub)parser and configures it
|
||||
def register_parser(parser: argparse.ArgumentParser) -> None:
|
||||
subparser = parser.add_subparsers(
|
||||
title="command",
|
||||
description="the command to run",
|
||||
help="the command to run",
|
||||
required=True,
|
||||
)
|
||||
add_parser = subparser.add_parser("add", help="Add a clan flake")
|
||||
register_add_parser(add_parser)
|
||||
list_parser = subparser.add_parser("list", help="List recently used flakes")
|
||||
register_list_parser(list_parser)
|
||||
@@ -1,124 +0,0 @@
|
||||
# !/usr/bin/env python3
|
||||
import argparse
|
||||
import dataclasses
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from clan_cli.clan.inspect import FlakeConfig, inspect_flake
|
||||
from clan_cli.clan_uri import ClanURI
|
||||
from clan_cli.dirs import user_history_file
|
||||
from clan_cli.errors import ClanError
|
||||
from clan_cli.locked_open import read_history_file, write_history_file
|
||||
from clan_cli.machines.list import list_nixos_machines
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class HistoryEntry:
|
||||
last_used: str
|
||||
flake: FlakeConfig
|
||||
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"]),
|
||||
settings=data.get("settings", {}),
|
||||
)
|
||||
|
||||
|
||||
def _merge_dicts(d1: dict, d2: dict) -> dict:
|
||||
# create a new dictionary that copies d1
|
||||
merged = dict(d1)
|
||||
# iterate over the keys and values of d2
|
||||
for key, value in d2.items():
|
||||
# if the key is in d1 and both values are dictionaries, merge them recursively
|
||||
if key in d1 and isinstance(d1[key], dict) and isinstance(value, dict):
|
||||
merged[key] = _merge_dicts(d1[key], value)
|
||||
# otherwise, update the value of the key in the merged dictionary
|
||||
else:
|
||||
merged[key] = value
|
||||
# return the merged dictionary
|
||||
return merged
|
||||
|
||||
|
||||
def list_history() -> list[HistoryEntry]:
|
||||
logs: list[HistoryEntry] = []
|
||||
if not user_history_file().exists():
|
||||
return []
|
||||
|
||||
try:
|
||||
parsed = read_history_file()
|
||||
for i, p in enumerate(parsed.copy()):
|
||||
# Everything from the settings dict is merged into the flake dict, and can override existing values
|
||||
parsed[i] = _merge_dicts(p, p.get("settings", {}))
|
||||
logs = [HistoryEntry.from_json(p) for p in parsed]
|
||||
except (json.JSONDecodeError, TypeError) as ex:
|
||||
msg = f"History file at {user_history_file()} is corrupted"
|
||||
raise ClanError(msg) from ex
|
||||
|
||||
return logs
|
||||
|
||||
|
||||
def new_history_entry(url: str, machine: str) -> HistoryEntry:
|
||||
flake = inspect_flake(url, machine)
|
||||
return HistoryEntry(
|
||||
flake=flake,
|
||||
last_used=datetime.datetime.now(tz=datetime.UTC).isoformat(),
|
||||
)
|
||||
|
||||
|
||||
def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]:
|
||||
history = list_history()
|
||||
new_entries: list[HistoryEntry] = []
|
||||
for machine in list_nixos_machines(uri.get_url()):
|
||||
new_entry = _add_maschine_to_history_list(uri.get_url(), machine, history)
|
||||
new_entries.append(new_entry)
|
||||
write_history_file(history)
|
||||
return new_entries
|
||||
|
||||
|
||||
def add_history(uri: ClanURI) -> HistoryEntry:
|
||||
user_history_file().parent.mkdir(parents=True, exist_ok=True)
|
||||
history = list_history()
|
||||
new_entry = _add_maschine_to_history_list(uri.get_url(), uri.machine_name, history)
|
||||
write_history_file(history)
|
||||
return new_entry
|
||||
|
||||
|
||||
def _add_maschine_to_history_list(
|
||||
uri_path: str, uri_machine: str, entries: list[HistoryEntry]
|
||||
) -> 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.last_used = datetime.datetime.now(tz=datetime.UTC).isoformat()
|
||||
return new_entry
|
||||
|
||||
new_entry = new_history_entry(uri_path, uri_machine)
|
||||
entries.append(new_entry)
|
||||
return new_entry
|
||||
|
||||
|
||||
def add_history_command(args: argparse.Namespace) -> None:
|
||||
if args.all:
|
||||
add_all_to_history(args.uri)
|
||||
else:
|
||||
add_history(args.uri)
|
||||
|
||||
|
||||
# takes a (sub)parser and configures it
|
||||
def register_add_parser(parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument(
|
||||
"uri", type=ClanURI.from_str, help="Path to the flake", default="."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--all", help="Add all machines", default=False, action="store_true"
|
||||
)
|
||||
parser.set_defaults(func=add_history_command)
|
||||
@@ -1,26 +0,0 @@
|
||||
# !/usr/bin/env python3
|
||||
import argparse
|
||||
from datetime import datetime
|
||||
|
||||
from .add import HistoryEntry, list_history
|
||||
|
||||
|
||||
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)
|
||||
if res.get(url) is None:
|
||||
res[url] = []
|
||||
res[url].append(history_entry)
|
||||
|
||||
for flake_url, entries in res.items():
|
||||
print(flake_url)
|
||||
for entry in entries:
|
||||
d = datetime.fromisoformat(entry.last_used)
|
||||
last_used = d.strftime("%d/%m/%Y %H:%M:%S")
|
||||
print(f" {entry.flake.flake_attr} ({last_used})")
|
||||
|
||||
|
||||
# takes a (sub)parser and configures it
|
||||
def register_list_parser(parser: argparse.ArgumentParser) -> None:
|
||||
parser.set_defaults(func=list_history_command)
|
||||
@@ -1,47 +0,0 @@
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from clan_cli.dirs import user_history_file
|
||||
from clan_cli.history.add import HistoryEntry
|
||||
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_history_add(
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
cmd = [
|
||||
"history",
|
||||
"add",
|
||||
f"clan://{test_flake_with_core.path}#vm1",
|
||||
]
|
||||
cli.run(cmd)
|
||||
|
||||
history_file = user_history_file()
|
||||
assert history_file.exists()
|
||||
history = [
|
||||
HistoryEntry.from_json(entry) for entry in json.loads(history_file.read_text())
|
||||
]
|
||||
assert str(history[0].flake.flake_url) == str(test_flake_with_core.path)
|
||||
|
||||
|
||||
@pytest.mark.impure
|
||||
def test_history_list(
|
||||
capture_output: CaptureOutput,
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
with capture_output as output:
|
||||
cli.run(["history", "list"])
|
||||
assert str(test_flake_with_core.path) not in output.out
|
||||
|
||||
cli.run(["history", "add", f"clan://{test_flake_with_core.path}#vm1"])
|
||||
|
||||
with capture_output as output:
|
||||
cli.run(["history", "list"])
|
||||
assert str(test_flake_with_core.path) in output.out
|
||||
Reference in New Issue
Block a user