yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan-app: Finish flash view. clan-cli: Flash cli now verifies if language and keymap are valid.

Browse Source
This commit is contained in:
Qubasa
2024-08-02 17:51:45 +02:00
parent f2e697f3e4
commit 3e9ebbc90f
27 changed files with 556 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
pkgs/clan-cli/clan_cli/api/cli.py Executable file
View File

@@ -0,0 +1,13 @@
#!/usr/bin/env python3
import argparse
import json
from clan_cli.api import API
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Debug the API.")
args = parser.parse_args()
schema = API.to_json_schema()
print(json.dumps(schema, indent=4))

25
pkgs/clan-cli/clan_cli/api/directory.py
View File

@@ -12,24 +12,26 @@ from . import API
@dataclass
class FileFilter:
title: str | None
mime_types: list[str] | None
patterns: list[str] | None
suffixes: list[str] | None
title: str | None = field(default=None)
mime_types: list[str] | None = field(default=None)
patterns: list[str] | None = field(default=None)
suffixes: list[str] | None = field(default=None)
@dataclass
class FileRequest:
# Mode of the os dialog window
mode: Literal["open_file", "select_folder", "save"]
mode: Literal["open_file", "select_folder", "save", "open_multiple_files"]
# Title of the os dialog window
title: str | None = None
title: str | None = field(default=None)
# Pre-applied filters for the file dialog
filters: FileFilter | None = None
filters: FileFilter | None = field(default=None)
initial_file: str | None = field(default=None)
initial_folder: str | None = field(default=None)
@API.register_abstract
def open_file(file_request: FileRequest) -> str | None:
def open_file(file_request: FileRequest) -> list[str] | None:
"""
Abstract api method to open a file dialog window.
It must return the name of the selected file or None if no file was selected.
@@ -88,6 +90,7 @@ def get_directory(current_path: str) -> Directory:
@dataclass
class BlkInfo:
name: str
path: str
rm: str
size: str
ro: bool
@@ -103,6 +106,7 @@ class Blockdevices:
def blk_from_dict(data: dict) -> BlkInfo:
return BlkInfo(
name=data["name"],
path=data["path"],
rm=data["rm"],
size=data["size"],
ro=data["ro"],
@@ -117,7 +121,10 @@ def show_block_devices() -> Blockdevices:
Abstract api method to show block devices.
It must return a list of block devices.
"""
cmd = nix_shell(["nixpkgs#util-linux"], ["lsblk", "--json"])
cmd = nix_shell(
["nixpkgs#util-linux"],
["lsblk", "--json", "--output", "PATH,NAME,RM,SIZE,RO,MOUNTPOINTS,TYPE"],
)
proc = run_no_stdout(cmd)
res = proc.stdout.strip()

2
pkgs/clan-cli/clan_cli/api/serde.py
View File

@@ -237,7 +237,7 @@ def from_dict(t: type[T], data: dict[str, Any], path: list[str] = []) -> T:
if field_name not in field_values:
formatted_path = " ".join(path)
raise ClanError(
f"Required field missing: '{field_name}' in {t} {formatted_path}, got Value: {data}"
f"Default value missing for: '{field_name}' in {t} {formatted_path}, got Value: {data}"
)
return t(**field_values) # type: ignore

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

