From 32809b6743cfd7292b25a13ffa87011be2323995 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Mon, 2 Nov 2026 21:06:22 +0100 Subject: [PATCH] Add guest info screen with backups, backup now and network Press i on a guest to see its backups and network. Backups from every backup-capable storage are listed newest first, with the backup jobs that cover the guest. The network part shows the netN settings (bridge, VLAN, MAC, IPv4/IPv6 setting, firewall) and the addresses in use. Press b in the info screen to back the guest up now: snapshot mode and zstd, like the scheduled job, to a storage you pick (the job's storage is preselected). It asks for confirmation, marks the guest busy and uses --wait. Fixed after review and a read-only live check: - prune-backups arrives as a JSON string; parsing it as a dict crashed the app. - CT addresses come from ip-addresses. inet6 holds only the link-local address, so the global SLAAC address was missing. - The info screen uses the guest's live status, and odd API data (null size, unknown address types) no longer kills the app. - The test fake now copies guest dicts; a shared constant was being mutated across tests. Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + docs/plans/guest-tui-mvp.md | 105 +++++++++++ src/cv4pve_tui/app.py | 217 ++++++++++++++++++++++ src/cv4pve_tui/cli.py | 118 ++++++++++++ tests/test_app.py | 354 +++++++++++++++++++++++++++++++++++- tests/test_cli.py | 264 +++++++++++++++++++++++++++ 6 files changed, 1057 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 510140f..e3a2325 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ uv run cv4pve-tui | `x` | stop - hard power-off (confirm) | | `r` | reboot (confirm) | | `e` | edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) | +| `i` | info: backups, backup job, network (config and runtime IPs); `b` there = backup now (snapshot, zstd, storage choice) | | `t` | shell in a running LXC container (`ssh -t root@ pct enter `) | | `g` | refresh | | `c` | switch cv4pve-cli context (cluster) | diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index b2cfad8..7e43e40 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -236,3 +236,108 @@ VM consoles (serial/VNC), configurable SSH user/host/options, termproxy/websocke - [x] Interactive test by the user: `t` on CT 104, `exit`, back in the table; screen clean after the reset. - [ ] Push `feat/lxc-shell`, merge to `main` (user-approved). + +--- + +# Guest info screen: backups, backup now, network (read-only) - delivery A + +## Problem + +Per guest, see its latest backups and network setup, and start a backup now. Network +*editing* is delivery B (separate, live-tested), not part of this. + +## Decisions (user-approved) + +- Backup now uses the scheduled job's settings: `--mode snapshot --compress zstd`. + Storage is chosen from the node's backup-capable storages, preselected to the job's + storage (`truenas-backup` here), `local` as alternative. +- Manual backups get `--notes-template "{{guestname}} (manual)"` so they are + recognisable. Note for the user: `truenas-backup` has storage prune `keep-all=1`, but + the job has `keep-last 3` per guest; a manual backup there likely counts towards it at + the job's next run (not verified). + +Observed CLI behaviour (cv4pve-cli 2.4.0, all read-only probes): + +- `get node storages -o json` -> list of dicts: `storage`, `content` + (comma list, e.g. `iso,snippets,import,backup,vztmpl`), `enabled`, `active`, `type`, + `avail` (bytes). +- `get cluster backup-jobs -o json` -> list of dicts: `id`, `enabled`, `storage`, + `schedule` (e.g. `mon,wed,fri 04:00`), `mode`, `compress`, `all` (1 = all guests), + `vmid` (comma list or null), `exclude` (comma list or null), `node` (null = all), + `next-run` (epoch), `prune-backups` (e.g. `{"keep-last":"3"}`). +- `get guest backups -o json` -> list of dicts: `volid` + (`truenas-backup:backup/vzdump-lxc-104-2026_11_02-04_11_47.tar.zst`), `ctime` (epoch), + `size` (bytes), `format`, `notes`, `protected`, `subtype` (`lxc`|`qemu`), `vmid`. +- `do backup guest [--key value]` - no `--yes` option. + vzdump params: `mode` snapshot|suspend|stop, `compress` 0|1|gzip|lzo|zstd, + `notes-template`. +- CT net config: `netN` = `name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=auto,type=veth` + (optional `gw`, `gw6`, `tag`, `firewall`, `rate`, `mtu`, ...). +- VM net config: `netN` = `virtio=02:24:A6:49:A8:25,bridge=vmbr0[,tag=..][,firewall=1]` + (first item is `=`); IP settings live in `ipconfigN` (cloud-init), e.g. + `ip=dhcp`. +- Runtime IPs, CT (running): `get ct interfaces -o json` -> list of + dicts: `name`, `hwaddr`, `inet` (e.g. `172.16.41.200/23`), `inet6` (may be absent/ + null), `ip-addresses` (JSON string). +- Runtime IPs, VM (running, guest agent): `get vm agent network -o json` + -> `[{"key": "result", "value": ""}]`; the JSON is a list of + `{"name", "hardware-address", "ip-addresses": [{"ip-address", "prefix", + "ip-address-type": "ipv4"|"ipv6"}]}`. Fails when the agent is not running. + +### `cli.py` + +- `backup_storages(guest) -> list[str]` - storages on the guest's node whose `content` + contains `backup` and that are enabled and active, in API order. +- `backup_jobs() -> list[dict]`. +- `job_storage(jobs, guest) -> str | None` - storage of the first enabled job covering + the guest: (`all` truthy and vmid not in `exclude`) or vmid in `vmid`; and `node` null + or equal to the guest's node. +- `backup_jobs()` parses `prune-backups` when the CLI returns it as a JSON string + (real output: `'{"keep-last":"3"}'`). +- `backups(guest, storages) -> list[dict]` - all backups of the guest on those + storages, newest first (`ctime` desc). Each dict gets `storage` (from volid prefix). +- `backup_now(guest, storage)` -> `do backup guest --mode + snapshot --compress zstd --notes-template "{{guestname}} (manual)" --wait`. +- `parse_net(value: str) -> dict` - `netN` string to an ordered dict of key -> value + (insertion order kept, so B can re-serialise unchanged keys). VM: the first + `=` item is kept as is (key = model). +- `nets(config) -> dict[str, dict]` - `netN` keys (sorted by N) -> `parse_net(value)`. +- `interfaces(guest) -> list[dict]` - runtime interfaces as + `{"name", "mac", "ipv4": [cidr...], "ipv6": [cidr...]}`, loopback (`lo`) excluded. + CT via `get ct interfaces` (addresses from its `ip-addresses` JSON string - `inet6` + alone holds only one, e.g. link-local, and misses the global SLAAC address), VM via + `get vm agent network`. Only called when the guest's live status is running. + +### `app.py` (UI contract) + +- Key `i` on a guest opens `InfoScreen(guest)` (a `Screen` or `ModalScreen`). Empty + table: nothing. +- InfoScreen loads in a thread worker: config -> `cli.nets`, `cli.interfaces` (only if + status is `running`; a failure there is shown inline as "unavailable: ", not + as a fatal error), `cli.backup_storages`, `cli.backup_jobs`, `cli.backups`. +- Shows: + - `DataTable#backups`: Date (local time, `YYYY-MM-DD HH:MM`), Storage, Size (GiB, one + decimal), Format, Notes. Newest first. A `Static#jobs` line naming the job(s) that + cover the guest (schedule, storage, `keep-last` if present) or "No backup job + covers this guest". + - `DataTable#nets`: Iface (net0...), Bridge, VLAN (`tag` or `-`), MAC, IPv4 (CT `ip`, + VM `ipconfigN` `ip`, or `-`), IPv6 (CT `ip6`, VM `ipconfigN` `ip6`, or `-`), + Firewall (`yes`/`no`). + - `DataTable#addresses`: Interface, MAC, IPv4, IPv6 (runtime), or the unavailable + line. +- Key `b` in InfoScreen: `BackupScreen` modal with `Select#storage` (options = + backup storages, preselected `job_storage` or the first), text saying mode snapshot, + zstd, and the guest; buttons `#yes` / `#no`; escape = no. On yes: guest marked busy + in `app.busy` (refuse with a warning if already busy), "backup started" notify, worker + runs `cli.backup_now`, then "backup done"/error notify, busy released, backup list + reloaded if the InfoScreen is still open. +- `escape` closes InfoScreen; `g` in InfoScreen reloads it. +- Every CLI failure -> error notify with `_message(exc)`, app keeps running. + +## Steps + +- [x] Tests (cli + UI contract) - red first. +- [x] Implementation - suite green, ruff clean, README keys `i` / `b`. +- [x] Read-only live check: open InfoScreen for CT 104 and VM 108, data shown. +- [ ] User tests "Backup now" on CT 104. +- [ ] Push `feat/guest-info`, merge to `main` (user-approved after the test). diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index bbb8664..a755502 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -2,6 +2,7 @@ import subprocess import sys +from datetime import datetime from textual import work from textual.app import App, ComposeResult, SuspendNotSupported @@ -128,6 +129,185 @@ class ContextScreen(ModalScreen[str | None]): self.dismiss(event.option.id) +class BackupScreen(ModalScreen[str | None]): + """Dismisses with the chosen storage, or None when cancelled.""" + + BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom + + def __init__(self, guest: dict, storages: list[str], chosen: str) -> None: + super().__init__() + self.guest, self.storages, self.chosen = guest, storages, chosen + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static( + f"Back up {self.guest['name']} ({self.guest['vmid']}) now: mode snapshot, zstd" + ) + yield Select( + [(s, s) for s in self.storages], + value=self.chosen, + allow_blank=False, + id="storage", + ) + with Horizontal(): + yield Button("Yes", id="yes", variant="primary") + yield Button("No", id="no") + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "yes": + self.dismiss(self.query_one("#storage", Select).value) + else: + self.dismiss(None) + + +class InfoScreen(ModalScreen): + """Backups, backup job and network of one guest (read-only), plus backup now.""" + + BINDINGS = [ # noqa: RUF012 - Textual idiom + ("escape", "dismiss", "Close"), + ("g", "reload", "Refresh"), + ("b", "backup", "Backup now"), + ] + DEFAULT_CSS = "InfoScreen DataTable { height: auto; max-height: 12; }" + + def __init__(self, guest: dict) -> None: + super().__init__() + self.guest = guest + self.storages: list[str] = [] + self.jobs: list[dict] = [] + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static(f"{self.guest['name']} ({self.guest['vmid']})", markup=False) + yield Static("", id="jobs", markup=False) + yield DataTable(id="backups") + yield DataTable(id="nets") + yield Static("", id="unavailable", markup=False) + yield DataTable(id="addresses") + yield Footer() + + def on_mount(self) -> None: + self.query_one("#backups", DataTable).add_columns( + "Date", "Storage", "Size (GiB)", "Format", "Notes" + ) + self.query_one("#nets", DataTable).add_columns( + "Iface", "Bridge", "VLAN", "MAC", "IPv4", "IPv6", "Firewall" + ) + self.query_one("#addresses", DataTable).add_columns( + "Interface", "MAC", "IPv4", "IPv6" + ) + self.action_reload() + + def action_reload(self) -> None: + self._load_worker() + + @work(thread=True) + def _load_worker(self) -> None: + self.load() + + def load(self) -> None: + """Runs in a worker thread.""" + guest = self.guest + try: + config = cli.guest_config(guest) + nets = cli.nets(config) + storages = cli.backup_storages(guest) + jobs = cli.backup_jobs() + backups = cli.backups(guest, storages) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.app.call_from_thread(self.app.notify, _message(exc), severity="error") + return + addresses, unavailable = [], "" + try: # live status: the guest table's status may be stale + running = cli.status(guest) == "running" + except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works + running, unavailable = False, f"unavailable: {_message(exc)}" + if running: + try: + addresses = cli.interfaces(guest) + except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works + unavailable = f"unavailable: {_message(exc)}" + self.app.call_from_thread( + self._show, config, nets, storages, jobs, backups, addresses, unavailable + ) + + def _show( + self, config, nets, storages, jobs, backups, addresses, unavailable + ) -> None: + if not self.is_attached: # closed while loading + return + try: + self._fill_screen( + config, nets, storages, jobs, backups, addresses, unavailable + ) + except Exception as exc: # noqa: BLE001 - odd API data must not kill the app + self.app.notify(f"info: {_message(exc)}", severity="error") + + def _fill_screen( + self, config, nets, storages, jobs, backups, addresses, unavailable + ) -> None: + self.storages, self.jobs = storages, jobs + covering = cli.covering_jobs(jobs, self.guest) + self.query_one("#jobs", Static).update( + "; ".join(_job_text(j) for j in covering) + or "No backup job covers this guest" + ) + table = self.query_one("#backups", DataTable).clear() + for b in backups: + table.add_row( + datetime.fromtimestamp(b["ctime"]).strftime("%Y-%m-%d %H:%M") + if b.get("ctime") + else "-", + b["storage"], + f"{b['size'] / 2**30:.1f}" if b.get("size") is not None else "-", + b.get("format") or "-", + b.get("notes") or "", + ) + table = self.query_one("#nets", DataTable).clear() + for name, net in nets.items(): + if cli.kind(self.guest) == "vm": # IP settings are cloud-init's ipconfigN + mac = next(iter(net.values()), "-") + ip = cli.parse_net(str(config.get(f"ipconfig{name[3:]}", ""))) + else: + mac, ip = net.get("hwaddr", "-"), net + table.add_row( + name, + net.get("bridge", "-"), + net.get("tag", "-"), + mac, + ip.get("ip", "-"), + ip.get("ip6", "-"), + "yes" if net.get("firewall") == "1" else "no", + ) + self.query_one("#unavailable", Static).update(unavailable) + table = self.query_one("#addresses", DataTable).clear() + for i in addresses: + table.add_row( + i["name"], i["mac"] or "-", ", ".join(i["ipv4"]), ", ".join(i["ipv6"]) + ) + + def action_backup(self) -> None: + if not self.storages: + self.app.notify("No backup storage available", severity="warning") + return + chosen = cli.job_storage(self.jobs, self.guest) + if chosen not in self.storages: + chosen = self.storages[0] + self.app.push_screen( + BackupScreen(self.guest, self.storages, chosen), + lambda storage: ( + self.app.start_backup(self.guest, storage) if storage else None + ), + ) + + +def _job_text(job: dict) -> str: + keep = (job.get("prune-backups") or {}).get("keep-last") + return f"{job['schedule']} to {job['storage']}" + ( + f", keep-last {keep}" if keep else "" + ) + + class GuestApp(App): BINDINGS = [ # noqa: RUF012 - Textual idiom ("s", "act('start')", "Start"), @@ -135,6 +315,7 @@ class GuestApp(App): ("x", "act('stop')", "Stop"), ("r", "act('reboot')", "Reboot"), ("e", "edit", "Edit"), + ("i", "info", "Info"), ("t", "shell", "Shell"), ("g", "reload", "Refresh"), ("c", "context", "Context"), @@ -340,6 +521,42 @@ class GuestApp(App): return self._open_edit(guest) + def action_info(self) -> None: + guest = self._selected() + if guest is not None: + self.push_screen(InfoScreen(guest)) + + def start_backup(self, guest: dict, storage: str) -> None: + who = f"{guest['name']} ({guest['vmid']})" + if guest["vmid"] in self.busy: + self.notify(f"{who} is busy", severity="warning") + return + self.busy.add(guest["vmid"]) + self.notify(f"Backup started: {who} to {storage}") + self._run_backup(guest, storage) + + @work(thread=True) + def _run_backup(self, guest: dict, storage: str) -> None: + who = f"{guest['name']} ({guest['vmid']})" + try: + cli.backup_now(guest, storage) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread( + self.notify, f"Backup failed: {_message(exc)}", severity="error" + ) + else: + self.call_from_thread(self.notify, f"Backup done: {who}") + finally: + for screen in list( + self.screen_stack + ): # reload while still busy, like _run_job + if ( + isinstance(screen, InfoScreen) + and screen.guest["vmid"] == guest["vmid"] + ): + screen.load() + self.busy.discard(guest["vmid"]) + def action_shell(self) -> None: guest = self._selected() if guest is None: diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index a90aecf..9b074a8 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -156,3 +156,121 @@ def contexts() -> list[dict]: def use_context(name: str) -> None: """Changes cv4pve-cli's GLOBAL current context.""" run("config", "use", name) + + +def backup_storages(guest: dict) -> list[str]: + """Enabled, active storages on the guest's node that accept backups, in API order.""" + items = json.loads(run("get", "node", "storages", guest["node"], "-o", "json")) + return [ + s["storage"] + for s in items + if "backup" in (s.get("content") or "").split(",") + and s.get("enabled") + and s.get("active") + ] + + +def backup_jobs() -> list[dict]: + jobs = json.loads(run("get", "cluster", "backup-jobs", "-o", "json")) + for job in jobs: # the CLI returns prune-backups as a JSON string + if isinstance(job.get("prune-backups"), str): + job["prune-backups"] = json.loads(job["prune-backups"]) + return jobs + + +def _ids(value) -> list[str]: + return str(value).split(",") if value else [] + + +def covering_jobs(jobs: list[dict], guest: dict) -> list[dict]: + """Enabled jobs that back up the guest (job order).""" + vmid = str(guest["vmid"]) + return [ + j + for j in jobs + if j.get("enabled") + and ( + j.get("all") + and vmid not in _ids(j.get("exclude")) + or vmid in _ids(j.get("vmid")) + ) + and j.get("node") in (None, guest["node"]) + ] + + +def job_storage(jobs: list[dict], guest: dict) -> str | None: + covering = covering_jobs(jobs, guest) + return covering[0]["storage"] if covering else None + + +def backups(guest: dict, storages: list[str]) -> list[dict]: + """The guest's backups on the given storages, newest first, each with its `storage`.""" + found = [] + for storage in storages: + out = run( + "get", "guest", "backups", guest["node"], storage, str(guest["vmid"]), + "-o", "json", + ) # fmt: skip + for item in json.loads(out): + item["storage"] = item["volid"].split(":", 1)[0] + found.append(item) + return sorted(found, key=lambda b: b["ctime"], reverse=True) + + +def backup_now(guest: dict, storage: str) -> None: + """Same settings as the scheduled job; no --yes option exists for this command.""" + run( + "do", "backup", "guest", guest["node"], str(guest["vmid"]), storage, + "--mode", "snapshot", "--compress", "zstd", + "--notes-template", "{{guestname}} (manual)", "--wait", + ) # fmt: skip + + +def parse_net(value: str) -> dict: + """`k=v,k=v` to an ordered dict; a VM's leading `=` is just the first pair.""" + return dict(item.partition("=")[::2] for item in value.split(",") if item) + + +def nets(config: dict) -> dict[str, dict]: + keys = sorted( + (k for k in config if re.fullmatch(r"net\d+", k)), key=lambda k: int(k[3:]) + ) + return {k: parse_net(str(config[k])) for k in keys} + + +def _addresses(listed: list[dict], types: dict[str, str]) -> dict[str, list[str]]: + """ip-addresses entries -> {"ipv4": [cidr], "ipv6": [cidr]}; unknown types ignored.""" + addrs = {"ipv4": [], "ipv6": []} + for a in listed: + family = types.get(a.get("ip-address-type")) + if family: + addrs[family].append(f"{a['ip-address']}/{a['prefix']}") + return addrs + + +def interfaces(guest: dict) -> list[dict]: + """Runtime interfaces of a running guest, without loopback.""" + args = (guest["node"], str(guest["vmid"]), "-o", "json") + if kind(guest) == "ct": + found = [] + for i in json.loads(run("get", "ct", "interfaces", *args)): + listed = i.get("ip-addresses") # all addresses; inet6 alone holds only one + if isinstance(listed, str): + listed = json.loads(listed) + if listed: + addrs = _addresses(listed, {"inet": "ipv4", "inet6": "ipv6"}) + else: + addrs = {"ipv4": [i["inet"]] if i.get("inet") else [], + "ipv6": [i["inet6"]] if i.get("inet6") else []} # fmt: skip + found.append({"name": i["name"], "mac": i.get("hwaddr"), **addrs}) + else: + items = json.loads(run("get", "vm", "agent", "network", *args)) + found = [] + for i in json.loads( + {x["key"]: x["value"] for x in items}.get("result") or "[]" + ): + addrs = _addresses( + i.get("ip-addresses") or [], {"ipv4": "ipv4", "ipv6": "ipv6"} + ) + found.append({"name": i["name"], "mac": i.get("hardware-address"), **addrs}) + return [i for i in found if i["name"] != "lo"] diff --git a/tests/test_app.py b/tests/test_app.py index 1811d43..b81d423 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -6,11 +6,13 @@ All cli functions are monkeypatched: cv4pve-cli is never executed. import contextlib import subprocess import threading +from datetime import datetime import pytest from textual.screen import ModalScreen from textual.widgets import Button, DataTable, Input, OptionList, Select, Static +from cv4pve_tui import app as appmod from cv4pve_tui import cli from cv4pve_tui.app import ConfirmScreen, ContextScreen, EditScreen, GuestApp @@ -41,10 +43,22 @@ class Fake: self.use_hook = None self.guests_by_context = {} self.list_hook = None + # delivery A (guest info screen) + self.net_configs = {} # vmid -> extra config keys (netN, ipconfigN) + self.storages = ["truenas-backup", "local"] + self.jobs = [BACKUP_JOB] + self.backup_list = list(BACKUPS) # newest first, as cli.backups returns them + self.iface_list, self.iface_error = [], None + self.iface_calls, self.backups_calls, self.backup_runs = [], [], [] + self.backup_error = None + self.backup_hook = None def list_guests(self): self.list_calls += 1 - snapshot = list(self.guests_by_context.get(self.current, self.guests)) + # copies: the app writes live status into these dicts; shared module constants must not change + snapshot = [ + dict(g) for g in self.guests_by_context.get(self.current, self.guests) + ] if ( self.list_hook ): # runs after reading: simulates a slow reply from that cluster @@ -52,7 +66,7 @@ class Fake: return snapshot def guest_config(self, guest): - return {"cores": guest["maxcpu"]} + return {"cores": guest["maxcpu"], **self.net_configs.get(guest["vmid"], {})} def disks(self, config, kind): return dict(self.disk_map) @@ -97,10 +111,61 @@ class Fake: raise self.use_error self.current = name + def interfaces(self, guest): + self.iface_calls.append(guest["vmid"]) + if self.iface_error: + raise self.iface_error + return list(self.iface_list) + + def backup_storages(self, guest): + return list(self.storages) + + def backup_jobs(self): + return list(self.jobs) + + def backups(self, guest, storages): + self.backups_calls.append((guest["vmid"], list(storages))) + return list(self.backup_list) + + def backup_now(self, guest, storage): + self.backup_runs.append((guest["vmid"], storage)) + if self.backup_hook: + self.backup_hook() + if self.backup_error: + raise self.backup_error + + +BACKUP_JOB = {"id": "backup-1", "enabled": 1, "storage": "truenas-backup", "all": 1, + "vmid": None, "exclude": None, "node": None, "schedule": "mon,wed,fri 04:00", + "mode": "snapshot", "compress": "zstd", "prune-backups": {"keep-last": "3"}} # fmt: skip +T_OLD, T_NEW = 1_762_000_000, 1_762_500_000 # epoch seconds +BACKUPS = [ + {"volid": "truenas-backup:backup/vzdump-qemu-108-new.vma.zst", "ctime": T_NEW, + "size": 3 * 2**30, "format": "vma.zst", "notes": "docker (manual)", "storage": "truenas-backup"}, + {"volid": "local:backup/vzdump-qemu-108-old.vma.zst", "ctime": T_OLD, + "size": int(1.5 * 2**30), "format": "vma.zst", "notes": "", "storage": "local"}, +] # fmt: skip +CT_CONFIG = { + "net0": "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=auto,type=veth" +} +VM_CONFIG = { + "net0": "virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,firewall=1", + "ipconfig0": "ip=dhcp", +} + @pytest.fixture def fake(monkeypatch): f = Fake() + # raising=False: the new names do not exist before the implementation (red run) + for name in ( + "interfaces", + "backup_storages", + "backup_jobs", + "backups", + "backup_now", + ): + monkeypatch.setattr(cli, name, getattr(f, name), raising=False) for name in ("list_guests", "guest_config", "disks", "action", "set_resources", "resize_disk", "pending_keys", "status", "contexts", "use_context"): # fmt: skip monkeypatch.setattr(cli, name, getattr(f, name)) @@ -921,3 +986,288 @@ async def test_terminal_is_reset_inside_suspend_before_ssh(fake, monkeypatch): async with app.run_test() as pilot: await press_t(app, pilot) assert resets == [[("enter", None)]] # inside suspend, before ssh ran + + +# --- delivery A: guest info screen / backup now ------------------------------ +# spec: docs/plans/guest-tui-mvp.md ("Guest info screen ... delivery A" > app.py) + + +def local_date(epoch): + return datetime.fromtimestamp(epoch).strftime("%Y-%m-%d %H:%M") + + +def cells(table, row): + return [str(c) for c in table.get_row_at(row)] + + +def cli_error(text): + with pytest.MonkeyPatch.context() as m: + m.setattr( + cli.subprocess, + "run", + lambda argv, **kw: subprocess.CompletedProcess(argv, 5, text, ""), + ) + with pytest.raises(cli.CliError) as exc: + cli.run("do", "x") + return exc.value + + +async def open_info(app, pilot, down=0): + await settle(app, pilot) + for _ in range(down): + await pilot.press("down") + await pilot.press("i") + await settle(app, pilot) + assert isinstance(app.screen, appmod.InfoScreen) + + +async def open_backup(app, pilot): + await pilot.press("b") + await settle(app, pilot) + assert isinstance(app.screen, appmod.BackupScreen) + + +async def test_i_opens_info_screen_with_backups_newest_first(fake): + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + table = app.screen.query_one("#backups", DataTable) + assert table.row_count == 2 + new, old = cells(table, 0), cells(table, 1) + assert local_date(T_NEW) in new and "truenas-backup" in new + assert any("3.0" in c for c in new) and "docker (manual)" in new + assert local_date(T_OLD) in old and "local" in old + assert any("1.5" in c for c in old) + assert fake.backups_calls == [(108, ["truenas-backup", "local"])] + + +async def test_info_jobs_line_names_schedule_storage_and_keep_last(fake): + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + text = str(app.screen.query_one("#jobs", Static).render()) + assert ( + "mon,wed,fri 04:00" in text + and "truenas-backup" in text + and "keep-last" in text + ) + + +async def test_info_jobs_line_when_no_job_covers_guest(fake): + fake.jobs = [] + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + text = str(app.screen.query_one("#jobs", Static).render()) + assert "No backup job covers this guest" in text + + +async def test_info_nets_row_for_ct_sample(fake): + fake.net_configs[100] = CT_CONFIG + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot, down=1) # CT 100 (stopped) + table = app.screen.query_one("#nets", DataTable) + assert table.row_count == 1 + assert cells(table, 0) == [ + "net0", + "vmbr0", + "-", + "BC:24:11:36:DE:78", + "dhcp", + "auto", + "no", + ] + + +async def test_info_nets_row_for_vm_uses_ipconfig_tag_and_firewall(fake): + fake.net_configs[108] = VM_CONFIG + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + table = app.screen.query_one("#nets", DataTable) + assert cells(table, 0) == [ + "net0", + "vmbr0", + "20", + "02:24:A6:49:A8:25", + "dhcp", + "-", + "yes", + ] + + +async def test_info_addresses_show_runtime_ips_for_running_guest(fake): + fake.iface_list = [{"name": "ens18", "mac": "02:24:A6:49:A8:25", + "ipv4": ["172.16.41.50/23"], "ipv6": ["fe80::1/64"]}] # fmt: skip + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + row = cells(app.screen.query_one("#addresses", DataTable), 0) + assert row[0] == "ens18" and row[1] == "02:24:A6:49:A8:25" + assert "172.16.41.50/23" in row[2] and "fe80::1/64" in row[3] + assert fake.iface_calls == [108] + + +async def test_info_does_not_query_interfaces_for_stopped_guest(fake): + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot, down=1) # CT 100 is stopped + assert fake.iface_calls == [] + + +async def test_info_interfaces_failure_is_inline_unavailable_not_fatal(fake): + fake.iface_error = cli_error("Error: QEMU guest agent is not running") + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + text = text_of(app.screen).lower() + assert "unavailable" in text and "guest agent is not running" in text + assert errors(app) == [] and app.is_running + assert ( + app.screen.query_one("#backups", DataTable).row_count == 2 + ) # rest still loaded + + +async def test_i_on_empty_table_does_nothing(fake): + fake.guests = [] + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + await pilot.press("i") + await settle(app, pilot) + assert not isinstance(app.screen, appmod.InfoScreen) + assert fake.backups_calls == [] + + +async def test_escape_closes_info_and_g_reloads_it(fake): + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + before = len(fake.backups_calls) + await pilot.press("g") + await settle(app, pilot) + assert len(fake.backups_calls) == before + 1 + await pilot.press("escape") + await settle(app, pilot) + assert not isinstance(app.screen, appmod.InfoScreen) + + +async def test_b_opens_backup_screen_preselecting_job_storage(fake): + fake.storages = ["local", "truenas-backup"] # job storage is not the first + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + await open_backup(app, pilot) + assert app.screen.query_one("#storage", Select).value == "truenas-backup" + text = text_of(app.screen).lower() + assert "snapshot" in text and "zstd" in text and "docker" in text + assert fake.backup_runs == [] + + +async def test_backup_storage_falls_back_to_first_without_job(fake): + fake.jobs = [] + fake.storages = ["local", "truenas-backup"] + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + await open_backup(app, pilot) + assert app.screen.query_one("#storage", Select).value == "local" + + +async def test_backup_yes_runs_backup_now_once_busy_then_done_and_reloads(fake): + app = make_app() + busy_seen = [] + fake.backup_hook = lambda: busy_seen.append(set(app.busy)) + async with app.run_test() as pilot: + await open_info(app, pilot) + before = len(fake.backups_calls) + await open_backup(app, pilot) + await press_button(app, pilot, "#yes") + assert fake.backup_runs == [(108, "truenas-backup")] + assert busy_seen and busy_seen[0], ( + "guest must be marked busy while the backup runs" + ) + assert app.busy == set() + msgs = [m.lower() for m, _ in app.seen] + assert any("backup started" in m for m in msgs) + assert any("backup done" in m for m in msgs) + assert errors(app) == [] + assert len(fake.backups_calls) > before # list reloaded + assert isinstance(app.screen, appmod.InfoScreen) + + +@pytest.mark.parametrize("how", ["no", "escape"]) +async def test_backup_declined_makes_no_call(fake, how): + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + await open_backup(app, pilot) + if how == "no": + await press_button(app, pilot, "#no") + else: + await pilot.press("escape") + await settle(app, pilot) + assert fake.backup_runs == [] + assert not isinstance(app.screen, appmod.BackupScreen) + assert app.busy == set() + + +async def test_backup_while_guest_busy_warns_and_makes_no_second_call(fake): + started, release = threading.Event(), threading.Event() + fake.backup_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + try: + await open_backup(app, pilot) + app.screen.query_one("#yes", Button).press() + for _ in range(200): + if started.is_set(): + break + await pilot.pause(0.01) + assert started.is_set() + await pilot.press("b") # second attempt while the first still runs + await pilot.pause() + if isinstance(app.screen, appmod.BackupScreen): # refusal may come at "yes" + app.screen.query_one("#yes", Button).press() + await pilot.pause() + assert any(sev == "warning" for _, sev in app.seen) + assert fake.backup_runs == [(108, "truenas-backup")] + finally: + release.set() + await settle(app, pilot) + assert fake.backup_runs == [(108, "truenas-backup")] + + +async def test_backup_now_failure_notifies_error_and_releases_busy(fake): + fake.backup_error = cli_error("Error: storage 'truenas-backup' is not online") + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + await open_backup(app, pilot) + await press_button(app, pilot, "#yes") + assert any("is not online" in m for m in errors(app)) + assert app.busy == set() and app.is_running + assert not any("backup done" in m.lower() for m, _ in app.seen) + + +# --- review findings: info screen robustness --------------------------------- + + +async def test_info_backup_with_missing_size_shows_dash_and_app_survives(fake): + fake.backup_list = [dict(BACKUPS[0], size=None)] + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot) + assert "-" in cells(app.screen.query_one("#backups", DataTable), 0) + assert app.is_running + + +async def test_info_uses_live_status_not_table_status(fake): + # the table status can be stale (listing lags); runtime IPs follow the live status + fake.live_status = {100: "running"} # CT 100 is "stopped" in the listing + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot, down=1) + assert fake.iface_calls == [100] diff --git a/tests/test_cli.py b/tests/test_cli.py index 98b1ac8..4bf5526 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -419,3 +419,267 @@ def test_use_context_unknown_name_raises_clierror(monkeypatch): with pytest.raises(cli.CliError) as exc: cli.use_context("x") assert "Context 'x' not found" in str(exc.value) + + +# --- delivery A: backups / jobs / net / interfaces --------------------------- +# spec: docs/plans/guest-tui-mvp.md ("Guest info screen ... delivery A" > cli.py) + +CT_NET = "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=auto,type=veth" +VM_NET = "virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,firewall=1" +VMID108 = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu"} + + +def storage(name, content="backup", enabled=1, active=1): + return {"storage": name, "content": content, "enabled": enabled, "active": active, + "type": "dir", "avail": 10**9} # fmt: skip + + +def test_backup_storages_filters_content_enabled_active_keeps_order(fake): + fake.result = ( + json.dumps([ + storage("zeta", "iso,backup,vztmpl"), + storage("nobackup", "iso,snippets,import"), + storage("off", enabled=0), + storage("inactive", active=0), + storage("alpha", "backup"), + ]), + "", + 0, + ) # fmt: skip + assert cli.backup_storages(VM) == ["zeta", "alpha"] + assert fake.argv == [ + "cv4pve-cli", + "get", + "node", + "storages", + "mainframe", + "-o", + "json", + ] + + +def test_backup_jobs_argv_and_returns_list(fake): + jobs = [{"id": "backup-1", "enabled": 1, "storage": "truenas-backup", "all": 1}] + fake.result = (json.dumps(jobs), "", 0) + assert cli.backup_jobs() == jobs + assert fake.argv == ["cv4pve-cli", "get", "cluster", "backup-jobs", "-o", "json"] + + +def job(storage="truenas-backup", enabled=1, all=0, vmid=None, exclude=None, node=None): + return {"id": "j", "enabled": enabled, "storage": storage, "all": all, "vmid": vmid, + "exclude": exclude, "node": node} # fmt: skip + + +@pytest.mark.parametrize( + "jobs,expected", + [ + ([job(all=1)], "truenas-backup"), # all=1 + ([job(all=1, exclude="108")], None), # excluded + ([job(all=1, exclude="100,108,200")], None), # excluded in a list + ([job(all=1, exclude="1080")], "truenas-backup"), # not a substring match + ([job(vmid="100,108,200")], "truenas-backup"), # vmid comma string + ([job(vmid="1080,200")], None), # not a substring match + ([job(vmid="100,200")], None), # not listed + ([job(all=1, node="othernode")], None), # node mismatch + ([job(all=1, node="mainframe")], "truenas-backup"), # node match + ([job(all=1, enabled=0)], None), # disabled + ([job(vmid=None, all=0)], None), # covers nothing + ([], None), # no job + ( + [job("first", enabled=0, all=1), job("second", all=1), job("third", all=1)], + "second", + ), + ], +) +def test_job_storage(jobs, expected): + assert cli.job_storage(jobs, VMID108) == expected + + +def test_backups_merges_storages_newest_first_and_adds_storage(monkeypatch): + per_storage = { + "truenas-backup": [ + { + "volid": "truenas-backup:backup/vzdump-qemu-108-a.vma.zst", + "ctime": 100, + "size": 1, + }, + { + "volid": "truenas-backup:backup/vzdump-qemu-108-b.vma.zst", + "ctime": 300, + "size": 3, + }, + ], + "local": [ + {"volid": "local:backup/vzdump-qemu-108-c.vma.zst", "ctime": 200, "size": 2} + ], + } + calls = [] + + def run(argv, **kwargs): + calls.append(list(argv)) + return subprocess.CompletedProcess( + argv, 0, json.dumps(per_storage[argv[5]]), "" + ) + + monkeypatch.setattr(cli.subprocess, "run", run) + result = cli.backups(VMID108, ["truenas-backup", "local"]) + assert [(b["ctime"], b["storage"]) for b in result] == [ + (300, "truenas-backup"), (200, "local"), (100, "truenas-backup"), + ] # fmt: skip + assert calls == [ + ["cv4pve-cli", "get", "guest", "backups", "mainframe", s, "108", "-o", "json"] + for s in ("truenas-backup", "local") + ] + + +def test_backup_now_exact_argv_without_yes(fake): + cli.backup_now(VMID108, "truenas-backup") + assert fake.argv == [ + "cv4pve-cli", "do", "backup", "guest", "mainframe", "108", "truenas-backup", + "--mode", "snapshot", "--compress", "zstd", + "--notes-template", "{{guestname}} (manual)", "--wait", + ] # fmt: skip + assert "-y" not in fake.argv and "--yes" not in fake.argv + + +def test_backup_now_failure_raises_clierror(monkeypatch): + FakeRun(monkeypatch, "Error: no space", returncode=5) + with pytest.raises(cli.CliError): + cli.backup_now(VMID108, "local") + + +def test_parse_net_ct_sample_keeps_order(): + net = cli.parse_net(CT_NET) + assert list(net.items()) == [ + ("name", "eth0"), ("bridge", "vmbr0"), ("hwaddr", "BC:24:11:36:DE:78"), + ("ip", "dhcp"), ("ip6", "auto"), ("type", "veth"), + ] # fmt: skip + + +def test_parse_net_vm_sample_first_item_is_model_mac(): + net = cli.parse_net(VM_NET) + assert list(net.items()) == [ + ("virtio", "02:24:A6:49:A8:25"), ("bridge", "vmbr0"), ("tag", "20"), ("firewall", "1"), + ] # fmt: skip + + +def test_nets_only_netN_keys_sorted_numerically(): + config = { + "cores": 4, + "net10": VM_NET, + "net2": CT_NET, + "ipconfig0": "ip=dhcp", + "net0": VM_NET, + } + result = cli.nets(config) + assert list(result) == ["net0", "net2", "net10"] + assert result["net2"] == cli.parse_net(CT_NET) + + +def test_interfaces_ct_inet_inet6_and_excludes_lo(fake): + fake.result = ( + json.dumps([ + {"name": "lo", "hwaddr": "00:00:00:00:00:00", "inet": "127.0.0.1/8", "inet6": "::1/128"}, + {"name": "eth0", "hwaddr": "BC:24:11:36:DE:78", "inet": "172.16.41.200/23", + "inet6": "fe80::be24:11ff:fe36:de78/64", "ip-addresses": "[]"}, + {"name": "eth1", "hwaddr": "BC:24:11:36:DE:79", "inet": "10.0.0.5/24", "ip-addresses": "[]"}, + ]), + "", + 0, + ) # fmt: skip + assert cli.interfaces(CT) == [ + {"name": "eth0", "mac": "BC:24:11:36:DE:78", "ipv4": ["172.16.41.200/23"], + "ipv6": ["fe80::be24:11ff:fe36:de78/64"]}, + {"name": "eth1", "mac": "BC:24:11:36:DE:79", "ipv4": ["10.0.0.5/24"], "ipv6": []}, + ] # fmt: skip + assert fake.argv == [ + "cv4pve-cli", + "get", + "ct", + "interfaces", + "mainframe", + "100", + "-o", + "json", + ] + + +def test_interfaces_vm_agent_result_json_string_and_excludes_lo(fake): + agent = [ + {"name": "lo", "hardware-address": "00:00:00:00:00:00", + "ip-addresses": [{"ip-address": "127.0.0.1", "prefix": 8, "ip-address-type": "ipv4"}]}, + {"name": "ens18", "hardware-address": "02:24:A6:49:A8:25", + "ip-addresses": [ + {"ip-address": "172.16.41.50", "prefix": 23, "ip-address-type": "ipv4"}, + {"ip-address": "fe80::24:a6ff:fe49:a825", "prefix": 64, "ip-address-type": "ipv6"}, + {"ip-address": "2001:db8::5", "prefix": 64, "ip-address-type": "ipv6"}, + ]}, + ] # fmt: skip + fake.result = (json.dumps([{"key": "result", "value": json.dumps(agent)}]), "", 0) + assert cli.interfaces(VMID108) == [ + {"name": "ens18", "mac": "02:24:A6:49:A8:25", "ipv4": ["172.16.41.50/23"], + "ipv6": ["fe80::24:a6ff:fe49:a825/64", "2001:db8::5/64"]}, + ] # fmt: skip + assert fake.argv == [ + "cv4pve-cli", + "get", + "vm", + "agent", + "network", + "mainframe", + "108", + "-o", + "json", + ] + + +def test_interfaces_vm_agent_failure_raises_clierror(monkeypatch): + FakeRun(monkeypatch, "Error: QEMU guest agent is not running", returncode=5) + with pytest.raises(cli.CliError): + cli.interfaces(VMID108) + + +def test_backup_jobs_parses_prune_backups_json_string(fake): + # real cv4pve-cli output: prune-backups is a JSON *string* + fake.result = ( + json.dumps( + [ + {"id": "j", "prune-backups": '{"keep-last":"3"}'}, + {"id": "k", "prune-backups": None}, + ] + ), + "", + 0, + ) + jobs = cli.backup_jobs() + assert jobs[0]["prune-backups"] == {"keep-last": "3"} + assert jobs[1]["prune-backups"] is None + + +def test_interfaces_vm_tolerates_null_addresses_and_unknown_types(fake): + result = [ + {"name": "eth0", "hardware-address": "aa", "ip-addresses": None}, + {"name": "eth1", "ip-addresses": [{"ip-address-type": "ipv5", "ip-address": "x", "prefix": 1}, + {"ip-address-type": "ipv4", "ip-address": "10.0.0.2", "prefix": 24}]}, + ] # fmt: skip + fake.result = (json.dumps([{"key": "result", "value": json.dumps(result)}]), "", 0) + found = cli.interfaces({"vmid": 108, "node": "mainframe", "type": "qemu"}) + assert found == [ + {"name": "eth0", "mac": "aa", "ipv4": [], "ipv6": []}, + {"name": "eth1", "mac": None, "ipv4": ["10.0.0.2/24"], "ipv6": []}, + ] + + +def test_interfaces_ct_lists_all_addresses_incl_global_ipv6(fake): + # real output: inet6 holds only one address (link-local); ip-addresses (JSON string) has all + addrs = [ + {"prefix": "23", "ip-address": "172.16.41.200", "ip-address-type": "inet"}, + {"ip-address-type": "inet6", "ip-address": "2a02:fe1:e77e:6004:be24:11ff:fe36:de78", "prefix": "64"}, + {"prefix": "64", "ip-address": "fe80::be24:11ff:fe36:de78", "ip-address-type": "inet6"}, + ] # fmt: skip + eth0 = {"name": "eth0", "hwaddr": "bc:24:11:36:de:78", "inet": "172.16.41.200/23", + "inet6": "fe80::be24:11ff:fe36:de78/64", "ip-addresses": json.dumps(addrs)} # fmt: skip + fake.result = (json.dumps([eth0]), "", 0) + found = cli.interfaces({"vmid": 104, "node": "mainframe", "type": "lxc"}) + assert found == [{"name": "eth0", "mac": "bc:24:11:36:de:78", "ipv4": ["172.16.41.200/23"], + "ipv6": ["2a02:fe1:e77e:6004:be24:11ff:fe36:de78/64", "fe80::be24:11ff:fe36:de78/64"]}] # fmt: skip