diff --git a/README.md b/README.md index fe489d6..657aee6 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ uv run cv4pve-tui | `Enter` then `e` | in info: edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) | | `Enter` then `b` | in info: backup now (snapshot, zstd, storage choice); `a` there = also show backups of earlier guests that had the same VMID | | `Enter` then `n` | in info: edit network (bridge, VLAN, firewall, IPv4/IPv6; VM IP via cloud-init at next boot); `u` there = undo the last network change of this guest (this session only) | +| `Enter` then `p` / `R` / `D` | in info: take a snapshot (name, description, VM: include RAM) / roll back to the snapshot under the cursor (confirms; starts the guest again if it is running) / delete the snapshot under the cursor (confirms) | | `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 2a3fcb8..abbe00f 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -446,3 +446,69 @@ API facts (Proxmox VE 9.2, `api usage ... config set -v`): 172.16.41.200/23 gw 172.16.40.1->DHCP. Bridge change cannot be tested live: the node has a single bridge (`vmbr0`). Final net0 string identical to the baseline. - [ ] Push `feat/network-edit`, merge to `main` (ask first). + +--- + +# Snapshots in the info screen (issue #8) + +## Decision (user-approved) + +Info screen section "Snapshots"; keys `p` take, `R` roll back, `D` delete. LXC and VM. + +CLI facts (cv4pve-cli 2.4.0, read-only probes): + +- `get guest snapshots -o json` -> list of dicts: `name`, + `description`, `snaptime` (epoch or null), `parent`, VM also `vmstate`. Always + contains a pseudo entry `name == "current"` ("You are here!") - not a snapshot. +- `create ct|vm snapshot ` (+ `--vmstate 1` for VM RAM), + no `--yes`; with `--wait`. +- `do rollback ct|vm ` needs `-y`; API param `start` (bool) + starts the guest after a successful rollback. +- `delete ct|vm snapshot ` needs `-y`. + DANGER: `delete ct|vm -y` (without `snapshot`) DESTROYS THE WHOLE GUEST. + +### `cli.py` + +- `valid_snapname(name) -> str` - ValueError unless it matches + `[A-Za-z][A-Za-z0-9_-]{0,39}` and is not `current` (case-insensitive). +- `snapshots(guest) -> list[dict]` - without `current`, newest first (`snaptime` + desc, null last). +- `take_snapshot(guest, name, description="", vmstate=False)` -> + `create snapshot + [--vmstate 1 (VM only)] --wait`; name validated first. +- `rollback_snapshot(guest, name, start=False)` -> + `do rollback [--start 1] -y --wait`; name validated. +- `delete_snapshot(guest, name)` -> + `delete snapshot -y --wait`; name validated first, and + the argv is asserted to contain `snapshot` right after the kind and the name right + after the vmid (defence against ever sending the guest-destroying form). + +### `app.py` (UI contract) + +- InfoScreen gets `Vertical#snapshots_section` (border_title "Snapshots") with + `DataTable#snapshots` (Name, Date `YYYY-MM-DD HH:MM` or `-`, Description, and for VMs + RAM `yes`/`no`) and `Static#snap_note`. Loaded with the rest; a failure of + `cli.snapshots` is shown in `#snap_note` ("unavailable: "), the rest of the + screen still works. Empty list -> `#snap_note` "No snapshots". +- `p`: `SnapshotScreen` dialog (DialogScreen style, title "Take snapshot: + ()"): `Input#snapname` prefilled `tui_YYYYMMDD_HHMM` (local time), + `Input#snapdesc`, `Checkbox#vmstate` (VMs only; not shown for CTs), buttons + `#yes`/`#no`, escape = no. Invalid name -> error notify, dialog stays, no call. + Yes -> busy guard, worker `cli.take_snapshot`, done/error notify, info reloads. +- `R`: roll back the snapshot under the cursor of `#snapshots`. `ConfirmScreen` + titled "Roll back" naming snapshot and date and saying everything changed since then + is lost; if the guest is running (live status) it says it will be started again and + `start=True` is passed. Yes -> busy guard, worker, notify, info reloads. +- `D`: delete the snapshot under the cursor. `ConfirmScreen` titled "Delete snapshot" + naming it. Yes -> busy guard, worker `cli.delete_snapshot`, notify, info reloads. +- `R`/`D` with no snapshot -> info notify, no dialog. + +## Steps + +- [x] Tests - red first. - [x] Implementation - suite green, ruff clean, README keys. +- Review fixes: rollback reads the live status right before rolling back (if it cannot + be read, no rollback); the info screen scrolls; a description starting with `-` is + refused (positional argument). +- [x] Live on CT 104: take `tui_live_test`, listed; roll back (CT running -> started + again); delete; list back to the baseline (no snapshots); CT running. +- [ ] PR with `Closes #8`, merge after user OK. diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index e1213db..c4c9654 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -8,7 +8,7 @@ from rich.text import Text from textual import work from textual.app import App, ComposeResult, SuspendNotSupported from textual.binding import Binding -from textual.containers import Horizontal, Vertical +from textual.containers import Horizontal, Vertical, VerticalScroll from textual.screen import ModalScreen from textual.widgets import ( Button, @@ -202,6 +202,51 @@ class BackupScreen(DialogScreen): self.dismiss(None) +class SnapshotScreen(DialogScreen): + """Dismisses with {name, description, vmstate}, or None when cancelled.""" + + BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom + + def __init__(self, guest: dict) -> None: + super().__init__() + self.guest = guest + + def compose(self) -> ComposeResult: + with self.box(f"Take snapshot: {self.guest['name']} ({self.guest['vmid']})"): + with Horizontal(classes="row"): + yield Label("Name") + yield Input(datetime.now().strftime("tui_%Y%m%d_%H%M"), id="snapname") + with Horizontal(classes="row"): + yield Label("Description") + yield Input(placeholder="optional", id="snapdesc") + if cli.kind(self.guest) == "vm": + yield Checkbox("Include RAM (vmstate)", id="vmstate") + with Horizontal(classes="buttons"): + 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(None) + return + try: + name = cli.valid_snapname(self.query_one("#snapname", Input).value) + description = cli.valid_description( + self.query_one("#snapdesc", Input).value.strip() + ) + except ValueError as exc: + self.app.notify(str(exc), severity="error") + return + vmstate = next(iter(self.query("#vmstate").results(Checkbox)), None) + self.dismiss( + { + "name": name, + "description": description, + "vmstate": bool(vmstate and vmstate.value), + } + ) + + _IPV4_MODES = [("DHCP", "dhcp"), ("Static", "static"), ("None", "none")] _IPV6_MODES = [("SLAAC", "auto"), *_IPV4_MODES] @@ -383,6 +428,9 @@ class InfoScreen(ModalScreen): ("n", "net", "Edit network"), ("u", "undo", "Undo network"), ("e", "edit", "Edit resources"), + ("p", "snapshot", "Take snapshot"), + ("R", "rollback", "Roll back"), + ("D", "delete_snapshot", "Delete snapshot"), ] DEFAULT_CSS = """ InfoScreen DataTable { height: auto; max-height: 12; } @@ -403,9 +451,10 @@ class InfoScreen(ModalScreen): self.show_all = False self.config: dict = {} self.running = False + self.snaps: list[dict] = [] def compose(self) -> ComposeResult: - with Vertical(): + with VerticalScroll(id="info_body"): # four sections do not fit short terminals yield Static(f"{self.guest['name']} ({self.guest['vmid']})", markup=False) with Vertical(id="resources_section", classes="section") as res: res.border_title = "Resources" @@ -415,6 +464,10 @@ class InfoScreen(ModalScreen): yield Static("", id="jobs", markup=False) yield DataTable(id="backups") yield Static("", id="hidden", markup=False) + with Vertical(id="snapshots_section", classes="section") as snaps: + snaps.border_title = "Snapshots" + yield DataTable(id="snapshots", cursor_type="row") + yield Static("", id="snap_note", markup=False) with Vertical(id="network_section", classes="section") as network: network.border_title = "Network" yield Static("Configured", classes="label first") @@ -422,13 +475,19 @@ class InfoScreen(ModalScreen): yield Static("In use", classes="label") yield Static("", id="unavailable", markup=False) yield DataTable(id="addresses") - yield Footer() + yield Footer() def on_mount(self) -> None: self.query_one("#resources", DataTable).add_columns("Resource", "Value") self.query_one("#backups", DataTable).add_columns( "Date", "Storage", "Size (GiB)", "Format", "Notes" ) + self.query_one("#snapshots", DataTable).add_columns( + "Name", + "Date", + "Description", + *(["RAM"] if cli.kind(self.guest) == "vm" else []), + ) self.query_one("#nets", DataTable).add_columns( "Iface", "Bridge", "VLAN", "MAC", "IPv4", "IPv6", "Firewall" ) @@ -466,20 +525,25 @@ class InfoScreen(ModalScreen): addresses = cli.interfaces(guest) except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works unavailable = f"unavailable: {_message(exc)}" + snaps, snap_note = [], "" + try: + snaps = cli.snapshots(guest) + except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works + snap_note = f"unavailable: {_message(exc)}" self.running = running self.app.call_from_thread( - self._show, config, nets, storages, jobs, backups, addresses, unavailable + self._show, + (config, nets, storages, jobs, backups, addresses, unavailable), + snaps, + snap_note, ) - def _show( - self, config, nets, storages, jobs, backups, addresses, unavailable - ) -> None: + def _show(self, loaded: tuple, snaps: list[dict], snap_note: str) -> None: if not self.is_attached: # closed while loading return try: - self._fill_screen( - config, nets, storages, jobs, backups, addresses, unavailable - ) + self._fill_screen(*loaded) + self._show_snapshots(snaps, snap_note) except Exception as exc: # noqa: BLE001 - odd API data must not kill the app self.app.notify(f"info: {_message(exc)}", severity="error") @@ -527,6 +591,22 @@ class InfoScreen(ModalScreen): ", ".join(i["ipv6"]), ) + def _show_snapshots(self, snaps: list[dict], note: str) -> None: + self.snaps = snaps + table = self.query_one("#snapshots", DataTable).clear() + vm = cli.kind(self.guest) == "vm" + for s in snaps: + _add_plain( + table, + s["name"], + _when(s.get("snaptime")), + s.get("description") or "", + *(["yes" if s.get("vmstate") else "no"] if vm else []), + ) + self.query_one("#snap_note", Static).update( + note or ("" if snaps else "No snapshots") + ) + def _show_backups(self) -> None: name = self.guest["name"] shown = [ @@ -600,6 +680,78 @@ class InfoScreen(ModalScreen): lambda ok: self.app.start_net_change(self.guest, old, None) if ok else None, ) + def _snapshot_under_cursor(self) -> dict | None: + """The snapshot at the cursor; None (with a notify) when there is none or the guest is busy.""" + row = self.query_one("#snapshots", DataTable).cursor_row + if not 0 <= row < len(self.snaps): + self.app.notify("No snapshots") + return None + return None if self._refuse_if_busy() else self.snaps[row] + + def action_snapshot(self) -> None: + if self._refuse_if_busy(): + return + self.app.push_screen(SnapshotScreen(self.guest), self._take_snapshot) + + def _take_snapshot(self, values: dict | None) -> None: + if values: + self.app.start_job( + self.guest, + "Snapshot", + lambda: cli.take_snapshot( + self.guest, + values["name"], + values["description"], + values["vmstate"], + ), + ) + + def action_rollback(self) -> None: + snap = self._snapshot_under_cursor() + if snap is None: + return + text = ( + f"Roll back to snapshot {snap['name']} ({_when(snap.get('snaptime'))})? " + "Everything changed since then is lost. " + "If the guest is running, it is started again afterwards." + ) + + def job() -> None: + # status read right before the rollback (not at screen load); if it cannot be + # read the rollback does not run, rather than leave a running guest stopped + start = cli.status(self.guest) == "running" + cli.rollback_snapshot(self.guest, snap["name"], start) + + self.app.push_screen( + ConfirmScreen(text, "Roll back"), + lambda ok: ( + self.app.start_job( + self.guest, + "Rollback", + job, + ) + if ok + else None + ), + ) + + def action_delete_snapshot(self) -> None: + snap = self._snapshot_under_cursor() + if snap is None: + return + self.app.push_screen( + ConfirmScreen(f"Delete snapshot {snap['name']}?", "Delete snapshot"), + lambda ok: ( + self.app.start_job( + self.guest, + "Delete snapshot", + lambda: cli.delete_snapshot(self.guest, snap["name"]), + ) + if ok + else None + ), + ) + def action_backup(self) -> None: if not self.storages: self.app.notify("No backup storage available", severity="warning") @@ -615,6 +767,10 @@ class InfoScreen(ModalScreen): ) +def _when(epoch: int | None) -> str: + return datetime.fromtimestamp(epoch).strftime("%Y-%m-%d %H:%M") if epoch else "-" + + def _ip_changed(old: str | None, new: str | None) -> bool: before, after = cli.parse_net(old or ""), cli.parse_net(new or "") return any(before.get(k) != after.get(k) for k in ("ip", "gw", "ip6", "gw6")) @@ -833,7 +989,7 @@ class GuestApp(App): finally: self.call_from_thread(setattr, self, "switching", False) - def _start_job(self, guest: dict, label: str, job) -> None: + 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" @@ -869,7 +1025,7 @@ class GuestApp(App): return run = lambda: cli.action(name, guest) if name == "start": - self._start_job(guest, name, run) + self.start_job(guest, name, run) return who = f"{guest['name']} ({guest['vmid']})" text = f"{name.capitalize()} {who}?" @@ -877,7 +1033,7 @@ class GuestApp(App): text = f"Hard power-off {who}? The guest is killed without shutting down." self.push_screen( ConfirmScreen(text, name.capitalize()), - lambda ok: self._start_job(guest, name, run) if ok else None, + lambda ok: self.start_job(guest, name, run) if ok else None, ) def edit_guest(self, guest: dict) -> None: @@ -1056,7 +1212,7 @@ class GuestApp(App): if {"cores", "memory"} & set(pending): self.call_from_thread(self.notify, "Change applies at next reboot") - self._start_job(guest, "edit", job) + self.start_job(guest, "edit", job) def _reset_terminal() -> None: diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index 7760ad2..6a5cc9e 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -385,3 +385,61 @@ def not_backed_up() -> set[int]: """VMIDs no backup job covers (Proxmox's own answer, incl. pool-based jobs).""" items = json.loads(run("get", "cluster", "not-backed-up", "-o", "json")) return {int(i["vmid"]) for i in items if i.get("vmid") is not None} + + +_SNAPNAME = re.compile(r"[A-Za-z][A-Za-z0-9_-]{0,39}") + + +def valid_snapname(name: str) -> str: + """The name unchanged, or ValueError; `current` is Proxmox's pseudo entry.""" + if not _SNAPNAME.fullmatch(name) or name.lower() == "current": + raise ValueError( + "snapshot name: a letter, then letters, digits, _ or - (max 40); not 'current'" + ) + return name + + +def snapshots(guest: dict) -> list[dict]: + """The guest's snapshots without the `current` pseudo entry, newest first.""" + out = run( + "get", "guest", "snapshots", guest["node"], _api_type(guest), + str(guest["vmid"]), "-o", "json", + ) # fmt: skip + found = [s for s in json.loads(out) if s["name"].lower() != "current"] + return sorted(found, key=lambda s: s.get("snaptime") or 0, reverse=True) + + +def valid_description(text: str) -> str: + """Passed as a positional argument: a leading '-' could be read as an option.""" + if text.startswith("-"): + raise ValueError("description must not start with '-'") + return text + + +def take_snapshot( + guest: dict, name: str, description: str = "", vmstate: bool = False +) -> None: + opts = ["--vmstate", "1"] if vmstate and kind(guest) == "vm" else [] + run( + "create", kind(guest), "snapshot", guest["node"], str(guest["vmid"]), + valid_snapname(name), valid_description(description) or "cv4pve-tui", *opts, "--wait", + ) # fmt: skip + + +def rollback_snapshot(guest: dict, name: str, start: bool = False) -> None: + opts = ["--start", "1"] if start else [] + run( + "do", "rollback", kind(guest), guest["node"], str(guest["vmid"]), + valid_snapname(name), *opts, "-y", "--wait", + ) # fmt: skip + + +def delete_snapshot(guest: dict, name: str) -> None: + """DANGER: without the `snapshot` word `delete -y` destroys the guest.""" + args = ( + "delete", kind(guest), "snapshot", guest["node"], str(guest["vmid"]), + valid_snapname(name), "-y", "--wait", + ) # fmt: skip + if args[2] != "snapshot" or args[4:6] != (str(guest["vmid"]), name): + raise RuntimeError("refusing to send a delete that is not a snapshot delete") + run(*args) diff --git a/tests/test_app.py b/tests/test_app.py index 909a45d..44eda92 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -12,6 +12,7 @@ import pytest from textual.screen import ModalScreen from textual.widgets import ( Button, + Checkbox, DataTable, Input, OptionList, @@ -63,6 +64,13 @@ class Fake: # delivery B (network editing) self.bridge_list = ["vmbr0"] self.set_calls, self.set_error, self.set_hook = [], None, None + # issue #8 (snapshots) + self.snap_list = [dict(s) for s in SNAPSHOTS] # newest first, no "current" + self.snap_error = None + self.snap_calls = [] + self.take_calls, self.rollback_calls, self.delete_calls = [], [], [] + self.snap_error_op = None + self.snap_hook = None def list_guests(self): self.list_calls += 1 @@ -169,11 +177,37 @@ class Fake: else: config[key] = value + def snapshots(self, guest): + self.snap_calls.append(guest["vmid"]) + if self.snap_error: + raise self.snap_error + return [dict(s) for s in self.snap_list] + + def _snap_op(self, calls, entry): + calls.append(entry) + if self.snap_hook: + self.snap_hook() + if self.snap_error_op: + raise self.snap_error_op + + def take_snapshot(self, guest, name, description="", vmstate=False): + self._snap_op(self.take_calls, (guest["vmid"], name, description, vmstate)) + + def rollback_snapshot(self, guest, name, start=False): + self._snap_op(self.rollback_calls, (guest["vmid"], name, start)) + + def delete_snapshot(self, guest, name): + self._snap_op(self.delete_calls, (guest["vmid"], name)) + 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 +SNAPSHOTS = [ # as cli.snapshots returns them: newest first, without "current" + {"name": "newer", "description": "second one", "snaptime": T_NEW, "parent": "older", "vmstate": 1}, + {"name": "older", "description": "first one", "snaptime": T_OLD, "parent": None, "vmstate": 0}, +] # fmt: skip 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"}, @@ -202,6 +236,10 @@ def fake(monkeypatch): "bridges", "set_net", "not_backed_up", + "snapshots", + "take_snapshot", + "rollback_snapshot", + "delete_snapshot", ): monkeypatch.setattr(cli, name, getattr(f, name), raising=False) for name in ("list_guests", "guest_config", "disks", "action", "set_resources", @@ -2151,3 +2189,460 @@ async def test_backup_coverage_failure_still_lists_guests(fake): assert app.query_one("#guests", DataTable).row_count == 2 assert any(sev == "warning" for _, sev in app.seen) # coverage unknown, said so assert app.is_running + + +# --- snapshots in the info screen (issue #8) --------------------------------- +# spec: docs/plans/guest-tui-mvp.md ("Snapshots in the info screen (issue #8)" > app.py) + + +async def open_snapshot_dialog(app, pilot): + await pilot.press("p") + await settle(app, pilot) + assert isinstance(app.screen, appmod.SnapshotScreen) + + +async def wait_started(pilot, event): + for _ in range(200): + if event.is_set(): + return + await pilot.pause(0.01) + raise AssertionError("worker never started") + + +async def test_info_snapshots_section_lists_newest_first_with_ram_for_vm(fake): + # InfoScreen: Vertical#snapshots_section "Snapshots", DataTable#snapshots Name/Date/Description/RAM + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + section = app.screen.query_one("#snapshots_section") + assert "Snapshots" in str(section.border_title) + table = app.screen.query_one("#snapshots", DataTable) + assert table.row_count == 2 + newer, older = cells(table, 0), cells(table, 1) + assert "newer" in newer and local_date(T_NEW) in newer and "second one" in newer + assert "older" in older and local_date(T_OLD) in older and "first one" in older + assert "yes" in newer and "no" in older + assert fake.snap_calls == [108] + + +async def test_info_snapshots_ct_has_no_ram_column(fake): + # DataTable#snapshots: RAM column for VMs only + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot, down=1) # CT 100 + table = app.screen.query_one("#snapshots", DataTable) + assert table.row_count == 2 + assert len(table.columns) == 3 + assert "yes" not in cells(table, 0) + + +async def test_info_snapshot_without_date_shows_dash(fake): + # Date: YYYY-MM-DD HH:MM or "-" + fake.snap_list = [ + { + "name": "nodate", + "description": "d", + "snaptime": None, + "parent": None, + "vmstate": 0, + } + ] + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + assert "-" in cells(app.screen.query_one("#snapshots", DataTable), 0) + assert app.is_running + + +async def test_info_no_snapshots_shows_note(fake): + # empty list -> #snap_note "No snapshots" + fake.snap_list = [] + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + assert app.screen.query_one("#snapshots", DataTable).row_count == 0 + assert "No snapshots" in str( + app.screen.query_one("#snap_note", Static).render() + ) + + +async def test_info_snapshots_failure_is_inline_unavailable_and_rest_works(fake): + # a failure of cli.snapshots -> "unavailable: " in #snap_note; rest of the screen still works + fake.snap_error = cli_error("Error: snapshot listing exploded") + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + note = str(app.screen.query_one("#snap_note", Static).render()) + assert "unavailable" in note and "exploded" in note + assert app.screen.query_one("#backups", DataTable).row_count == 2 + assert isinstance(app.screen, appmod.InfoScreen) and app.is_running + + +async def test_p_opens_snapshot_dialog_with_prefilled_name_and_title(fake): + # p: SnapshotScreen, title "Take snapshot: ()", Input#snapname tui_YYYYMMDD_HHMM + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + before = datetime.now().strftime("tui_%Y%m%d_%H%M") + await open_snapshot_dialog(app, pilot) + after = datetime.now().strftime("tui_%Y%m%d_%H%M") + assert app.screen.query_one("#snapname", Input).value in {before, after} + title = str(app.screen.query_one("#dialog").border_title) + assert "Take snapshot" in title and "docker" in title and "108" in title + app.screen.query_one("#snapdesc", Input) + app.screen.query_one("#yes", Button) + app.screen.query_one("#no", Button) + assert fake.take_calls == [] + + +async def test_snapshot_dialog_vmstate_checkbox_only_for_vm(fake): + # Checkbox#vmstate: VMs only; not shown for CTs + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot, down=1) + await open_snapshot_dialog(app, pilot) + assert len(app.screen.query("#vmstate")) == 0 + await pilot.press("escape") + await settle(app, pilot) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await open_snapshot_dialog(app, pilot) + assert len(app.screen.query("#vmstate")) == 1 + + +@pytest.mark.parametrize("how", ["no", "escape"]) +async def test_snapshot_dialog_declined_makes_no_call(fake, how): + # SnapshotScreen: buttons #yes/#no, escape = no + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await open_snapshot_dialog(app, pilot) + if how == "no": + await press_button(app, pilot, "#no") + else: + await pilot.press("escape") + await settle(app, pilot) + assert fake.take_calls == [] + assert isinstance(app.screen, appmod.InfoScreen) + assert app.busy == set() + + +@pytest.mark.parametrize("bad", ["", "my snap", "1abc", "current", "a;rm", "x" * 41]) +async def test_snapshot_invalid_name_error_notify_dialog_stays_no_call(fake, bad): + # Invalid name -> error notify, dialog stays, no call + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await open_snapshot_dialog(app, pilot) + app.screen.query_one("#snapname", Input).value = bad + await press_button(app, pilot, "#yes") + assert errors(app), "an error notification is expected" + assert isinstance(app.screen, appmod.SnapshotScreen) + assert fake.take_calls == [] and app.busy == set() + + +async def test_snapshot_yes_takes_vm_snapshot_once_busy_then_reloads(fake): + # Yes -> busy guard, worker cli.take_snapshot(guest, name, desc, vmstate), notify, info reloads + app = make_app() + busy_seen = [] + fake.snap_hook = lambda: busy_seen.append(set(app.busy)) + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + before = len(fake.snap_calls) + await open_snapshot_dialog(app, pilot) + app.screen.query_one("#snapname", Input).value = "before_upgrade" + app.screen.query_one("#snapdesc", Input).value = "pre upgrade" + app.screen.query_one("#vmstate", Checkbox).value = True + await press_button(app, pilot, "#yes") + assert fake.take_calls == [(108, "before_upgrade", "pre upgrade", True)] + assert busy_seen and busy_seen[0], ( + "guest must be busy while the snapshot is taken" + ) + assert app.busy == set() + assert errors(app) == [] and app.seen + assert len(fake.snap_calls) > before # info reloaded + assert isinstance(app.screen, appmod.InfoScreen) + + +async def test_snapshot_vmstate_unticked_passes_false(fake): + # Checkbox#vmstate off -> vmstate False + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await open_snapshot_dialog(app, pilot) + app.screen.query_one("#snapname", Input).value = "plain" + app.screen.query_one("#snapdesc", Input).value = "d" + app.screen.query_one("#vmstate", Checkbox).value = False + await press_button(app, pilot, "#yes") + assert fake.take_calls == [(108, "plain", "d", False)] + + +async def test_snapshot_ct_never_passes_vmstate(fake): + # CT: no checkbox, so vmstate is False + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot, down=1) + await open_snapshot_dialog(app, pilot) + app.screen.query_one("#snapname", Input).value = "ctsnap" + app.screen.query_one("#snapdesc", Input).value = "d" + await press_button(app, pilot, "#yes") + assert fake.take_calls == [(100, "ctsnap", "d", False)] + + +async def test_snapshot_failure_notifies_error_and_releases_busy(fake): + # CliError -> error notify, busy released + fake.snap_error_op = cli_error("Error: snapshot is not possible: no space left") + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await open_snapshot_dialog(app, pilot) + app.screen.query_one("#snapname", Input).value = "willfail" + await press_button(app, pilot, "#yes") + assert any("no space left" in m for m in errors(app)) + assert app.busy == set() and app.is_running + + +# --- rollback --------------------------------------------------------------- + + +async def open_confirm_on_row(app, pilot, key, row=1): + await open_info(app, pilot) + app.screen.query_one("#snapshots", DataTable).move_cursor(row=row) + await pilot.pause() + await pilot.press(key) + await settle(app, pilot) + assert isinstance(app.screen, ConfirmScreen) + + +async def test_R_confirm_names_second_snapshot_and_rolls_back_running_vm_with_start( + fake, +): + # R: ConfirmScreen "Roll back" naming snapshot+date, data-loss warning, running -> started again, start=True + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, "R") + assert "Roll back" in str(app.screen.query_one("#dialog").border_title) + text = text_of(app.screen) + assert "older" in text and "newer" not in text + assert local_date(T_OLD)[:10] in text + assert "lost" in text.lower() + assert "started" in text.lower() + assert fake.rollback_calls == [] + await press_button(app, pilot, "#yes") + assert fake.rollback_calls == [(108, "older", True)] + assert app.busy == set() and errors(app) == [] + + +async def test_R_uses_live_status_for_start_flag(fake): + # "running (live status)": CT is "stopped" in the listing but running live -> start=True + fake.live_status = {100: "running"} + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot, down=1) + app.screen.query_one("#snapshots", DataTable).move_cursor(row=0) + await pilot.press("R") + await settle(app, pilot) + await press_button(app, pilot, "#yes") + assert fake.rollback_calls == [(100, "newer", True)] + + +@pytest.mark.parametrize("guest_down,vmid", [(0, 108), (1, 100)]) +async def test_R_stopped_guest_rolls_back_without_start(fake, guest_down, vmid): + # stopped (live status; VM listing says running but live says stopped) -> start=False + fake.live_status = {108: "stopped"} + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot, down=guest_down) + app.screen.query_one("#snapshots", DataTable).move_cursor(row=1) + await pilot.press("R") + await settle(app, pilot) + await press_button(app, pilot, "#yes") + assert fake.rollback_calls == [(vmid, "older", False)] + + +@pytest.mark.parametrize("key,calls", [("R", "rollback_calls"), ("D", "delete_calls")]) +@pytest.mark.parametrize("how", ["no", "escape"]) +async def test_R_D_declined_make_no_call(fake, key, calls, how): + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, key) + if how == "no": + await press_button(app, pilot, "#no") + else: + await pilot.press("escape") + await settle(app, pilot) + assert getattr(fake, calls) == [] + assert isinstance(app.screen, appmod.InfoScreen) and app.busy == set() + + +async def test_R_failure_notifies_error_and_releases_busy(fake): + fake.snap_error_op = cli_error("Error: rollback refused: snapshot is locked") + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, "R") + await press_button(app, pilot, "#yes") + assert any("locked" in m for m in errors(app)) + assert app.busy == set() and app.is_running + + +async def test_R_reloads_info_after_rollback(fake): + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, "R") + before = len(fake.snap_calls) + await press_button(app, pilot, "#yes") + assert len(fake.snap_calls) > before + + +# --- delete ----------------------------------------------------------------- + + +async def test_D_confirm_names_snapshot_then_deletes_it_busy_and_reloads(fake): + # D: ConfirmScreen "Delete snapshot" naming it; yes -> busy guard, cli.delete_snapshot, notify, reload + app = make_app() + busy_seen = [] + fake.snap_hook = lambda: busy_seen.append(set(app.busy)) + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, "D") + assert "Delete snapshot" in str(app.screen.query_one("#dialog").border_title) + text = text_of(app.screen) + assert "older" in text and "newer" not in text + assert fake.delete_calls == [] + before = len(fake.snap_calls) + await press_button(app, pilot, "#yes") + assert fake.delete_calls == [(108, "older")] + assert busy_seen and busy_seen[0] + assert app.busy == set() and errors(app) == [] and app.seen + assert len(fake.snap_calls) > before + assert fake.rollback_calls == [] and fake.take_calls == [] + + +async def test_D_targets_cursor_row_first_snapshot_for_ct(fake): + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot, down=1) + await pilot.press("D") + await settle(app, pilot) + assert "newer" in text_of(app.screen) + await press_button(app, pilot, "#yes") + assert fake.delete_calls == [(100, "newer")] + + +async def test_D_failure_notifies_error_and_releases_busy(fake): + fake.snap_error_op = cli_error("Error: snapshot 'older' does not exist") + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, "D") + await press_button(app, pilot, "#yes") + assert any("does not exist" in m for m in errors(app)) + assert app.busy == set() and app.is_running + + +# --- no snapshots / busy ---------------------------------------------------- + + +@pytest.mark.parametrize("key", ["R", "D"]) +async def test_R_D_without_snapshots_give_info_notify_and_no_dialog(fake, key): + # R/D with no snapshot -> info notify, no dialog + fake.snap_list = [] + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await pilot.press(key) + await settle(app, pilot) + assert isinstance(app.screen, appmod.InfoScreen) + assert [sev for _, sev in app.seen if sev != "information"] == [] + assert app.seen + assert fake.rollback_calls == [] and fake.delete_calls == [] + + +@pytest.mark.parametrize( + "key,calls", [("R", "rollback_calls"), ("D", "delete_calls"), ("p", "take_calls")] +) +async def test_snapshot_keys_while_guest_busy_warn_and_make_no_second_call( + fake, key, calls +): + # busy guest -> warning, no call (busy held by a running take_snapshot) + started, release = threading.Event(), threading.Event() + fake.snap_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + try: + await open_snapshot_dialog(app, pilot) + app.screen.query_one("#snapname", Input).value = "first" + app.screen.query_one("#yes", Button).press() + await wait_started(pilot, started) + assert ( + fake.take_calls == [(108, "first", "", False)] + or len(fake.take_calls) == 1 + ) + app.screen.query_one("#snapshots", DataTable).move_cursor(row=0) + await pilot.press(key) + await pilot.pause() + if isinstance(app.screen, (appmod.SnapshotScreen, ConfirmScreen)): + app.screen.query_one( + "#yes", Button + ).press() # refusal may come at "yes" + await pilot.pause() + assert any(sev == "warning" for _, sev in app.seen) + assert fake.rollback_calls == [] and fake.delete_calls == [] + assert len(fake.take_calls) == 1 + finally: + release.set() + await settle(app, pilot) + assert ( + fake.rollback_calls == [] + and fake.delete_calls == [] + and len(fake.take_calls) == 1 + ) + + +# --- #8 review findings --------------------------------------------------------------- + + +async def test_rollback_start_flag_uses_status_at_rollback_time(fake): + # opened while running, stopped before "yes": do not start it again + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, "R") + fake.live_status = {108: "stopped"} + await press_button(app, pilot, "#yes") + assert fake.rollback_calls == [(108, "older", False)] + + +async def test_rollback_aborts_when_status_cannot_be_read(fake, monkeypatch): + # an unknown status must not silently leave a running guest stopped + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_confirm_on_row(app, pilot, "R") + + def broken(guest): + raise cli.CliError("Error: status unavailable") + + monkeypatch.setattr(cli, "status", broken) + await press_button(app, pilot, "#yes") + assert fake.rollback_calls == [] + assert errors(app) and app.busy == set() + + +async def test_snapshot_description_starting_with_dash_is_refused(fake): + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await pilot.press("p") + await settle(app, pilot) + app.screen.query_one("#snapdesc", Input).value = "-y" + await press_button(app, pilot, "#yes") + assert fake.take_calls == [] and errors(app) + assert isinstance(app.screen, appmod.SnapshotScreen) + + +async def test_info_screen_scrolls_on_short_terminals(fake): + from textual.containers import VerticalScroll + + app = make_app() + async with app.run_test(size=(100, 24)) as pilot: + await open_info(app, pilot) + assert isinstance(app.screen.query_one("#info_body"), VerticalScroll) diff --git a/tests/test_cli.py b/tests/test_cli.py index 6fff074..61da543 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1003,3 +1003,248 @@ def test_not_backed_up_returns_vmids(fake): def test_not_backed_up_empty(fake): fake.result = ("[]", "", 0) assert cli.not_backed_up() == set() + + +# --- snapshots (issue #8); spec: docs/plans/guest-tui-mvp.md ("Snapshots in the info screen") --- +# Sections below name the spec bullet they cover. + +GOOD_NAMES = [ + "a", + "snap1", + "tui_20251102_1430", + "A-b_c", + "x" * 40, + "snapshot", + "currentx", + "curren", +] +BAD_NAMES = [ + "", + "current", + "Current", + "CURRENT", + "1snap", + "_snap", + "-snap", + "my snap", + "a,b", + ";rm", + "a;rm -rf", + "x" * 41, + "snäp", + "snap\n", + "snap ", +] + + +@pytest.mark.parametrize("name", GOOD_NAMES) +def test_valid_snapname_accepts_pattern( + name, +): # valid_snapname: [A-Za-z][A-Za-z0-9_-]{0,39} + assert cli.valid_snapname(name) == name + + +@pytest.mark.parametrize("name", BAD_NAMES) +def test_valid_snapname_rejects( + name, +): # valid_snapname: ValueError outside pattern / "current" + with pytest.raises(ValueError): + cli.valid_snapname(name) + + +@given(st.text()) +def test_delete_snapshot_with_rejected_name_raises_and_never_runs(name): + # delete_snapshot: name validated first; no subprocess call for an invalid name + try: + cli.valid_snapname(name) + except ValueError: + pass + else: + return # accepted names are covered by the argv tests + with pytest.MonkeyPatch.context() as m: + run = FakeRun(m) + with pytest.raises(ValueError): + cli.delete_snapshot(CT, name) + with pytest.raises(ValueError): + cli.delete_snapshot(VM, name) + assert run.calls == [] + + +SNAPS_RAW = [ + {"name": "old", "description": "first", "snaptime": 1_000, "parent": None}, + {"name": "undated", "description": "", "snaptime": None, "parent": "old"}, + { + "name": "current", + "description": "You are here!", + "snaptime": None, + "parent": "new", + }, + {"name": "new", "description": "second", "snaptime": 3_000, "parent": "old"}, + {"name": "mid", "description": "", "snaptime": 2_000, "parent": "old"}, +] + + +def test_snapshots_drops_current_sorts_newest_first_null_last(fake): # snapshots() + fake.result = (json.dumps(SNAPS_RAW), "", 0) + assert [s["name"] for s in cli.snapshots(CT)] == ["new", "mid", "old", "undated"] + + +def test_snapshots_drops_current_case_insensitively( + fake, +): # snapshots(): current is not a snapshot + fake.result = ( + json.dumps([{"name": "Current", "snaptime": None}, SNAPS_RAW[0]]), + "", + 0, + ) + assert [s["name"] for s in cli.snapshots(CT)] == ["old"] + + +def test_snapshots_only_current_gives_empty_list( + fake, +): # snapshots(): "No snapshots" case + fake.result = (json.dumps([SNAPS_RAW[2]]), "", 0) + assert cli.snapshots(CT) == [] + + +@pytest.mark.parametrize( + "guest,kind", + [(CT, "lxc"), (VM, "qemu")], +) +def test_snapshots_argv_uses_qemu_or_lxc( + fake, guest, kind +): # CLI facts: get guest snapshots + fake.result = ("[]", "", 0) + cli.snapshots(guest) + assert fake.argv == [ + "cv4pve-cli", "get", "guest", "snapshots", "mainframe", kind, str(guest["vmid"]), "-o", "json", + ] # fmt: skip + + +def test_snapshots_failure_raises_clierror( + monkeypatch, +): # snapshots(): failure surfaces as CliError + FakeRun(monkeypatch, "Error: boom", returncode=1) + with pytest.raises(cli.CliError): + cli.snapshots(CT) + + +def test_take_snapshot_ct_argv_has_no_vmstate_and_wait_last(fake): # take_snapshot + cli.take_snapshot(CT, "snap1", "hello") + assert fake.argv == [ + "cv4pve-cli", "create", "ct", "snapshot", "mainframe", "100", "snap1", "hello", "--wait", + ] # fmt: skip + + +def test_take_snapshot_vm_without_vmstate(fake): # take_snapshot + cli.take_snapshot(VM, "snap1", "hello") + assert fake.argv == [ + "cv4pve-cli", "create", "vm", "snapshot", "mainframe", "108", "snap1", "hello", "--wait", + ] # fmt: skip + + +def test_take_snapshot_vm_with_vmstate_before_wait( + fake, +): # take_snapshot: --vmstate 1 (VM only) + cli.take_snapshot(VM, "snap1", "hello", vmstate=True) + assert fake.argv == [ + "cv4pve-cli", "create", "vm", "snapshot", "mainframe", "108", "snap1", "hello", + "--vmstate", "1", "--wait", + ] # fmt: skip + + +def test_take_snapshot_ct_ignores_vmstate(fake): # take_snapshot: vmstate only for a VM + cli.take_snapshot(CT, "snap1", "hello", vmstate=True) + assert "--vmstate" not in fake.argv and fake.argv[-1] == "--wait" + + +def test_take_snapshot_default_description( + fake, +): # take_snapshot: description or "cv4pve-tui" + cli.take_snapshot(CT, "snap1") + assert fake.argv[7] == "cv4pve-tui" + + +def test_take_snapshot_empty_description_falls_back( + fake, +): # take_snapshot: `description or "cv4pve-tui"` + cli.take_snapshot(VM, "snap1", "") + assert fake.argv[7] == "cv4pve-tui" + + +@pytest.mark.parametrize("name", ["", "current", "1x", "a b", ";rm"]) +def test_take_snapshot_invalid_name_no_call( + fake, name +): # take_snapshot: name validated first + with pytest.raises(ValueError): + cli.take_snapshot(VM, name, "d") + assert fake.calls == [] + + +@pytest.mark.parametrize("guest,kind", [(CT, "ct"), (VM, "vm")]) +def test_rollback_argv_without_start(fake, guest, kind): # rollback_snapshot + cli.rollback_snapshot(guest, "snap1") + assert fake.argv == [ + "cv4pve-cli", "do", "rollback", kind, "mainframe", str(guest["vmid"]), "snap1", "-y", "--wait", + ] # fmt: skip + + +@pytest.mark.parametrize("guest,kind", [(CT, "ct"), (VM, "vm")]) +def test_rollback_argv_with_start( + fake, guest, kind +): # rollback_snapshot: [--start 1] -y --wait + cli.rollback_snapshot(guest, "snap1", start=True) + assert fake.argv == [ + "cv4pve-cli", "do", "rollback", kind, "mainframe", str(guest["vmid"]), "snap1", + "--start", "1", "-y", "--wait", + ] # fmt: skip + + +@pytest.mark.parametrize("name", ["", "current", "Current", "9", "a b", "a,b"]) +def test_rollback_invalid_name_no_call(fake, name): # rollback_snapshot: name validated + with pytest.raises(ValueError): + cli.rollback_snapshot(CT, name, start=True) + assert fake.calls == [] + + +@pytest.mark.parametrize("guest,kind", [(CT, "ct"), (VM, "vm")]) +def test_delete_snapshot_exact_argv(fake, guest, kind): # delete_snapshot + cli.delete_snapshot(guest, "snap1") + assert fake.argv == [ + "cv4pve-cli", "delete", kind, "snapshot", "mainframe", str(guest["vmid"]), "snap1", "-y", "--wait", + ] # fmt: skip + + +@given( + name=st.from_regex(r"[A-Za-z][A-Za-z0-9_-]{0,39}", fullmatch=True).filter( + lambda n: n.lower() != "current" + ), + guest=st.sampled_from([CT, VM]), +) +def test_delete_never_produces_guest_destroying_form(name, guest): + # DANGER: `delete -y` destroys the guest; "snapshot" must sit at index 3 + kind = {"lxc": "ct", "qemu": "vm"}[guest["type"]] + with pytest.MonkeyPatch.context() as m: + run = FakeRun(m) + cli.delete_snapshot(guest, name) + argv = run.argv + assert argv[1:3] == ["delete", kind] + assert argv[3] == "snapshot" # never the node directly after the kind + assert ( + argv[4] == guest["node"] and argv[5] == str(guest["vmid"]) and argv[6] == name + ) + assert argv[1:3] != ["delete", kind] or argv[3] != guest["node"] + + +def test_delete_with_snapshot_like_name_still_has_snapshot_at_index_3(fake): + # a valid name equal to the subcommand word must not shift the positions + cli.delete_snapshot(CT, "snapshot") + assert fake.argv[3:7] == ["snapshot", "mainframe", "100", "snapshot"] + + +def test_take_snapshot_refuses_description_that_looks_like_an_option(fake): + with pytest.raises(ValueError): + cli.take_snapshot( + {"vmid": 104, "node": "mainframe", "type": "lxc"}, "snap1", "-y" + ) + assert fake.calls == []