@@ -6,7 +6,7 @@ import os
import shutil
import textwrap
from collections.abc import Sequence
from dataclasses import dataclass
from dataclasses import dataclass, field
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any
@@ -19,23 +19,105 @@ from .completions import add_dynamic_completer, complete_machines
from .errors import ClanError
from .facts.secret_modules import SecretStoreBase
from .machines.machines import Machine
from .nix import nix_shell
from .nix import nix_build, nix_shell
log = logging.getLogger(__name__)
@dataclass
class SystemConfig:
language: str | None = field(default=None)
keymap: str | None = field(default=None)
ssh_keys_path: list[str] | None = field(default=None)
@API.register
def list_possible_keymaps() -> list[str]:
cmd = nix_build(["nixpkgs#kbd"])
result = run(cmd, log=Log.STDERR, error_msg="Failed to find kbdinfo")
keymaps_dir = Path(result.stdout.strip()) / "share" / "keymaps"
if not keymaps_dir.exists():
raise FileNotFoundError(f"Keymaps directory '{keymaps_dir}' does not exist.")
keymap_files = []
for root, _, files in os.walk(keymaps_dir):
for file in files:
if file.endswith(".map.gz"):
# Remove '.map.gz' ending
name_without_ext = file[:-7]
keymap_files.append(name_without_ext)
return keymap_files
@API.register
def list_possible_languages() -> list[str]:
cmd = nix_build(["nixpkgs#glibcLocales"])
result = run(cmd, log=Log.STDERR, error_msg="Failed to find glibc locales")
locale_file = Path(result.stdout.strip()) / "share" / "i18n" / "SUPPORTED"
if not locale_file.exists():
raise FileNotFoundError(f"Locale file '{locale_file}' does not exist.")
with locale_file.open() as f:
lines = f.readlines()
languages = []
for line in lines:
if line.startswith("#"):
continue
if "SUPPORTED-LOCALES" in line:
continue
# Split by '/' and take the first part
language = line.split("/")[0].strip()
languages.append(language)
return languages
@API.register
def flash_machine(
machine: Machine,
*,
mode: str,
disks: dict[str, str],
system_config: dict[str, Any],
system_config: SystemConfig,
dry_run: bool,
write_efi_boot_entries: bool,
debug: bool,
extra_args: list[str] = [],
) -> None:
system_config_nix: dict[str, Any] = {}
if system_config.language:
if system_config.language not in list_possible_languages():
raise ClanError(
f"Language '{system_config.language}' is not a valid language. "
f"Run 'clan flash --list-languages' to see a list of possible languages."
)
system_config_nix["i18n"] = {"defaultLocale": system_config.language}
if system_config.keymap:
if system_config.keymap not in list_possible_keymaps():
raise ClanError(
f"Keymap '{system_config.keymap}' is not a valid keymap. "
f"Run 'clan flash --list-keymaps' to see a list of possible keymaps."
)
system_config_nix["console"] = {"keyMap": system_config.keymap}
if system_config.ssh_keys_path:
root_keys = []
for key_path in map(lambda x: Path(x), system_config.ssh_keys_path):
try:
root_keys.append(key_path.read_text())
except OSError as e:
raise ClanError(f"Cannot read SSH public key file: {key_path}: {e}")
system_config_nix["users"] = {
"users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}}
}
secret_facts_module = importlib.import_module(machine.secret_facts_module)
secret_facts_store: SecretStoreBase = secret_facts_module.SecretStore(
machine=machine
@@ -76,7 +158,7 @@ def flash_machine(
disko_install.extend(
[
"--system-config",
json.dumps(system_config),
json.dumps(system_config_nix),
]
)
disko_install.extend(["--option", "dry-run", "true"])
@@ -94,15 +176,13 @@ class FlashOptions:
flake: FlakeId
machine: str
disks: dict[str, str]
ssh_keys_path: list[Path]
dry_run: bool
confirm: bool
debug: bool
mode: str
language: str
keymap: str
write_efi_boot_entries: bool
nix_options: list[str]
system_config: SystemConfig
class AppendDiskAction(argparse.Action):
@@ -126,17 +206,29 @@ def flash_command(args: argparse.Namespace) -> None:
flake=args.flake,
machine=args.machine,
disks=args.disk,
ssh_keys_path=args.ssh_pubkey,
dry_run=args.dry_run,
confirm=not args.yes,
debug=args.debug,
mode=args.mode,
language=args.language,
keymap=args.keymap,
system_config=SystemConfig(
language=args.language,
keymap=args.keymap,
ssh_keys_path=args.ssh_pubkey,
),
write_efi_boot_entries=args.write_efi_boot_entries,
nix_options=args.option,
)
if args.list_languages:
for language in list_possible_languages():
print(language)
return
if args.list_keymaps:
for keymap in list_possible_keymaps():
print(keymap)
return
machine = Machine(opts.machine, flake=opts.flake)
if opts.confirm and not opts.dry_run:
disk_str = ", ".join(f"{name}={device}" for name, device in opts.disks.items())
@@ -148,28 +240,11 @@ def flash_command(args: argparse.Namespace) -> None:
if ask != "y":
return
extra_config: dict[str, Any] = {}
if opts.ssh_keys_path:
root_keys = []
for key_path in opts.ssh_keys_path:
try:
root_keys.append(key_path.read_text())
except OSError as e:
raise ClanError(f"Cannot read SSH public key file: {key_path}: {e}")
extra_config["users"] = {
"users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}}
}
if opts.keymap:
extra_config["console"] = {"keyMap": opts.keymap}
if opts.language:
extra_config["i18n"] = {"defaultLocale": opts.language}
flash_machine(
machine,
mode=opts.mode,
disks=opts.disks,
system_config=extra_config,
system_config=opts.system_config,
dry_run=opts.dry_run,
debug=opts.debug,
write_efi_boot_entries=opts.write_efi_boot_entries,
@@ -221,6 +296,18 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
type=str,
help="system language",
)
parser.add_argument(
"--list-languages",
help="List possible languages",
default=False,
action="store_true",
)
parser.add_argument(
"--list-keymaps",
help="List possible keymaps",
default=False,
action="store_true",
)
parser.add_argument(
"--keymap",
type=str,

50
pkgs/clan-cli/tests/test_deserializers.py
View File

@@ -19,6 +19,7 @@ from clan_cli.inventory import (
ServiceBorgbackupRoleServer,
ServiceMeta,
)
from clan_cli.machines import machines
def test_simple() -> None:
@@ -73,6 +74,55 @@ def test_nested() -> None:
assert from_dict(Person, person_dict) == expected_person
def test_nested_nullable() -> None:
@dataclass
class SystemConfig:
language: str | None = field(default=None)
keymap: str | None = field(default=None)
ssh_keys_path: list[str] | None = field(default=None)
@dataclass
class FlashOptions:
machine: machines.Machine
mode: str
disks: dict[str, str]
system_config: SystemConfig
dry_run: bool
write_efi_boot_entries: bool
debug: bool
data = {
"machine": {
"name": "flash-installer",
"flake": {"loc": "git+https://git.clan.lol/clan/clan-core"},
},
"mode": "format",
"disks": {"main": "/dev/sda"},
"system_config": {"language": "en_US.utf-8", "keymap": "en"},
"dry_run": False,
"write_efi_boot_entries": False,
"debug": False,
"op_key": "jWnTSHwYhSgr7Qz3u4ppD",
}
expected = FlashOptions(
machine=machines.Machine(
name="flash-installer",
flake=machines.FlakeId("git+https://git.clan.lol/clan/clan-core"),
),
mode="format",
disks={"main": "/dev/sda"},
system_config=SystemConfig(
language="en_US.utf-8", keymap="en", ssh_keys_path=None
),
dry_run=False,
write_efi_boot_entries=False,
debug=False,
)
assert from_dict(FlashOptions, data) == expected
def test_simple_field_missing() -> None:
@dataclass
class Person: