Add snapshots to the info screen: take, roll back, delete
A "Snapshots" section on the info screen lists the snapshots, newest first, without Proxmox's "current" pseudo entry. Keys: - p takes a snapshot. The name defaults to tui_YYYYMMDD_HHMM, the description is optional, and VMs can include RAM. - R rolls back to the snapshot under the cursor. The confirm names the snapshot and its date. A running guest is started again; its status is read right before the rollback, and if it cannot be read the rollback does not run. - D deletes the snapshot under the cursor, after a confirm. `delete ct|vm <node> <vmid> -y` without the snapshot subcommand destroys the whole guest. delete_snapshot therefore validates the name first (letter first, then [A-Za-z0-9_-], not "current"), writes "snapshot" into the argv as a literal, and checks the argv shape before running it. A property test covers rejected names. The info screen now scrolls, because four sections do not fit short terminals. A description starting with "-" is refused, since it is a positional argument and could be read as an option. Live test on CT 104: take, roll back (started again) and delete. The snapshot list, status, config and IPv4/IPv6 match the baseline. Closes #8 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit was merged in pull request #21.
This commit is contained in:
1 parent
520de16cc6
commit
ec35a4c996
6 files changed
+1034
-13
No files matched your search
@@ -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) |
|
||||
|
||||
@@ -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
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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 == []
|
||||
Reference in new issue
Block a user