Snapshots in the info screen: take, roll back, delete #21

Merged
Duckboot merged 1 commits from feat/snapshots-8 into main 2026-11-02 23:28:19 +01:00
6 changed files with 1034 additions and 13 deletions

No files matched your search

+1
View File
@@ -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@<node> pct enter <vmid>`) |
| `g` | refresh |
| `c` | switch cv4pve-cli context (cluster) |
+66
View File
@@ -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 <node> <qemu|lxc> <vmid> -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 <node> <vmid> <snapname> <descr>` (+ `--vmstate 1` for VM RAM),
no `--yes`; with `--wait`.
- `do rollback ct|vm <node> <vmid> <snapname>` needs `-y`; API param `start` (bool)
starts the guest after a successful rollback.
- `delete ct|vm snapshot <node> <vmid> <snapname>` needs `-y`.
DANGER: `delete ct|vm <node> <vmid> -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 <kind> snapshot <node> <vmid> <name> <description or "cv4pve-tui">
[--vmstate 1 (VM only)] --wait`; name validated first.
- `rollback_snapshot(guest, name, start=False)` ->
`do rollback <kind> <node> <vmid> <name> [--start 1] -y --wait`; name validated.
- `delete_snapshot(guest, name)` ->
`delete <kind> snapshot <node> <vmid> <name> -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: <summary>"), the rest of the
screen still works. Empty list -> `#snap_note` "No snapshots".
- `p`: `SnapshotScreen` dialog (DialogScreen style, title "Take snapshot: <name>
(<vmid>)"): `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.
+169 -13
View File
@@ -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")
@@ -429,6 +482,12 @@ class InfoScreen(ModalScreen):
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:
+58
View File
@@ -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 <kind> <node> <vmid> -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)
+495
View File
@@ -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: <summary>" 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: <name> (<vmid>)", 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)
+245
View File
@@ -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 <kind> <node> <vmid> -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 == []