From f51aff3db5da677616246e412a385bf71075615e Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Mon, 2 Nov 2026 20:12:05 +0100 Subject: [PATCH 1/2] Add Textual TUI for managing Proxmox guests via cv4pve-cli List VMs and LXC containers from the current cv4pve-cli context and start, shutdown, stop and reboot them, change cores and memory, and grow disks. The app shells out to `cv4pve-cli -o json` and stores no credentials. - Every mutating call uses --wait: without it cv4pve-cli exits 0 even when the task fails. - shutdown, stop and reboot need confirmation; disk growth is increment-only and asks again because it cannot be undone. - After an action the acted-on guest's status is read from status/current, because /cluster/resources lags a few seconds. - Notifies when a cores/memory change is pending until reboot. Tests mock cv4pve-cli (85 tests). Live-tested edit, shutdown, start and reboot on CT 104 (mainframe), restored afterwards; disk resize was not run live. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 4 + README.md | 34 +++ docs/plans/guest-tui-mvp.md | 118 ++++++++++ pyproject.toml | 21 ++ src/cv4pve_tui/__init__.py | 1 + src/cv4pve_tui/app.py | 278 ++++++++++++++++++++++ src/cv4pve_tui/cli.py | 140 +++++++++++ tests/test_app.py | 449 ++++++++++++++++++++++++++++++++++++ tests/test_cli.py | 372 +++++++++++++++++++++++++++++ uv.lock | 195 ++++++++++++++++ 10 files changed, 1612 insertions(+) create mode 100644 .gitignore create mode 100644 docs/plans/guest-tui-mvp.md create mode 100644 pyproject.toml create mode 100644 src/cv4pve_tui/__init__.py create mode 100644 src/cv4pve_tui/app.py create mode 100644 src/cv4pve_tui/cli.py create mode 100644 tests/test_app.py create mode 100644 tests/test_cli.py create mode 100644 uv.lock diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..17cd2fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ diff --git a/README.md b/README.md index 5876510..1a47505 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,36 @@ # cv4pve-cli-tui +Textual TUI for [cv4pve-cli](https://github.com/Corsinvest/cv4pve-cli): list Proxmox VE +guests (VMs and LXC), start/shutdown/stop/reboot them, and change cores, memory and +disk size. + +It shells out to `cv4pve-cli` and uses its current context (`cv4pve-cli config use `); +it stores no credentials itself. Tested with cv4pve-cli 2.4.0. + +## Run + +```sh +uv run cv4pve-tui +``` + +| Key | Action | +|-----|--------| +| `s` | start | +| `S` | shutdown (confirm) | +| `x` | stop - hard power-off (confirm) | +| `r` | reboot (confirm) | +| `e` | edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) | +| `g` | refresh | +| `q` | quit | + +Empty fields in the edit dialog are left unchanged. If a CPU/memory change cannot be +hot-plugged, the app tells you it applies at the next reboot. + +## Development + +```sh +uv sync +uv run pytest +``` + +Tests mock `cv4pve-cli`; they never touch a cluster. Design: `docs/plans/guest-tui-mvp.md`. diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md new file mode 100644 index 0000000..2411be2 --- /dev/null +++ b/docs/plans/guest-tui-mvp.md @@ -0,0 +1,118 @@ +# Guest TUI MVP + +## Problem + +Manage Proxmox guests (VMs and LXC containers) from the terminal without remembering +`cv4pve-cli` syntax: list them, start/shutdown/stop/reboot, and change CPU cores, +memory and disk size. + +## Approach + +Thin subprocess wrapper around `cv4pve-cli` (tested against 2.4.0) plus a Textual app. +The app never handles credentials: it uses whatever context `cv4pve-cli config use` +selected. + +Observed CLI behaviour the design relies on: + +- `get guests -o json` -> list of dicts with `vmid`, `type` (`qemu`|`lxc`), `node`, + `name`, `status`, `maxcpu`, `maxmem` (bytes), `maxdisk` (bytes), ... +- `show vm|ct -o json` -> list of `{"key": ..., "value": ...}`. +- `api get /nodes////pending -o json` -> list of + `{"key", "value", "pending", "delete"}`; `pending` non-null means not yet applied. +- Async actions print a UPID and exit 0 **even when the task fails**. With `--wait` + the CLI waits for the task and exits non-zero (5) with the error text on failure. + Every mutating call therefore uses `--wait`. +- `do stop` refuses to run without `-y` (verified); `-y` is accepted by all four + lifecycle actions, so it is always passed. +- Errors: non-zero exit, message on stdout/stderr. Sync failures are multi-line + (`Error: ...` / `Call: ...` / `Alias: ...`). Success is judged by exit code only, + never by parsing UPID text. +- Each call costs ~0.5 s (.NET startup), so calls run in worker threads. +- `get guests` (`/cluster/resources`) lags a few seconds behind actions; after an + action the app reads the acted-on guest's live status from + `api get /nodes////status/current` (key/value list). + +### Module `cv4pve_tui/cli.py` + +- `CliError(Exception)` - `str()` is the CLI's combined, stripped output; + `.summary` is one line for the UI: the first non-empty line that is not a UPID, + `Error:`-prefix stripped, plus the following indented detail line if the first + line ends with `:`. +- `run(*args) -> str` - `subprocess.run(["cv4pve-cli", *args], capture_output=True, + text=True, stdin=DEVNULL, timeout=330)`; non-zero exit -> `CliError`. +- `kind(guest) -> "vm" | "ct"` - from `type` (`qemu` -> `vm`, `lxc` -> `ct`). +- `list_guests() -> list[dict]` - sorted by `vmid`. +- `guest_config(guest) -> dict` - key/value list flattened to a dict. +- `disks(config, kind) -> dict[str, str]` - resizable disks -> size string. + VM: `scsiN|virtioN|sataN|ideN` with a `size=`, excluding `media=cdrom` and + cloud-init. CT: `rootfs`, `mpN`. +- `ACTIONS = ("start", "shutdown", "stop", "reboot")`; + `action(name, guest)` -> `do -y --wait`; + unknown name -> `ValueError`. +- `set_resources(guest, cores=None, memory_mb=None)` -> + `set config [--cores N] [--memory M] --wait`; values must be + positive ints; nothing to set -> `ValueError`. +- `resize_disk(guest, disk, grow_gb)` -> `do resize disk + +G --wait`; `grow_gb` must be a positive int (growth only - Proxmox cannot + shrink, and an increment cannot shrink by construction). +- `pending_keys(guest) -> list[str]` - keys with a non-null `pending`. + +### Module `cv4pve_tui/app.py` + +- `DataTable` of guests: VMID, name, type, node, status, cores, memory (MiB), + disk (GiB). Cursor row = selected guest. +- Keys: `s` start, `S` shutdown, `x` stop (hard), `r` reboot, `e` edit, + `g` refresh, `q` quit. +- `shutdown`, `stop`, `reboot` open a confirm modal naming the guest; `stop` says it + is a hard power-off. `start` needs no confirmation. +- `e` opens an edit modal: cores, memory MiB, disk select + "grow by GiB". + Empty field = unchanged. Disk growth gets its own confirm (irreversible). +- After `set_resources`, if `pending_keys` contains `cores` or `memory`, notify that + the change applies at next reboot. +- All CLI calls in thread workers; results/errors via `notify` (errors show + `CliError.summary`); table refreshes after each action. +- A "working" notice is shown when an action starts (`--wait` can block minutes), + and a second action on a guest that is already busy is refused. +- Action keys on an empty table do nothing (no crash). + +### UI contract (shared by tests and implementation) + +- `app.py` does `from cv4pve_tui import cli` and always calls `cli.(...)`, + so tests monkeypatch `cv4pve_tui.cli.`. +- `class GuestApp(App)`; `def main()` runs it. +- Guest table: `DataTable` with id `guests`, one row per guest, row key = str(vmid), + in `list_guests()` order. Loaded on mount. +- `ConfirmScreen(ModalScreen[bool])`: buttons `#yes` / `#no`; its text names the guest + (name and vmid); for `stop` it contains the words "hard power-off". +- `EditScreen(ModalScreen)`: `Input#cores`, `Input#memory`, `Select#disk`, + `Input#grow`, buttons `#apply` / `#cancel`. Inputs are pre-filled with nothing + (empty = unchanged); placeholders show current values. Disk options come from + `cli.disks(cli.guest_config(guest), cli.kind(guest))`. + On apply: if cores/memory given -> `cli.set_resources(guest, cores=..., memory_mb=...)` + (only the given ones, as ints); if grow given -> `ConfirmScreen` first, then + `cli.resize_disk(guest, disk, grow_gb)`. Non-integer or <= 0 input -> `notify` + with severity "error" and no CLI call. +- Thread workers (`@work(thread=True)` or `run_worker(..., thread=True)`); tests wait + with `await app.workers.wait_for_complete()` then `await pilot.pause()`. + +## Out of scope + +Create/delete, snapshots, migration, console, backup, in-app context switching, +sockets/vCPU/hotplug settings, live auto-refresh. + +## Steps + +- [x] Project skeleton (`uv`, `pyproject.toml`, entry point `cv4pve-tui`) - + `uv run python -c "import cv4pve_tui"` succeeds. +- [x] Tests for `cli.py` with `subprocess.run` mocked - fail before implementation. +- [x] `cli.py` - tests pass. +- [x] Tests for the app with Textual `run_test` and `cli` functions mocked. +- [x] `app.py` - tests pass. +- [x] Read-only smoke test against the real cluster (listing, edit dialog disk options, + pending check). +- [x] Live edit (cores/memory), shutdown, start, reboot on CT 104 (mainframe), restored + to baseline afterwards. Disk resize not run live (irreversible). + +## Open questions + +None blocking. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2fc7a53 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[project] +name = "cv4pve-tui" +version = "0.1.0" +description = "Textual TUI for cv4pve-cli: manage Proxmox VE guests" +readme = "README.md" +requires-python = ">=3.12" +dependencies = ["textual>=8.2"] + +[project.scripts] +cv4pve-tui = "cv4pve_tui.app:main" + +[dependency-groups] +dev = ["pytest>=8", "pytest-asyncio>=1.0"] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.pytest.ini_options] +testpaths = ["tests"] +asyncio_mode = "auto" diff --git a/src/cv4pve_tui/__init__.py b/src/cv4pve_tui/__init__.py new file mode 100644 index 0000000..461396e --- /dev/null +++ b/src/cv4pve_tui/__init__.py @@ -0,0 +1 @@ +"""Textual TUI for cv4pve-cli.""" diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py new file mode 100644 index 0000000..e2c3df7 --- /dev/null +++ b/src/cv4pve_tui/app.py @@ -0,0 +1,278 @@ +"""Textual app for managing Proxmox guests through cv4pve-cli.""" + +from textual import work +from textual.app import App, ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Button, DataTable, Footer, Header, Input, Select, Static + +from cv4pve_tui import cli + + +def _message(exc: Exception) -> str: + return ( + exc.summary if isinstance(exc, cli.CliError) else f"{type(exc).__name__}: {exc}" + ) + + +class ConfirmScreen(ModalScreen[bool]): + BINDINGS = [("escape", "dismiss(False)", "No")] # noqa: RUF012 - Textual idiom + + def __init__(self, message: str) -> None: + super().__init__() + self.message = message + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static(self.message) + with Horizontal(): + yield Button("Yes", id="yes", variant="error") + yield Button("No", id="no") + + def on_button_pressed(self, event: Button.Pressed) -> None: + self.dismiss(event.button.id == "yes") + + +class EditScreen(ModalScreen): + """Dismisses with a dict of validated values, or None when cancelled.""" + + BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom + + def __init__(self, guest: dict, config: dict, disks: dict[str, str]) -> None: + super().__init__() + self.guest, self.config, self.disks = guest, config, disks + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static( + f"Edit {self.guest['name']} ({self.guest['vmid']}); empty = unchanged" + ) + yield Input( + placeholder=f"cores (now {self.config.get('cores', '?')})", id="cores" + ) + yield Input( + placeholder=f"memory MiB (now {self.config.get('memory', '?')})", + id="memory", + ) + yield Select( + [(f"{k} ({v})", k) for k, v in self.disks.items()], + prompt="disk", + id="disk", + ) + yield Input(placeholder="grow by GiB", id="grow") + with Horizontal(): + yield Button("Apply", id="apply", variant="primary") + yield Button("Cancel", id="cancel") + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "cancel": + self.dismiss(None) + return + try: + values = {name: self._number(name) for name in ("cores", "memory", "grow")} + except ValueError as exc: + self.app.notify(str(exc), severity="error") + return + disk = self.query_one("#disk", Select).value + if values["grow"] is not None and disk is Select.NULL: + self.app.notify("Choose a disk to grow", severity="error") + return + values["disk"] = None if disk is Select.NULL else disk + self.dismiss(values) + + def _number(self, name: str) -> int | None: + text = self.query_one(f"#{name}", Input).value.strip() + if not text: + return None + if not text.isdecimal() or int(text) <= 0: + raise ValueError(f"{name} must be a positive integer") + return int(text) + + +class GuestApp(App): + BINDINGS = [ # noqa: RUF012 - Textual idiom + ("s", "act('start')", "Start"), + ("S", "act('shutdown')", "Shutdown"), + ("x", "act('stop')", "Stop"), + ("r", "act('reboot')", "Reboot"), + ("e", "edit", "Edit"), + ("g", "reload", "Refresh"), + ("q", "quit", "Quit"), + ] + + def __init__(self) -> None: + super().__init__() + self.guests: dict[str, dict] = {} + self.busy: set[int] = set() + + def compose(self) -> ComposeResult: + yield Header() + yield DataTable(id="guests", cursor_type="row") + yield Footer() + + def on_mount(self) -> None: + table = self.query_one("#guests", DataTable) + table.add_columns( + "VMID", + "Name", + "Type", + "Node", + "Status", + "Cores", + "Memory (MiB)", + "Disk (GiB)", + ) + self.action_reload() + + def _selected(self) -> dict | None: + table = self.query_one("#guests", DataTable) + if not table.row_count: + return None + return self.guests[ + table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value + ] + + def _fill(self, guests: list[dict]) -> None: + table = self.query_one("#guests", DataTable) + current = self._selected() + table.clear() + self.guests = {str(g["vmid"]): g for g in guests} + for g in guests: + table.add_row( + g["vmid"], g["name"], g["type"], g["node"], g["status"], + g.get("maxcpu", "-"), _scaled(g.get("maxmem"), 2**20), _scaled(g.get("maxdisk"), 2**30), + key=str(g["vmid"]), + ) # fmt: skip + if current and str(current["vmid"]) in self.guests: + table.move_cursor(row=table.get_row_index(str(current["vmid"]))) + + def _reload(self, acted: dict | None = None) -> None: + """Runs in a worker thread. `acted` gets its live status: the listing lags.""" + try: + guests = cli.list_guests() + for g in guests: + if acted and g["vmid"] == acted["vmid"]: + g["status"] = cli.status(g) + except Exception as exc: # noqa: BLE001 - any failure must reach the user, not kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + return + self.call_from_thread(self._fill, guests) + + @work(thread=True) + def action_reload(self) -> None: + self._reload() + + def _start_job(self, guest: dict, label: str, job) -> None: + if guest["vmid"] in self.busy: + self.notify( + f"{guest['name']} ({guest['vmid']}) is busy", severity="warning" + ) + return + self.busy.add(guest["vmid"]) + self.notify(f"{label} {guest['name']} ({guest['vmid']}) ...") + self._run_job(guest, label, job) + + @work(thread=True) + def _run_job(self, guest: dict, label: str, job) -> None: + try: + job() + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread( + self.notify, f"{label} failed: {_message(exc)}", severity="error" + ) + else: + self.call_from_thread( + self.notify, f"{label} done: {guest['name']} ({guest['vmid']})" + ) + finally: + self.busy.discard(guest["vmid"]) + self._reload(acted=guest) + + def action_act(self, name: str) -> None: + guest = self._selected() + if guest is None: + return + run = lambda: cli.action(name, guest) + if name == "start": + self._start_job(guest, name, run) + return + who = f"{guest['name']} ({guest['vmid']})" + text = f"{name.capitalize()} {who}?" + if name == "stop": + text = f"Hard power-off {who}? The guest is killed without shutting down." + self.push_screen( + ConfirmScreen(text), + lambda ok: self._start_job(guest, name, run) if ok else None, + ) + + def action_edit(self) -> None: + guest = self._selected() + if guest is None: + return + if guest["vmid"] in self.busy: + self.notify( + f"{guest['name']} ({guest['vmid']}) is busy", severity="warning" + ) + return + self._open_edit(guest) + + @work(thread=True) + def _open_edit(self, guest: dict) -> None: + try: + config = cli.guest_config(guest) + disks = cli.disks(config, cli.kind(guest)) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + return + self.call_from_thread( + self.push_screen, + EditScreen(guest, config, disks), + lambda v: self._edit_done(guest, v), + ) + + def _edit_done(self, guest: dict, values: dict | None) -> None: + if not values or not any(values[k] for k in ("cores", "memory", "grow")): + return + if values["grow"]: + who = f"{guest['name']} ({guest['vmid']})" + self.push_screen( + ConfirmScreen( + f"Grow {values['disk']} of {who} by {values['grow']} GiB? This cannot be undone." + ), + lambda ok: ( + self._apply(guest, values) if ok else None + ), # "No" cancels the whole edit + ) + else: + self._apply(guest, values) + + def _apply(self, guest: dict, values: dict) -> None: + cores, memory, grow = values["cores"], values["memory"], values["grow"] + + def job() -> None: + if cores or memory: + cli.set_resources(guest, cores=cores, memory_mb=memory) + if grow: + cli.resize_disk(guest, values["disk"], grow) + if cores or memory: + try: + pending = cli.pending_keys(guest) + except Exception as exc: # noqa: BLE001 - only a notice; the change itself succeeded + self.call_from_thread( + self.notify, + f"pending check failed: {_message(exc)}", + severity="warning", + ) + return + if {"cores", "memory"} & set(pending): + self.call_from_thread(self.notify, "Change applies at next reboot") + + self._start_job(guest, "edit", job) + + +def _scaled(value, unit: int): + return "-" if value is None else round(value / unit) + + +def main() -> None: + GuestApp().run() diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py new file mode 100644 index 0000000..d665612 --- /dev/null +++ b/src/cv4pve_tui/cli.py @@ -0,0 +1,140 @@ +"""Thin subprocess wrapper around cv4pve-cli.""" + +import json +import re +import subprocess + + +class CliError(Exception): + """cv4pve-cli failed; str() is its combined, stripped output.""" + + @property + def summary(self) -> str: + """One line for the UI: first non-UPID line, plus its indented detail line.""" + lines = str(self).splitlines() + for i, line in enumerate(lines): + text = line.strip() + if not text or text.startswith("UPID:"): + continue + text = text.removeprefix("Error:").strip() + if text.endswith(":") and i + 1 < len(lines) and lines[i + 1][:1].isspace(): + text += " " + lines[i + 1].strip() + return text + return str(self) + + +def run(*args: str) -> str: + try: + proc = subprocess.run( + ["cv4pve-cli", *args], + check=False, + capture_output=True, + text=True, + stdin=subprocess.DEVNULL, + timeout=330, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise CliError(str(exc)) from exc + if proc.returncode != 0: + raise CliError( + "\n".join(s.strip() for s in (proc.stdout, proc.stderr) if s.strip()) + ) + return proc.stdout + + +def kind(guest: dict) -> str: + return "vm" if guest["type"] == "qemu" else "ct" + + +def _api_type(guest: dict) -> str: + return "qemu" if guest["type"] == "qemu" else "lxc" + + +def list_guests() -> list[dict]: + guests = json.loads(run("get", "guests", "-o", "json")) + for g in guests: + g["name"] = ( + g.get("name") or f"VM {g['vmid']}" + ) # unnamed guests have no/null name + return sorted(guests, key=lambda g: g["vmid"]) + + +def guest_config(guest: dict) -> dict: + out = run("show", kind(guest), guest["node"], str(guest["vmid"]), "-o", "json") + return {item["key"]: item["value"] for item in json.loads(out)} + + +_VM_DISK = re.compile(r"(scsi|virtio|sata|ide)\d+") +_CT_DISK = re.compile(r"rootfs|mp\d+") + + +def disks(config: dict, kind: str) -> dict[str, str]: + pattern = _VM_DISK if kind == "vm" else _CT_DISK + result = {} + for key, value in config.items(): + if not pattern.fullmatch(key): + continue + value = str(value) + size = re.search(r"size=(\w+)", value) + if size and "media=cdrom" not in value and "cloudinit" not in value: + result[key] = size.group(1) + return result + + +ACTIONS = ("start", "shutdown", "stop", "reboot") + + +def action(name: str, guest: dict) -> None: + if name not in ACTIONS: + raise ValueError(f"unknown action: {name}") + run("do", name, kind(guest), guest["node"], str(guest["vmid"]), "-y", "--wait") + + +def _positive(value, what: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise ValueError(f"{what} must be a positive integer") + return value + + +def set_resources( + guest: dict, cores: int | None = None, memory_mb: int | None = None +) -> None: + opts = [] + if cores is not None: + opts += ["--cores", str(_positive(cores, "cores"))] + if memory_mb is not None: + opts += ["--memory", str(_positive(memory_mb, "memory"))] + if not opts: + raise ValueError("nothing to set") + run( + "set", kind(guest), "config", guest["node"], str(guest["vmid"]), *opts, "--wait" + ) + + +def resize_disk(guest: dict, disk: str, grow_gb: int) -> None: + _positive(grow_gb, "grow") + run( + "do", + "resize", + kind(guest), + "disk", + guest["node"], + str(guest["vmid"]), + disk, + f"+{grow_gb}G", + "--wait", + ) + + +def pending_keys(guest: dict) -> list[str]: + path = f"/nodes/{guest['node']}/{_api_type(guest)}/{guest['vmid']}/pending" + items = json.loads(run("api", "get", path, "-o", "json")) + return [i["key"] for i in items if i.get("pending") is not None] + + +def status(guest: dict) -> str: + """Live status; get guests (/cluster/resources) lags a few seconds behind actions.""" + path = f"/nodes/{guest['node']}/{_api_type(guest)}/{guest['vmid']}/status/current" + return { + i["key"]: i["value"] for i in json.loads(run("api", "get", path, "-o", "json")) + }["status"] diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..3ddd6a1 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,449 @@ +"""Textual pilot tests for cv4pve_tui.app; spec: docs/plans/guest-tui-mvp.md ("UI contract"). + +All cli functions are monkeypatched: cv4pve-cli is never executed. +""" + +import subprocess +import threading + +import pytest +from textual.screen import ModalScreen +from textual.widgets import Button, DataTable, Input, Select, Static + +from cv4pve_tui import cli +from cv4pve_tui.app import ConfirmScreen, EditScreen, GuestApp + +VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu", "status": "running", + "maxcpu": 4, "maxmem": 8192 * 2**20, "maxdisk": 10 * 2**30} # fmt: skip +CT = {"vmid": 100, "name": "backrest", "node": "mainframe", "type": "lxc", "status": "stopped", + "maxcpu": 1, "maxmem": 512 * 2**20, "maxdisk": 8 * 2**30} # fmt: skip + + +class Fake: + """Recorder standing in for the cli module functions the app calls.""" + + def __init__(self): + self.guests = [VM, CT] + self.pending = [] + self.disk_map = {"scsi0": "10G"} + self.list_calls = 0 + self.actions, self.sets, self.resizes, self.pending_calls = [], [], [], [] + self.action_error = None + self.action_hook = None + self.live_status, self.status_calls = {}, [] + + def list_guests(self): + self.list_calls += 1 + return list(self.guests) + + def guest_config(self, guest): + return {"cores": guest["maxcpu"]} + + def disks(self, config, kind): + return dict(self.disk_map) + + def action(self, name, guest): + self.actions.append((name, guest["vmid"])) + if self.action_hook: + self.action_hook() + if self.action_error: + raise self.action_error + + def set_resources(self, guest, cores=None, memory_mb=None): + given = {"cores": cores, "memory_mb": memory_mb} + self.sets.append({k: v for k, v in given.items() if v is not None}) + + def resize_disk(self, guest, disk, grow_gb): + self.resizes.append((guest["vmid"], disk, grow_gb)) + + def pending_keys(self, guest): + self.pending_calls.append(guest["vmid"]) + return list(self.pending) + + def status(self, guest): + self.status_calls.append(guest["vmid"]) + return self.live_status.get(guest["vmid"], guest["status"]) + + +@pytest.fixture +def fake(monkeypatch): + f = Fake() + for name in ("list_guests", "guest_config", "disks", "action", "set_resources", + "resize_disk", "pending_keys", "status"): # fmt: skip + monkeypatch.setattr(cli, name, getattr(f, name)) + return f + + +def make_app(): + app = GuestApp() + app.seen = [] # (message, severity) of every notify + original = app.notify + + def spy(message, **kwargs): + app.seen.append((str(message), kwargs.get("severity", "information"))) + return original(message, **kwargs) + + app.notify = spy + return app + + +async def settle(app, pilot): + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + + +def text_of(screen): + return " ".join(str(w.render()) for w in screen.query(Static)) + + +def errors(app): + return [m for m, sev in app.seen if sev == "error"] + + +async def press_button(app, pilot, selector): + app.screen.query_one(selector, Button).press() + await settle(app, pilot) + + +# --- table ------------------------------------------------------------------ + + +async def test_table_loads_one_row_per_guest_in_list_order(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + table = app.query_one("#guests", DataTable) + assert table.row_count == 2 + assert [k.value for k in table.rows] == ["108", "100"] + cells = [str(c) for c in table.get_row("108")] + assert "docker" in cells and "mainframe" in cells and "running" in cells + assert "8192" in cells # memory in MiB + + +async def test_refresh_key_reloads_guests(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + before = fake.list_calls + fake.guests = [CT] + await pilot.press("g") + await settle(app, pilot) + assert fake.list_calls == before + 1 + assert app.query_one("#guests", DataTable).row_count == 1 + + +# --- lifecycle actions ------------------------------------------------------ + + +async def test_start_runs_without_confirm_and_refreshes_table(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + before = fake.list_calls + await pilot.press("s") + await settle(app, pilot) + assert fake.actions == [("start", 108)] + assert not isinstance(app.screen, ModalScreen) + assert fake.list_calls > before + + +async def test_action_targets_the_cursor_row(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press("down", "s") + await settle(app, pilot) + assert fake.actions == [("start", 100)] + + +@pytest.mark.parametrize( + "key,name", [("S", "shutdown"), ("x", "stop"), ("r", "reboot")] +) +async def test_confirmed_action_calls_cli_only_after_yes(fake, key, name): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press(key) + await pilot.pause() + assert isinstance(app.screen, ConfirmScreen) + assert "docker" in text_of(app.screen) and "108" in text_of(app.screen) + assert fake.actions == [] + await press_button(app, pilot, "#yes") + assert fake.actions == [(name, 108)] + + +@pytest.mark.parametrize("key", ["S", "x", "r"]) +async def test_declined_action_does_not_call_cli(fake, key): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press(key) + await pilot.pause() + await press_button(app, pilot, "#no") + assert fake.actions == [] + assert not isinstance(app.screen, ModalScreen) + + +async def test_stop_confirm_says_hard_power_off(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press("x") + await pilot.pause() + assert "hard power-off" in text_of(app.screen).lower() + + +@pytest.mark.parametrize("key", ["s", "S", "x", "r", "e"]) +async def test_keys_on_empty_table_do_nothing(fake, key): + fake.guests = [] + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press(key) + await settle(app, pilot) + assert fake.actions == [] and fake.sets == [] and fake.resizes == [] + assert not isinstance(app.screen, ModalScreen) + + +async def test_cli_error_is_notified_with_summary(fake, monkeypatch): + with monkeypatch.context() as m: # build a real CliError through cli.run + m.setattr(cli.subprocess, "run", lambda argv, **kw: subprocess.CompletedProcess( + argv, 5, "UPID:mainframe:0:0:0:qmstart:108:root@pam:\nstart failed: no quorum\n", "")) # fmt: skip + with pytest.raises(cli.CliError) as exc: + cli.run("do", "start") + fake.action_error = exc.value + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press("s") + await settle(app, pilot) + assert any(exc.value.summary in m for m in errors(app)) + assert not any("UPID" in m for m in errors(app)) + + +async def test_busy_guest_refuses_second_action(fake): + started, release = threading.Event(), threading.Event() + fake.action_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + try: + await pilot.press("s") + for _ in range(200): + if started.is_set(): + break + await pilot.pause(0.01) + assert started.is_set() + assert app.seen, "a working notice is expected while the action runs" + await pilot.press("s") + await pilot.pause() + assert fake.actions == [("start", 108)] + finally: + release.set() + await settle(app, pilot) + assert fake.actions == [("start", 108)] + + +# --- edit flow -------------------------------------------------------------- + + +async def open_edit(app, pilot): + await settle(app, pilot) + await pilot.press("e") + await settle(app, pilot) + assert isinstance(app.screen, EditScreen) + + +def fill(app, **values): + for key, value in values.items(): + w = app.screen.query_one(f"#{key}", Select if key == "disk" else Input) + w.value = value + + +async def test_edit_applies_only_given_fields_as_ints(fake): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, cores="8") + await press_button(app, pilot, "#apply") + assert fake.sets == [{"cores": 8}] + await pilot.press("e") + await settle(app, pilot) + fill(app, cores="2", memory="4096") + await press_button(app, pilot, "#apply") + assert fake.sets[-1] == {"cores": 2, "memory_mb": 4096} + assert all(isinstance(v, int) for v in fake.sets[-1].values()) + + +@pytest.mark.parametrize("field", ["cores", "memory"]) +@pytest.mark.parametrize("bad", ["abc", "0", "-4", "2.5"]) +async def test_edit_invalid_resource_input_errors_without_cli_call(fake, field, bad): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, **{field: bad}) + await press_button(app, pilot, "#apply") + assert errors(app) + assert fake.sets == [] and fake.resizes == [] + + +@pytest.mark.parametrize("bad", ["abc", "0", "-1"]) +async def test_edit_invalid_grow_errors_without_cli_call(fake, bad): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, disk="scsi0", grow=bad) + await press_button(app, pilot, "#apply") + assert errors(app) + assert fake.resizes == [] + + +async def test_edit_with_empty_fields_changes_nothing(fake): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + await press_button(app, pilot, "#apply") + assert fake.sets == [] and fake.resizes == [] + + +async def test_edit_cancel_changes_nothing(fake): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, cores="8") + await press_button(app, pilot, "#cancel") + assert fake.sets == [] and fake.resizes == [] + assert not isinstance(app.screen, ModalScreen) + + +async def test_grow_needs_confirm_then_resizes(fake): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, disk="scsi0", grow="5") + await press_button(app, pilot, "#apply") + assert isinstance(app.screen, ConfirmScreen) + assert fake.resizes == [] + await press_button(app, pilot, "#yes") + assert fake.resizes == [(108, "scsi0", 5)] + + +async def test_grow_declined_does_not_resize(fake): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, disk="scsi0", grow="5") + await press_button(app, pilot, "#apply") + await press_button(app, pilot, "#no") + assert fake.resizes == [] + + +@pytest.mark.parametrize( + "pending,expect_notice", [(["memory"], True), (["cores"], True), ([], False)] +) +async def test_pending_cores_or_memory_notifies_reboot(fake, pending, expect_notice): + fake.pending = pending + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, memory="4096") + await press_button(app, pilot, "#apply") + assert fake.pending_calls == [108] + assert any("reboot" in m.lower() for m, _ in app.seen) is expect_notice + + +# --- review findings: robustness -------------------------------------------- + + +async def test_guest_without_size_fields_still_listed(fake): + # e.g. a guest on an unreachable node: the API omits maxcpu/maxmem/maxdisk + fake.guests = [ + { + "vmid": 999, + "name": "ghost", + "node": "down", + "type": "qemu", + "status": "unknown", + } + ] + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + assert app.query_one("#guests", DataTable).row_count == 1 + + +async def test_unexpected_error_in_action_is_notified_not_fatal(fake): + fake.action_error = KeyError("key") + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press("s") + await settle(app, pilot) + assert errors(app) + assert app.is_running and app.busy == set() + + +async def test_unexpected_error_in_listing_is_notified_not_fatal(fake, monkeypatch): + def broken(): + raise ValueError("bad json") + + monkeypatch.setattr(cli, "list_guests", broken) + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + assert errors(app) and app.is_running + + +async def test_pending_check_failure_does_not_skip_confirmed_grow(fake, monkeypatch): + def broken(guest): + raise cli.CliError("Error: boom") + + monkeypatch.setattr(cli, "pending_keys", broken) + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, memory="4096", disk="scsi0", grow="5") + await press_button(app, pilot, "#apply") + await press_button(app, pilot, "#yes") + assert fake.sets == [{"memory_mb": 4096}] + assert fake.resizes == [(108, "scsi0", 5)] + + +async def test_declining_grow_cancels_whole_edit(fake): + app = make_app() + async with app.run_test() as pilot: + await open_edit(app, pilot) + fill(app, cores="8", disk="scsi0", grow="5") + await press_button(app, pilot, "#apply") + await press_button(app, pilot, "#no") + assert fake.sets == [] and fake.resizes == [] + + +@pytest.mark.parametrize("key", ["S", "e"]) +async def test_escape_closes_modal_without_cli_call(fake, key): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press(key) + await settle(app, pilot) + assert isinstance(app.screen, ModalScreen) + await pilot.press("escape") + await settle(app, pilot) + assert not isinstance(app.screen, ModalScreen) + assert fake.actions == [] and fake.sets == [] + + +async def test_table_shows_live_status_after_action_not_lagging_list(fake): + # /cluster/resources lags a few seconds behind; the acted-on guest must show live status + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + fake.live_status = {108: "stopped"} # list_guests still says "running" + await pilot.press("S") + await settle(app, pilot) + await press_button(app, pilot, "#yes") + assert fake.status_calls == [108] + assert "stopped" in [ + str(c) for c in app.query_one("#guests", DataTable).get_row("108") + ] diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..db4e3c4 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,372 @@ +"""Tests for cv4pve_tui.cli; spec: docs/plans/guest-tui-mvp.md ("Module cv4pve_tui/cli.py"). + +subprocess.run is always mocked: cv4pve-cli is never executed. +""" + +import json +import subprocess + +import pytest + +from cv4pve_tui import cli + +VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu"} +CT = {"vmid": 100, "name": "backrest", "node": "mainframe", "type": "lxc"} + + +class FakeRun: + def __init__(self, monkeypatch, stdout="", stderr="", returncode=0): + self.calls = [] + self.result = (stdout, stderr, returncode) + monkeypatch.setattr(cli.subprocess, "run", self) + + def __call__(self, argv, **kwargs): + self.calls.append((list(argv), kwargs)) + out, err, rc = self.result + return subprocess.CompletedProcess(argv, rc, out, err) + + @property + def argv(self): + assert len(self.calls) == 1 + return self.calls[0][0] + + +@pytest.fixture +def fake(monkeypatch): + return FakeRun(monkeypatch) + + +def error_for(monkeypatch, stdout, stderr="", returncode=1): + FakeRun(monkeypatch, stdout, stderr, returncode) + with pytest.raises(cli.CliError) as exc: + cli.run("do", "start") + return exc.value + + +# --- run / CliError --------------------------------------------------------- + + +def test_run_passes_exact_subprocess_arguments(fake): + fake.result = ("ok\n", "", 0) + assert cli.run("get", "guests").strip() == "ok" + argv, kwargs = fake.calls[0] + assert argv == ["cv4pve-cli", "get", "guests"] + assert kwargs["capture_output"] is True + assert kwargs["text"] is True + assert kwargs["stdin"] == subprocess.DEVNULL + assert kwargs["timeout"] == 330 + + +def test_run_nonzero_exit_raises_clierror_with_stripped_output(monkeypatch): + err = error_for(monkeypatch, " Error: boom\n\n", returncode=5) + assert str(err) == "Error: boom" + + +def test_clierror_includes_stderr_when_stdout_is_empty(monkeypatch): + err = error_for(monkeypatch, "", stderr="Error: only on stderr\n") + assert "only on stderr" in str(err) + + +def test_summary_multiline_sync_failure(monkeypatch): + out = ( + "Error: Proxmox VE refused the call (HTTP 500):\n" + " Configuration file 'nodes/mainframe/qemu-server/999.conf' does not exist\n" + "\n" + "Call: POST /nodes/mainframe/qemu/999/status/start\n" + "Alias: prod\n" + ) + summary = error_for(monkeypatch, out).summary + assert "\n" not in summary + assert summary.startswith("Proxmox VE refused the call (HTTP 500):") + assert "Error:" not in summary + assert "Configuration file" in summary and "does not exist" in summary + assert "Call:" not in summary and "Alias:" not in summary + + +def test_summary_skips_upid_line_exit_5(monkeypatch): + out = "UPID:mainframe:0001:0002:0003:qmstart:999:root@pam:\nConfiguration file 'x' does not exist\n" + summary = error_for(monkeypatch, out, returncode=5).summary + assert summary == "Configuration file 'x' does not exist" + + +def test_summary_single_line_strips_error_prefix(monkeypatch): + assert error_for(monkeypatch, "Error: no such guest\n").summary == "no such guest" + + +# --- kind / list_guests / guest_config -------------------------------------- + + +def test_kind_maps_type(): + assert cli.kind({"type": "qemu"}) == "vm" + assert cli.kind({"type": "lxc"}) == "ct" + + +def test_list_guests_argv_and_sorted_by_vmid(fake): + fake.result = (json.dumps([CT | {"vmid": 200}, VM, CT]), "", 0) + guests = cli.list_guests() + assert fake.argv == ["cv4pve-cli", "get", "guests", "-o", "json"] + assert [g["vmid"] for g in guests] == [100, 108, 200] + + +def test_guest_config_flattens_key_value_list(fake): + fake.result = ( + json.dumps([{"key": "cores", "value": 4}, {"key": "memory", "value": 8192}]), + "", + 0, + ) + assert cli.guest_config(VM) == {"cores": 4, "memory": 8192} + assert fake.argv == ["cv4pve-cli", "show", "vm", "mainframe", "108", "-o", "json"] + + +def test_guest_config_uses_ct_for_lxc(fake): + fake.result = ("[]", "", 0) + cli.guest_config(CT) + assert fake.argv == ["cv4pve-cli", "show", "ct", "mainframe", "100", "-o", "json"] + + +# --- disks ------------------------------------------------------------------ + + +def test_disks_vm_keeps_sized_disks_only(): + config = { + "cores": 4, + "memory": 8192, + "scsi0": "local-lvm:vm-108-disk-0,discard=on,size=10G,ssd=1", + "virtio1": "local-lvm:vm-108-disk-2,size=32G", + "ide2": "local-lvm:vm-108-cloudinit,media=cdrom", + "sata0": "local-lvm:vm-108-cloudinit,size=4M", # cloud-init drive without media=cdrom + "ide3": "none,media=cdrom", + "efidisk0": "local-lvm:vm-108-disk-1,efitype=4m,size=4M", + "net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0", + } + assert cli.disks(config, "vm") == {"scsi0": "10G", "virtio1": "32G"} + + +def test_disks_ct_rootfs_and_mount_points(): + config = { + "cores": 1, + "memory": 512, + "rootfs": "local-lvm:vm-100-disk-0,size=8G", + "mp0": "local-lvm:vm-100-disk-1,mp=/data,size=2G", + "net0": "name=eth0,bridge=vmbr0", + } + assert cli.disks(config, "ct") == {"rootfs": "8G", "mp0": "2G"} + + +def test_disks_without_disks_is_empty(): + assert cli.disks({"cores": 1}, "vm") == {} + assert cli.disks({"cores": 1}, "ct") == {} + + +# --- action ----------------------------------------------------------------- + + +def test_actions_constant(): + assert cli.ACTIONS == ("start", "shutdown", "stop", "reboot") + + +@pytest.mark.parametrize("name", cli.ACTIONS) +def test_action_argv_vm(fake, name): + cli.action(name, VM) + assert fake.argv == [ + "cv4pve-cli", + "do", + name, + "vm", + "mainframe", + "108", + "-y", + "--wait", + ] + + +def test_action_argv_ct(fake): + cli.action("stop", CT) + assert fake.argv == [ + "cv4pve-cli", + "do", + "stop", + "ct", + "mainframe", + "100", + "-y", + "--wait", + ] + + +def test_action_unknown_name_raises_and_does_not_call(fake): + with pytest.raises(ValueError): + cli.action("destroy", VM) + assert fake.calls == [] + + +def test_action_failure_raises_clierror(monkeypatch): + FakeRun(monkeypatch, "UPID:x:\nstart failed\n", returncode=5) + with pytest.raises(cli.CliError): + cli.action("start", VM) + + +# --- set_resources ---------------------------------------------------------- + + +def test_set_resources_cores_and_memory(fake): + cli.set_resources(VM, cores=4, memory_mb=8192) + assert fake.argv == [ + "cv4pve-cli", "set", "vm", "config", "mainframe", "108", + "--cores", "4", "--memory", "8192", "--wait", + ] # fmt: skip + + +def test_set_resources_only_given_fields(fake): + cli.set_resources(CT, memory_mb=1024) + assert fake.argv == [ + "cv4pve-cli", + "set", + "ct", + "config", + "mainframe", + "100", + "--memory", + "1024", + "--wait", + ] + fake.calls.clear() + cli.set_resources(CT, cores=2) + assert fake.argv == [ + "cv4pve-cli", + "set", + "ct", + "config", + "mainframe", + "100", + "--cores", + "2", + "--wait", + ] + + +def test_set_resources_nothing_to_set(fake): + with pytest.raises(ValueError): + cli.set_resources(VM) + assert fake.calls == [] + + +@pytest.mark.parametrize( + "kwargs", [{"cores": 0}, {"cores": -1}, {"memory_mb": 0}, {"memory_mb": -512}] +) +def test_set_resources_rejects_non_positive(fake, kwargs): + with pytest.raises(ValueError): + cli.set_resources(VM, **kwargs) + assert fake.calls == [] + + +# --- resize_disk ------------------------------------------------------------ + + +def test_resize_disk_argv(fake): + cli.resize_disk(VM, "scsi0", 5) + assert fake.argv == [ + "cv4pve-cli", + "do", + "resize", + "vm", + "disk", + "mainframe", + "108", + "scsi0", + "+5G", + "--wait", + ] + + +def test_resize_disk_ct_argv(fake): + cli.resize_disk(CT, "rootfs", 2) + assert fake.argv == [ + "cv4pve-cli", + "do", + "resize", + "ct", + "disk", + "mainframe", + "100", + "rootfs", + "+2G", + "--wait", + ] + + +@pytest.mark.parametrize("grow", [0, -1, -10]) +def test_resize_disk_growth_must_be_positive(fake, grow): + with pytest.raises(ValueError): + cli.resize_disk(VM, "scsi0", grow) + assert fake.calls == [] + + +# --- pending_keys ----------------------------------------------------------- + + +def test_pending_keys_only_non_null_pending(fake): + fake.result = ( + json.dumps([ + {"key": "cores", "value": 4, "pending": None, "delete": None}, + {"key": "memory", "value": 8192, "pending": 16384}, + {"key": "balloon", "value": 0, "pending": 0}, + ]), + "", + 0, + ) # fmt: skip + assert sorted(cli.pending_keys(VM)) == ["balloon", "memory"] + assert fake.argv == [ + "cv4pve-cli", + "api", + "get", + "/nodes/mainframe/qemu/108/pending", + "-o", + "json", + ] + + +def test_pending_keys_ct_path_and_none_pending(fake): + fake.result = (json.dumps([{"key": "cores", "value": 1, "pending": None}]), "", 0) + assert cli.pending_keys(CT) == [] + assert fake.argv[3] == "/nodes/mainframe/lxc/100/pending" + + +def test_missing_binary_or_timeout_becomes_cli_error(monkeypatch): + for exc in ( + FileNotFoundError("cv4pve-cli"), + subprocess.TimeoutExpired("cv4pve-cli", 330), + ): + + def boom(*a, _exc=exc, **k): + raise _exc + + monkeypatch.setattr(subprocess, "run", boom) + with pytest.raises(cli.CliError): + cli.run("get", "version") + + +@pytest.mark.parametrize("raw", [{}, {"name": None}, {"name": ""}]) +def test_list_guests_fills_missing_name(fake, raw): + guest = {k: v for k, v in VM.items() if k != "name"} | raw + fake.result = (json.dumps([guest]), "", 0) + assert cli.list_guests()[0]["name"] == "VM 108" + + +def test_status_reads_live_status_current(fake): + fake.result = ( + json.dumps( + [{"key": "cpus", "value": 1}, {"key": "status", "value": "stopped"}] + ), + "", + 0, + ) + assert cli.status(CT) == "stopped" + assert fake.argv == [ + "cv4pve-cli", + "api", + "get", + "/nodes/mainframe/lxc/100/status/current", + "-o", + "json", + ] diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..194b9dc --- /dev/null +++ b/uv.lock @@ -0,0 +1,195 @@ +version = 1 +revision = 5 +requires-python = ">=3.12" + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "cv4pve-tui" +version = "0.1.0" +source = { editable = "." } +dependencies = [ + { name = "textual" }, +] + +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-asyncio" }, +] + +[package.metadata] +requires-dist = [{ name = "textual", specifier = ">=8.2" }] + +[package.metadata.requires-dev] +dev = [ + { name = "pytest", specifier = ">=8" }, + { name = "pytest-asyncio", specifier = ">=1.0" }, +] + +[[package]] +name = "iniconfig" +version = "2.3.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/01/e1/2069291243c926a2ff1cd706c7f3eeb9b62144bf60f77c9fb9ff2fb26bd3/iniconfig-2.3.1.tar.gz", hash = "sha256:67f4b9c50da0dedf52af349e7749a80a9057a5031199791b906c3bb3ae878960", size = 21209, upload-time = "2026-10-06T22:48:38.076Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/56/43/4ca9e49d27a1fcf6bece6f6aec0ea46bb9112489b93d4b688fb415457bdb/iniconfig-2.3.1-py3-none-any.whl", hash = "sha256:9121e2c1fdb355232495be3194c8dfe87ccc2d5dee45947b78e68f499790d7a7", size = 7552, upload-time = "2026-10-06T22:48:36.959Z" }, +] + +[[package]] +name = "linkify-it-py" +version = "2.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/45/98/7a1a5f31fd5c7ba93e963b168e244b8e3dd705b3d2a718e3c3307583bf57/linkify_it_py-2.2.0.tar.gz", hash = "sha256:907acd2d17ac1fbb9ddb62c8957ccbd6158cac602231a15c3b0cd1e215f03cee", size = 32939, upload-time = "2026-08-29T07:07:08.305Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/13/d4/1152d1c7ab42d8b908be64fd200ddc870dc9d4925e951198702084aa1a7d/linkify_it_py-2.2.0-py3-none-any.whl", hash = "sha256:3adc40eb5af300b2605fcfdb968c24e1d780a90f1f2221af7c15e5111e94d443", size = 21971, upload-time = "2026-08-29T07:07:07.164Z" }, +] + +[[package]] +name = "markdown-it-py" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mdurl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" }, +] + +[package.optional-dependencies] +linkify = [ + { name = "linkify-it-py" }, +] + +[[package]] +name = "mdit-py-plugins" +version = "0.6.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/59/fc/f8d0863f8862f25602c0404d75568e89fb6b4109804645e5cdfb1be5cf56/mdit_py_plugins-0.6.1.tar.gz", hash = "sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0", size = 56114, upload-time = "2026-05-13T09:03:38.91Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a5/69/6da5581c6a7fede7dc261bf4e67d6adca4196f176b43288b55b3db395b6e/mdit_py_plugins-0.6.1-py3-none-any.whl", hash = "sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d", size = 66663, upload-time = "2026-05-13T09:03:37.76Z" }, +] + +[[package]] +name = "mdurl" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, +] + +[[package]] +name = "packaging" +version = "26.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412, upload-time = "2026-08-04T18:15:28.737Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" }, +] + +[[package]] +name = "platformdirs" +version = "4.14.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9a/6e/886dc7c9d21e2d95aceb696f1e31271742033aeef3661f6e9c8921493592/platformdirs-4.14.3.tar.gz", hash = "sha256:021add1ab113483edbcd4694f5fbb2d688b893ea30d35e1155bb371cf4be0be8", size = 71124, upload-time = "2026-11-01T16:27:32.928Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/93/41/f0aa003909ee77a9e7e55f8659abb635ca23ab0d022a449bbb18da87a1ef/platformdirs-4.14.3-py3-none-any.whl", hash = "sha256:cba2d05fe5e18b26b1fa6f376b8e6234c012abe1f8696c486fface7d0054e931", size = 35276, upload-time = "2026-11-01T16:27:31.498Z" }, +] + +[[package]] +name = "pluggy" +version = "1.7.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bf/db/7fc19e6f2dc92a966727031389fc2e08b558f0f25eb7403c1119ad4713cd/pluggy-1.7.0.tar.gz", hash = "sha256:d1eaa46ebb595891b860ab086b4d09c8588af65ebd4361b8e8f4bb8920b90ba8", size = 123304, upload-time = "2026-10-15T09:50:58.343Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/40/9e/2b38731e0fc536806f16490e1a12d7f0dc2a1235aa8cc07bcc75416a7daa/pluggy-1.7.0-py3-none-any.whl", hash = "sha256:7dd7b0d8832ba3cb632c306926ded123429211b83641b35dc5c41ad2d34f9bec", size = 27082, upload-time = "2026-10-15T09:50:56.808Z" }, +] + +[[package]] +name = "pygments" +version = "2.21.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" }, +] + +[[package]] +name = "pytest" +version = "9.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/79/42/bb4641652e58dedeac2ec51abc1f8cae4fee4ec63899a0999016229a0f2a/pytest-9.2.0.tar.gz", hash = "sha256:97c4f3b1f3c7414aa36465e09b8220f1d03627993bd51b1282e193753c80557b", size = 1797555, upload-time = "2026-10-27T21:30:20.728Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/51/39/1f5d8c8877a0fa327db959c19341d4be61660de1ab6b7391f9eb805bcd67/pytest-9.2.0-py3-none-any.whl", hash = "sha256:07acdcf88ec174f907dc01af70bf52d49b014bbab03faeeade31c96b18c2e588", size = 411776, upload-time = "2026-10-27T21:30:19.131Z" }, +] + +[[package]] +name = "pytest-asyncio" +version = "1.4.1.post1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/73/7f/c60a4338420b33453d4f814ac0c930abd399f162b9bdd21af4b9e363037f/pytest_asyncio-1.4.1.post1.tar.gz", hash = "sha256:f582d9ea4a1c082195abe8bd29a1fe106aeb1cd435e3dad2d3bc2749935c36a2", size = 145360, upload-time = "2026-11-01T18:26:03.732Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e3/de/adbc7578959cfab07e5e7c05cf6d54224a51e7667ea2c07458b6bc571e75/pytest_asyncio-1.4.1.post1-py3-none-any.whl", hash = "sha256:0755e488cc0a2b9bd2eab7292cc6390d4cbb8390f1020745647bae5a708e0165", size = 17140, upload-time = "2026-11-01T18:26:02.168Z" }, +] + +[[package]] +name = "rich" +version = "15.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, +] + +[[package]] +name = "textual" +version = "8.2.8" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py", extra = ["linkify"] }, + { name = "mdit-py-plugins" }, + { name = "platformdirs" }, + { name = "pygments" }, + { name = "rich" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/00/21/39a76b01bd5eea82a04baaca7580e105d8c59450df03998345bb2cfb307b/textual-8.2.8.tar.gz", hash = "sha256:3f106a9fbc73e39dd266c9712432087de78a6d644084c7c241d6a25c3169115b", size = 1860502, upload-time = "2026-06-30T06:51:24.495Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fb/be/35261223d9416a0751cdff1c7b4a6f881387218a12d439fe22fefebc8c04/textual-8.2.8-py3-none-any.whl", hash = "sha256:267375fd402dc8d981457212efa71f0e3365fd17bba144ba9bb3ed7563cb374a", size = 731418, upload-time = "2026-06-30T06:51:26.364Z" }, +] + +[[package]] +name = "typing-extensions" +version = "4.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f6/cc/6253133b5bb138fc3306cebfbda2c520f545d36b5be2c7255cc528bb45d6/typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5", size = 113555, upload-time = "2026-07-02T08:40:05.92Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/49/d3/b8441a820a491ddfc024b0b0cf0393375b75ea13866d9c66727e54c2fc80/typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", size = 45571, upload-time = "2026-07-02T08:40:04.659Z" }, +] -- 2.39.5 From 62b79ab29efb3dab5287ab3d7fc5bf532bee9a49 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Mon, 2 Nov 2026 20:22:53 +0100 Subject: [PATCH 2/2] Add context listing and switching to the TUI Press c to list cv4pve-cli contexts and switch between them. The active context is always shown in the header. - Switching runs `cv4pve-cli config use`, which changes the global current context. cv4pve-cli 2.4.0 has no per-call context option. - `config list` has no JSON output, so its text lines are parsed. - The guest table is cleared as soon as another context is chosen, so the old cluster's guests cannot be acted on. - A generation counter drops guest listings that started before a switch, and g and c are refused while a switch runs. Without this, a reload already in flight could refill the table with the old cluster's guests, and VMIDs collide across clusters. 103 tests pass. The switch was tested live from mainframe to mainframe-edge and back. Co-Authored-By: Claude Opus 5.5 --- README.md | 4 + docs/plans/guest-tui-mvp.md | 63 ++++++++ src/cv4pve_tui/app.py | 130 +++++++++++++++- src/cv4pve_tui/cli.py | 18 +++ tests/test_app.py | 288 +++++++++++++++++++++++++++++++++++- tests/test_cli.py | 49 ++++++ 6 files changed, 543 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 1a47505..7d83837 100644 --- a/README.md +++ b/README.md @@ -21,8 +21,12 @@ uv run cv4pve-tui | `r` | reboot (confirm) | | `e` | edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) | | `g` | refresh | +| `c` | switch cv4pve-cli context (cluster) | | `q` | quit | +Switching context runs `cv4pve-cli config use`, which changes cv4pve-cli's global current +context: other shells and scripts follow, and it stays after the app exits. + Empty fields in the edit dialog are left unchanged. If a CPU/memory change cannot be hot-plugged, the app tells you it applies at the next reboot. diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index 2411be2..3f73265 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -116,3 +116,66 @@ sockets/vCPU/hotplug settings, live auto-refresh. ## Open questions None blocking. + +--- + +# Context switching + +## Problem + +Several cv4pve-cli contexts (clusters) exist; switch between them inside the TUI +instead of quitting and running `cv4pve-cli config use`. + +## Decision + +Option A: switching runs `cv4pve-cli config use `, which changes the **global** +current context (all shells and scripts follow) and stays after the app exits. cv4pve-cli +2.4.0 has no per-call context flag or env var (verified), so there is no app-local option. + +Observed CLI behaviour: + +- `config list` has no `-o json`. Text output, one line per context; current one + prefixed with `* `, others with two spaces: + `* mainframe mainframe.iot.home.arpa:8006 (api-token)` + ` mainframe-edge mainframe-edge.home.arpa:8006 (api-token)` +- `config use ` -> exit 0 on success; unknown name -> exit 3, + `Error: Context '' not found.`, current context unchanged. + +### `cli.py` + +- `contexts() -> list[dict]` - `run("config", "list")`, each line parsed to + `{"name", "host", "auth", "current": bool}` (regex + `^(\*)?\s*(\S+)\s+(\S+)\s+\((.*)\)\s*$`); non-matching lines are ignored. +- `use_context(name)` - `run("config", "use", name)`. + +### `app.py` (UI contract) + +- The active context is always visible: `app.sub_title == f"context: {name}"` + (shown in the Header). Set on mount from `cli.contexts()` and after every switch. + If `contexts()` fails: error notify, sub_title `context: ?`. +- Key `c` opens `ContextScreen(ModalScreen[str | None])` with `OptionList#contexts`, + one option per context, option id = context name, current one marked with `*`. + Selecting an option dismisses with its name; `escape` dismisses with `None`. +- `c` while any guest is busy -> warning notify, no screen (VMIDs collide across + clusters, so the busy guard would point at the wrong cluster). +- Selecting the current context or dismissing -> no CLI call. +- A reload started before a switch must not refill the table (generation counter); + `g` and `c` are refused while a switch runs; the post-action reload finishes before + the guest stops being busy. +- Selecting another context: table cleared immediately (old cluster's guests must not + stay actionable), then in a worker `cli.use_context(name)`, sub_title updated, guests + reloaded. On `CliError`: error notify with summary, sub_title re-read from + `cli.contexts()`, guests reloaded. + +## Out of scope + +Adding/editing/deleting contexts, `config verify`, restoring the original context on exit. + +## Steps + +- [x] Tests for `contexts`/`use_context` and the UI contract above - red first. +- [x] Implementation - full suite green, ruff clean. +- [x] Live: open the list, switch to `mainframe-edge` and back to `mainframe` + (only `config use` and read-only listing; no guest actions); final context + is `mainframe`. +- [ ] Push `feat/guest-tui-mvp`, merge to `main` (user-approved). diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index e2c3df7..7d6bb6b 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -4,7 +4,17 @@ from textual import work from textual.app import App, ComposeResult from textual.containers import Horizontal, Vertical from textual.screen import ModalScreen -from textual.widgets import Button, DataTable, Footer, Header, Input, Select, Static +from textual.widgets import ( + Button, + DataTable, + Footer, + Header, + Input, + OptionList, + Select, + Static, +) +from textual.widgets.option_list import Option from cv4pve_tui import cli @@ -89,6 +99,32 @@ class EditScreen(ModalScreen): return int(text) +class ContextScreen(ModalScreen[str | None]): + """Dismisses with the chosen context name, or None when cancelled.""" + + BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom + + def __init__(self, contexts: list[dict]) -> None: + super().__init__() + self.contexts = contexts + + def compose(self) -> ComposeResult: + with Vertical(): + yield OptionList( + *( + Option( + f"{'*' if c['current'] else ' '} {c['name']} {c['host']}", + id=c["name"], + ) + for c in self.contexts + ), + id="contexts", + ) + + def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None: + self.dismiss(event.option.id) + + class GuestApp(App): BINDINGS = [ # noqa: RUF012 - Textual idiom ("s", "act('start')", "Start"), @@ -97,6 +133,7 @@ class GuestApp(App): ("r", "act('reboot')", "Reboot"), ("e", "edit", "Edit"), ("g", "reload", "Refresh"), + ("c", "context", "Context"), ("q", "quit", "Quit"), ] @@ -104,6 +141,9 @@ class GuestApp(App): super().__init__() self.guests: dict[str, dict] = {} self.busy: set[int] = set() + # bumped on every context switch; replies started earlier come from the old cluster + self.generation = 0 + self.switching = False def compose(self) -> ComposeResult: yield Header() @@ -122,8 +162,29 @@ class GuestApp(App): "Memory (MiB)", "Disk (GiB)", ) + self._show_context() self.action_reload() + @work(thread=True) + def _show_context(self) -> None: + self._read_context() + + def _read_context(self) -> None: + """Runs in a worker thread: show the CLI's current context in the Header.""" + gen = self.generation + try: + name = next(c["name"] for c in cli.contexts() if c["current"]) + except StopIteration: + name = "?" + except Exception as exc: # noqa: BLE001 - any failure must reach the user, not kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + name = "?" + self.call_from_thread(self._set_context_title, name, gen) + + def _set_context_title(self, name: str, gen: int) -> None: + if gen == self.generation: + self.sub_title = f"context: {name}" + def _selected(self) -> dict | None: table = self.query_one("#guests", DataTable) if not table.row_count: @@ -132,7 +193,9 @@ class GuestApp(App): table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value ] - def _fill(self, guests: list[dict]) -> None: + def _fill(self, guests: list[dict], gen: int) -> None: + if gen != self.generation: # listed before a context switch: old cluster + return table = self.query_one("#guests", DataTable) current = self._selected() table.clear() @@ -148,6 +211,7 @@ class GuestApp(App): def _reload(self, acted: dict | None = None) -> None: """Runs in a worker thread. `acted` gets its live status: the listing lags.""" + gen = self.generation try: guests = cli.list_guests() for g in guests: @@ -156,12 +220,66 @@ class GuestApp(App): except Exception as exc: # noqa: BLE001 - any failure must reach the user, not kill the app self.call_from_thread(self.notify, _message(exc), severity="error") return - self.call_from_thread(self._fill, guests) + self.call_from_thread(self._fill, guests, gen) + + def action_reload(self) -> None: + if self.switching: + self.notify("Switching context; refresh after it ends", severity="warning") + return + self._reload_worker() @work(thread=True) - def action_reload(self) -> None: + def _reload_worker(self) -> None: self._reload() + def action_context(self) -> None: + if self.busy or self.switching: + self.notify( + "Busy (guest action or context switch); try again when it ends", + severity="warning", + ) + return + self._open_context() + + @work(thread=True) + def _open_context(self) -> None: + try: + contexts = cli.contexts() + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + return + current = next((c["name"] for c in contexts if c["current"]), None) + self.call_from_thread( + self.push_screen, + ContextScreen(contexts), + lambda name: self._context_chosen(name, current), + ) + + def _context_chosen(self, name: str | None, current: str | None) -> None: + if name is None or name == current: + return + self.query_one( + "#guests", DataTable + ).clear() # old cluster's guests: not actionable + self.guests = {} + self.generation += 1 + self.switching = True + self._switch_context(name) + + @work(thread=True) + def _switch_context(self, name: str) -> None: + try: + try: + cli.use_context(name) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + self._read_context() + else: + self.call_from_thread(setattr, self, "sub_title", f"context: {name}") + self._reload() + finally: + self.call_from_thread(setattr, self, "switching", False) + def _start_job(self, guest: dict, label: str, job) -> None: if guest["vmid"] in self.busy: self.notify( @@ -185,8 +303,10 @@ class GuestApp(App): self.notify, f"{label} done: {guest['name']} ({guest['vmid']})" ) finally: + self._reload( + acted=guest + ) # before un-busying, so no context switch can start mid-reload self.busy.discard(guest["vmid"]) - self._reload(acted=guest) def action_act(self, name: str) -> None: guest = self._selected() diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index d665612..a90aecf 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -138,3 +138,21 @@ def status(guest: dict) -> str: return { i["key"]: i["value"] for i in json.loads(run("api", "get", path, "-o", "json")) }["status"] + + +_CONTEXT_LINE = re.compile(r"^(\*)?\s*(\S+)\s+(\S+)\s+\((.*)\)\s*$") + + +def contexts() -> list[dict]: + """Parse `config list` (text only); lines that do not look like a context are skipped.""" + found = (_CONTEXT_LINE.match(line) for line in run("config", "list").splitlines()) + return [ + {"name": m[2], "host": m[3], "auth": m[4], "current": bool(m[1])} + for m in found + if m + ] + + +def use_context(name: str) -> None: + """Changes cv4pve-cli's GLOBAL current context.""" + run("config", "use", name) diff --git a/tests/test_app.py b/tests/test_app.py index 3ddd6a1..dde40c9 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -8,10 +8,10 @@ import threading import pytest from textual.screen import ModalScreen -from textual.widgets import Button, DataTable, Input, Select, Static +from textual.widgets import Button, DataTable, Input, OptionList, Select, Static from cv4pve_tui import cli -from cv4pve_tui.app import ConfirmScreen, EditScreen, GuestApp +from cv4pve_tui.app import ConfirmScreen, ContextScreen, EditScreen, GuestApp VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu", "status": "running", "maxcpu": 4, "maxmem": 8192 * 2**20, "maxdisk": 10 * 2**30} # fmt: skip @@ -31,10 +31,24 @@ class Fake: self.action_error = None self.action_hook = None self.live_status, self.status_calls = {}, [] + self.current = "mainframe" + self.names = ["mainframe", "mainframe-edge"] + self.contexts_error = None + self.contexts_calls = 0 + self.use_calls = [] + self.use_error = None + self.use_hook = None + self.guests_by_context = {} + self.list_hook = None def list_guests(self): self.list_calls += 1 - return list(self.guests) + snapshot = list(self.guests_by_context.get(self.current, self.guests)) + if ( + self.list_hook + ): # runs after reading: simulates a slow reply from that cluster + self.list_hook() + return snapshot def guest_config(self, guest): return {"cores": guest["maxcpu"]} @@ -64,12 +78,30 @@ class Fake: self.status_calls.append(guest["vmid"]) return self.live_status.get(guest["vmid"], guest["status"]) + def contexts(self): + self.contexts_calls += 1 + if self.contexts_error: + raise self.contexts_error + return [ + {"name": n, "host": f"{n}.home.arpa:8006", "auth": "api-token", + "current": n == self.current} + for n in self.names + ] # fmt: skip + + def use_context(self, name): + self.use_calls.append(name) + if self.use_hook: + self.use_hook() + if self.use_error: + raise self.use_error + self.current = name + @pytest.fixture def fake(monkeypatch): f = Fake() for name in ("list_guests", "guest_config", "disks", "action", "set_resources", - "resize_disk", "pending_keys", "status"): # fmt: skip + "resize_disk", "pending_keys", "status", "contexts", "use_context"): # fmt: skip monkeypatch.setattr(cli, name, getattr(f, name)) return f @@ -447,3 +479,251 @@ async def test_table_shows_live_status_after_action_not_lagging_list(fake): assert "stopped" in [ str(c) for c in app.query_one("#guests", DataTable).get_row("108") ] + + +# --- context switching ------------------------------------------------------ +# spec: docs/plans/guest-tui-mvp.md ("Context switching" > "app.py (UI contract)") + + +async def open_contexts(app, pilot): + await settle(app, pilot) + await pilot.press("c") + await settle(app, pilot) + assert isinstance(app.screen, ContextScreen) + return app.screen.query_one("#contexts", OptionList) + + +async def choose(app, pilot, index): + ol = app.screen.query_one("#contexts", OptionList) + ol.focus() + ol.highlighted = index + await pilot.pause() + await pilot.press("enter") + await settle(app, pilot) + + +async def test_sub_title_shows_active_context_after_mount(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + assert app.sub_title == "context: mainframe" + + +async def test_contexts_failure_notifies_and_sub_title_is_unknown(fake): + fake.contexts_error = cli.CliError("Error: no config") + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + assert errors(app) + assert app.sub_title == "context: ?" + + +async def test_c_opens_context_screen_listing_contexts_with_current_marked(fake): + app = make_app() + async with app.run_test() as pilot: + ol = await open_contexts(app, pilot) + assert ol.option_count == 2 + options = [ol.get_option_at_index(i) for i in range(2)] + assert [o.id for o in options] == ["mainframe", "mainframe-edge"] + assert "*" in str(options[0].prompt) + assert "*" not in str(options[1].prompt) + assert "mainframe-edge" in str(options[1].prompt) + + +async def test_selecting_other_context_switches_and_reloads(fake): + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + before = fake.list_calls + await choose(app, pilot, 1) + assert fake.use_calls == ["mainframe-edge"] + assert app.sub_title == "context: mainframe-edge" + assert fake.list_calls > before + assert not isinstance(app.screen, ModalScreen) + assert app.query_one("#guests", DataTable).row_count == 2 + + +async def test_table_is_cleared_while_switch_is_in_progress(fake): + started, release = threading.Event(), threading.Event() + fake.use_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + assert app.query_one("#guests", DataTable).row_count == 2 + try: + ol = app.screen.query_one("#contexts", OptionList) + ol.focus() + ol.highlighted = 1 + await pilot.pause() + await pilot.press("enter") + for _ in range(200): + if started.is_set(): + break + await pilot.pause(0.01) + assert started.is_set() + await pilot.pause() + assert app.query_one("#guests", DataTable).row_count == 0 + finally: + release.set() + await settle(app, pilot) + assert app.query_one("#guests", DataTable).row_count == 2 + + +async def test_selecting_current_context_makes_no_cli_call(fake): + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + before = fake.list_calls + await choose(app, pilot, 0) + assert fake.use_calls == [] + assert not isinstance(app.screen, ModalScreen) + assert app.sub_title == "context: mainframe" + assert fake.list_calls == before + + +async def test_escape_closes_context_screen_without_cli_call(fake): + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + await pilot.press("escape") + await settle(app, pilot) + assert not isinstance(app.screen, ModalScreen) + assert fake.use_calls == [] + assert app.sub_title == "context: mainframe" + + +async def test_c_while_guest_busy_warns_and_opens_nothing(fake): + started, release = threading.Event(), threading.Event() + fake.action_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + try: + await pilot.press("s") + for _ in range(200): + if started.is_set(): + break + await pilot.pause(0.01) + assert started.is_set() + await pilot.press("c") + await pilot.pause() + assert not isinstance(app.screen, ContextScreen) + assert any(sev == "warning" for _, sev in app.seen) + finally: + release.set() + await settle(app, pilot) + assert fake.use_calls == [] + + +async def test_use_context_failure_notifies_rereads_context_and_reloads(fake): + err = cli.CliError("Error: Context 'mainframe-edge' not found.") + + def hook(): # the CLI state differs from what the app assumed + fake.current = "mainframe-edge" + + fake.use_hook, fake.use_error = hook, err + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + contexts_before, list_before = fake.contexts_calls, fake.list_calls + await choose(app, pilot, 1) + assert any(err.summary in m for m in errors(app)) + assert fake.contexts_calls > contexts_before + assert app.sub_title == "context: mainframe-edge" # whatever contexts() says + assert fake.list_calls > list_before + assert app.query_one("#guests", DataTable).row_count == 2 + + +# --- review finding: no old-cluster rows after a switch ---------------------- + +EDGE = {"vmid": 300, "name": "edge-vm", "node": "edge", "type": "qemu", "status": "running", + "maxcpu": 2, "maxmem": 2**30, "maxdisk": 2**31} # fmt: skip + + +async def until(pilot, cond): + for _ in range(300): + if cond(): + return + await pilot.pause(0.01) + raise AssertionError("condition not reached") + + +def blocking_once(): + started, release = threading.Event(), threading.Event() + + def hook(): + if not started.is_set(): + started.set() + release.wait(10) + + return hook, started, release + + +def rows(app): + return [k.value for k in app.query_one("#guests", DataTable).rows] + + +async def switch_to_edge(app, pilot): + await pilot.press("c") + await until(pilot, lambda: isinstance(app.screen, ContextScreen)) + ol = app.screen.query_one("#contexts", OptionList) + ol.focus() + ol.highlighted = 1 + await pilot.pause() + await pilot.press("enter") + + +async def test_reload_in_flight_before_switch_cannot_restore_old_rows(fake): + fake.guests_by_context = {"mainframe-edge": [EDGE]} + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + hook, started, release = blocking_once() + fake.list_hook = hook + try: + await pilot.press("g") # reads mainframe, then stalls + await until(pilot, started.is_set) + await switch_to_edge(app, pilot) + await until(pilot, lambda: rows(app) == ["300"]) + finally: + release.set() + await settle(app, pilot) + assert rows(app) == ["300"] + assert app.sub_title == "context: mainframe-edge" + + +async def test_refresh_refused_while_switching(fake): + started, release = threading.Event(), threading.Event() + fake.use_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + try: + await switch_to_edge(app, pilot) + await until(pilot, started.is_set) + before = fake.list_calls + await pilot.press("g") + await pilot.pause() + assert fake.list_calls == before + assert any(sev == "warning" for _, sev in app.seen) + finally: + release.set() + await settle(app, pilot) + + +async def test_context_refused_while_post_action_reload_runs(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + hook, started, release = blocking_once() + fake.list_hook = hook + try: + await pilot.press("s") # start is instant; the reload after it stalls + await until(pilot, started.is_set) + await pilot.press("c") + await pilot.pause(0.1) + assert not isinstance(app.screen, ContextScreen) + assert any(sev == "warning" for _, sev in app.seen) + finally: + release.set() + await settle(app, pilot) diff --git a/tests/test_cli.py b/tests/test_cli.py index db4e3c4..98b1ac8 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -370,3 +370,52 @@ def test_status_reads_live_status_current(fake): "-o", "json", ] + + +# --- contexts / use_context (spec: "Context switching" > cli.py) ------------- + +CONTEXT_LIST = ( + "* mainframe mainframe.iot.home.arpa:8006 (api-token)\n" + " mainframe-edge mainframe-edge.home.arpa:8006 (api-token)\n" +) + + +def test_contexts_parses_current_name_host_auth(fake): + fake.result = (CONTEXT_LIST, "", 0) + assert cli.contexts() == [ + {"name": "mainframe", "host": "mainframe.iot.home.arpa:8006", + "auth": "api-token", "current": True}, + {"name": "mainframe-edge", "host": "mainframe-edge.home.arpa:8006", + "auth": "api-token", "current": False}, + ] # fmt: skip + + +def test_contexts_runs_config_list(fake): + fake.result = (CONTEXT_LIST, "", 0) + cli.contexts() + assert fake.argv == ["cv4pve-cli", "config", "list"] + + +def test_contexts_ignores_junk_and_empty_lines(fake): + fake.result = ( + "\nContexts:\n" + CONTEXT_LIST + "\n \nnot a context line\n", + "", + 0, + ) + assert [c["name"] for c in cli.contexts()] == ["mainframe", "mainframe-edge"] + + +def test_contexts_empty_output_gives_empty_list(fake): + assert cli.contexts() == [] + + +def test_use_context_runs_config_use_with_name(fake): + cli.use_context("mainframe-edge") + assert fake.argv == ["cv4pve-cli", "config", "use", "mainframe-edge"] + + +def test_use_context_unknown_name_raises_clierror(monkeypatch): + FakeRun(monkeypatch, "Error: Context 'x' not found.\n", returncode=3) + with pytest.raises(cli.CliError) as exc: + cli.use_context("x") + assert "Context 'x' not found" in str(exc.value) -- 2.39.5