Duckboot/cv4pve-cli-tui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restore a guest from a backup in the backup list #32

Merged
Duckboot merged 1 commits from feat/restore-15 into main 2026-11-03 01:40:12 +01:00
Conversation 0 Commits 1 Files Changed 6
+862 -3
6 changed files with 862 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+1
View File
@@ -47,6 +47,7 @@ uv run cv4pve-tui
| `Enter` then `o` | in info: edit autostart, startup order (order, up and down delay) and tags (one confirmed `set config` call; empty fields remove the setting) |
| `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) |
| `Enter` then `f` | in info: fstrim in a running VM with the guest agent enabled (confirms; containers: use `pct fstrim` on the node) |
| `Enter` then `r` | in info: restore the backup under the cursor, as a new VMID (default, the original stays untouched; a copy keeps the original's network settings, so leave Start off until they are changed) or overwriting the guest itself (stopped guests only; type the VMID, then confirm) |
| `t` | shell in a running LXC container (`ssh -t root@<node> pct enter <vmid>`) |
| `l` | tasks: recent cluster tasks (start, duration, node, type, VMID, user, status; `!` marks failures); `Enter` shows the log of the task under the cursor, `g` reloads. In info: `l` = only this guest's tasks |
| `U` | package updates: pending updates per online node (node, package, current, new, origin, priority), read-only; `g` reloads. The list is as of the node's last `apt update`. The frame's bottom border shows the count |
docs/plans/guest-tui-mvp.md
+60
View File
@@ -735,3 +735,63 @@ CLI facts:
- yes -> busy guard, worker: `cli.agent_ping` then `cli.fstrim`; done notify with the
trimmed total in GiB/MiB and the number of paths (errors per path mentioned);
CliError -> error notify (ping failure says the agent did not answer).
---
# Restore a guest from a backup (issue #15)
## Decision
From the info screen's backup list, key `r` ("Restore backup"; `R` stays snapshot
rollback). Two targets: a new VMID (default, original untouched) or overwriting the
guest itself (destructive). Not run live: a live test would create a guest that then
needs deleting, which the issue reserves for the user's approval. Tested with mocks.
CLI facts (cv4pve-cli 2.4.0, read-only probes):
- `do restore ct|vm <node> <vmid> <volume> [--key value] -y --wait` - "creates/overwrites
the target vmid". API params: `storage` (target storage for the disks), `force`
(bool; required to overwrite an existing guest), `start` (bool), `unique` (VM: new
MACs).
- `get cluster nextid -o json` -> a JSON number, e.g. `105`.
- Backup entries carry `volid` (`truenas-backup:backup/vzdump-lxc-104-...tar.zst`),
`subtype` (`lxc`|`qemu`), `vmid`.
- Node storages: `content` lists `rootdir` (CT disks) and/or `images` (VM disks).
### `cli.py`
- `next_id() -> int`.
- `valid_volid(volid) -> str` - ValueError unless it matches
`^[A-Za-z0-9][A-Za-z0-9_.-]*:backup/vzdump-(lxc|qemu)-\d+-\S+$` (positional argument).
- `restore_storages(guest) -> list[str]` - enabled+active storages on the guest's node
whose `content` has `rootdir` (CT) or `images` (VM), API order.
- `restore(guest, volid, vmid, storage, force=False, start=False)` ->
`do restore <kind> <node> <vmid> <volid> --storage <storage> [--force 1] [--start 1]
-y --wait`; kind from the backup's guest kind (the guest dict's `type`); `vmid` a
positive int; volid validated; `force=True` only allowed when `vmid ==
guest["vmid"]` (ValueError otherwise - never overwrite some other guest).
### `app.py` (UI contract)
- `r` on the info screen: the backup under the cursor of `#backups` (the visible rows;
hidden earlier-guest backups are not restorable unless shown with `a`). No backup ->
info notify.
- `RestoreScreen` (DialogScreen, title "Restore <date> backup of <name> (<vmid>)"):
`RadioSet#target` with `RadioButton#new` (default, "New guest") and
`RadioButton#overwrite` ("Overwrite <name> (<vmid>)"); `Input#newid` prefilled with
`cli.next_id()` (used for "new"); `Select#storage` (`restore_storages`, preselected
to the storage of the guest's rootfs/first disk if listed); `Checkbox#start` (off);
`Input#confirm` (typed VMID, only needed for overwrite); `#apply`/`#cancel`; escape =
cancel. A note says a new copy keeps the original's network settings (same static
IP), so leave Start off until it is changed.
- Apply, new: `newid` must be a positive int not used by any guest in the current
listing (app.guests) -> else error notify, no call. Then `ConfirmScreen` titled
"Restore" summarising backup, target VMID, storage, start; yes -> busy guard on the
source guest, worker `cli.restore(guest, volid, newid, storage, force=False,
start=...)`, notify, main list reloads.
- Apply, overwrite: the guest must be stopped (live `cli.status` at apply time) -> else
error notify; `Input#confirm` must equal the VMID -> else error notify; then
`ConfirmScreen` titled "Overwrite" saying the guest's current disks and config are
replaced and cannot be undone; yes -> `cli.restore(guest, volid, guest vmid, storage,
force=True, start=...)`.
- All CliErrors -> error notify; busy released; markup-safe texts.
src/cv4pve_tui/app.py
+183 -3
View File
@@ -10,6 +10,7 @@ from textual import work
from textual.app import App, ComposeResult, SuspendNotSupported
from textual.binding import Binding
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.content import Content
from textual.screen import ModalScreen
from textual.widgets import (
Button,
@@ -21,6 +22,8 @@ from textual.widgets import (
Label,
Log,
OptionList,
RadioButton,
RadioSet,
Select,
Static,
)
@@ -702,6 +705,83 @@ class UpdatesScreen(ModalScreen):
self.query_one("#updates_note", Static).update("\n".join(lines))
class RestoreScreen(DialogScreen):
"""Restore one backup; dismisses with {overwrite, vmid, storage, start}, or None."""
BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom
def __init__(
self, guest: dict, backup: dict, storages: list[str], chosen: str, newid: int
) -> None:
super().__init__()
self.guest, self.backup = guest, backup
self.storages, self.chosen, self.newid = storages, chosen, newid
def compose(self) -> ComposeResult:
who = f"{self.guest['name']} ({self.guest['vmid']})"
with self.box(f"Restore {_backup_label(self.backup, self.guest)}"):
with RadioSet(id="target"):
yield RadioButton(Content("New guest"), value=True, id="new")
yield RadioButton(Content(f"Overwrite {who}"), id="overwrite")
with Horizontal(classes="row"):
yield Label("New VMID")
yield Input(str(self.newid), id="newid")
with Horizontal(classes="row"):
yield Label("Storage")
yield Select(
[(s, s) for s in self.storages],
value=self.chosen,
allow_blank=False,
id="storage",
)
yield Checkbox("Start after restore", id="start")
with Horizontal(classes="row"):
yield Label("Confirm")
yield Input(placeholder="type the VMID to overwrite", id="confirm")
yield Static(
"A new copy keeps the original's MAC address and network settings "
"(same static IP): leave Start off until they are changed.",
markup=False,
)
with Horizontal(classes="buttons"):
yield Button("Apply", id="apply", variant="primary")
yield Button("Cancel", id="cancel")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "cancel":
self.dismiss(None)
return
overwrite = self.query_one("#overwrite", RadioButton).value
if overwrite:
vmid = self.guest["vmid"]
if self.query_one("#confirm", Input).value != str(vmid):
self.app.notify(
f"Type {vmid} to confirm the overwrite", severity="error"
)
return
else:
text = self.query_one("#newid", Input).value.strip()
if (
not text.isdecimal()
or int(text) <= 0
or str(int(text)) in self.app.guests
):
self.app.notify(
"The new VMID must be a positive number not used by a guest",
severity="error",
)
return
vmid = int(text)
self.dismiss(
{
"overwrite": overwrite,
"vmid": vmid,
"storage": self.query_one("#storage", Select).value,
"start": self.query_one("#start", Checkbox).value,
}
)
class InfoScreen(ModalScreen):
"""Backups, backup job and network of one guest (read-only), plus backup now."""
@@ -719,6 +799,7 @@ class InfoScreen(ModalScreen):
("D", "delete_snapshot", "Delete snapshot"),
("l", "tasks", "Tasks"),
("f", "fstrim", "fstrim"),
("r", "restore", "Restore backup"),
]
DEFAULT_CSS = """
InfoScreen DataTable { height: auto; max-height: 12; }
@@ -902,11 +983,13 @@ class InfoScreen(ModalScreen):
note or ("" if snaps else "No snapshots")
)
def _visible_backups(self) -> list[dict]:
name = self.guest["name"]
return [b for b in self.backups if self.show_all or not _earlier_guest(b, name)]
def _show_backups(self) -> None:
name = self.guest["name"]
shown = [
b for b in self.backups if self.show_all or not _earlier_guest(b, name)
]
shown = self._visible_backups()
table = self.query_one("#backups", DataTable).clear()
for b in shown:
_add_plain(
@@ -1065,6 +1148,95 @@ class InfoScreen(ModalScreen):
),
)
def action_restore(self) -> None:
if not self.config: # the storage preselection needs the loaded config
self.app.notify("Info is still loading; try again in a moment")
return
shown = self._visible_backups()
row = self.query_one("#backups", DataTable).cursor_row
if not 0 <= row < len(shown):
self.app.notify("No backups")
elif not self._refuse_if_busy():
self._open_restore(shown[row])
@work(thread=True)
def _open_restore(self, backup: dict) -> None:
try:
newid = cli.next_id()
storages = cli.restore_storages(self.guest)
except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app
self.app.call_from_thread(self.app.notify, _message(exc), severity="error")
return
if not storages:
self.app.call_from_thread(
self.app.notify,
"No storage can hold restored disks",
severity="warning",
)
return
disks = cli.disks(self.config, cli.kind(self.guest))
key = "rootfs" if "rootfs" in disks else next(iter(disks), None)
chosen = str(self.config.get(key, "")).split(":")[0]
if chosen not in storages:
chosen = storages[0]
self.app.call_from_thread(
self.app.push_screen,
RestoreScreen(self.guest, backup, storages, chosen, newid),
lambda values: self._restore_chosen(backup, values) if values else None,
)
def _restore_chosen(self, backup: dict, values: dict) -> None:
guest, vmid, volid = self.guest, values["vmid"], backup["volid"]
who = f"{guest['name']} ({guest['vmid']})"
what = (
f"the {_backup_label(backup, self.guest)} onto storage {values['storage']}"
)
start = bool(values["start"])
text = f"{'Start' if start else 'Do not start'} the guest after the restore."
def job() -> None:
if values["overwrite"] and cli.status(guest) != "stopped":
raise cli.CliError(f"{who} must be stopped to be overwritten")
cli.restore(
guest, volid, vmid, values["storage"], values["overwrite"], start
)
def confirmed(ok: bool) -> None:
if ok:
label = "Overwrite" if values["overwrite"] else "Restore"
self.app.start_job(guest, label, job)
if values["overwrite"]:
self._confirm_overwrite(
ConfirmScreen(
f"Overwrite {who} with {what}? Its current disks and configuration "
f"are replaced and this cannot be undone. {text}",
"Overwrite",
),
confirmed,
)
else:
self.app.push_screen(
ConfirmScreen(f"Restore {what} as new guest {vmid}? {text}", "Restore"),
confirmed,
)
@work(thread=True)
def _confirm_overwrite(self, screen: ConfirmScreen, confirmed) -> None:
try: # live status: the guest table's status may be stale
stopped = cli.status(self.guest) == "stopped"
except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app
self.app.call_from_thread(self.app.notify, _message(exc), severity="error")
return
if stopped:
self.app.call_from_thread(self.app.push_screen, screen, confirmed)
else:
self.app.call_from_thread(
self.app.notify,
"Stop the guest before overwriting it from a backup",
severity="error",
)
def action_fstrim(self) -> None:
who = f"{self.guest['name']} ({self.guest['vmid']})"
if not self.config: # running/agent are only known once the screen has loaded
@@ -1133,6 +1305,14 @@ def _ip_changed(old: str | None, new: str | None) -> bool:
return any(before.get(k) != after.get(k) for k in ("ip", "gw", "ip6", "gw6"))
def _backup_label(backup: dict, guest: dict) -> str:
"""'<date> backup of <name> (<vmid>)', naming a deleted earlier guest when it is one."""
label = f"{_when(backup.get('ctime'))} backup of {guest['name']} ({guest['vmid']})"
if _earlier_guest(backup, guest["name"]):
label += f" - earlier guest: {backup.get('notes')}"
return label
def _earlier_guest(backup: dict, name: str) -> bool:
"""VMIDs are reused: notes (the job's {{guestname}}) naming another guest mean a deleted one."""
notes = backup.get("notes") or ""
src/cv4pve_tui/cli.py
+61
View File
@@ -180,6 +180,67 @@ def backup_storages(guest: dict) -> list[str]:
]
def next_id() -> int:
return int(json.loads(run("get", "cluster", "nextid", "-o", "json")))
_STORAGE = r"[A-Za-z0-9][A-Za-z0-9_.-]*"
_VOLID = re.compile(rf"{_STORAGE}:backup/vzdump-(lxc|qemu)-[0-9]+-\S+")
def valid_volid(volid: str) -> str:
"""The volid is a positional argument: refuse anything that is not a vzdump volume."""
if not _VOLID.fullmatch(volid):
raise ValueError(f"not a backup volume id: {volid!r}")
return volid
def restore_storages(guest: dict) -> list[str]:
"""Enabled, active storages on the guest's node that hold its disks, in API order."""
content = "images" if guest["type"] == "qemu" else "rootdir"
items = json.loads(run("get", "node", "storages", guest["node"], "-o", "json"))
return [
s["storage"]
for s in items
if content in (s.get("content") or "").split(",")
and s.get("enabled")
and s.get("active")
]
def restore(
guest: dict,
volid: str,
vmid: int,
storage: str,
force: bool = False,
start: bool = False,
) -> None:
"""Restore `volid` as `vmid`; `force` (overwrite) only for the guest's own VMID."""
_positive(vmid, "vmid")
if force and vmid != guest["vmid"]:
raise ValueError(
"refusing to overwrite a guest other than the one restored from"
)
valid_volid(volid)
backup_type = re.search(r":backup/vzdump-(lxc|qemu)-", volid).group(1)
if backup_type != _api_type(guest): # a reused VMID can carry the other guest type
raise ValueError(
f"a {backup_type} backup cannot be restored as this {kind(guest)}"
)
if not re.fullmatch(_STORAGE, storage):
raise ValueError(f"invalid storage name: {storage!r}")
opts = ["--storage", storage]
if force:
opts += ["--force", "1"]
if start:
opts += ["--start", "1"]
run(
"do", "restore", kind(guest), guest["node"], str(vmid), volid, *opts,
"-y", "--wait", timeout=3600,
) # fmt: skip
def backup_jobs() -> list[dict]:
jobs = json.loads(run("get", "cluster", "backup-jobs", "-o", "json"))
for job in jobs: # the CLI returns prune-backups as a JSON string
tests/test_app.py
+411
View File
@@ -19,6 +19,7 @@ from textual.widgets import (
DataTable,
Input,
OptionList,
RadioButton,
Select,
Static,
)
@@ -99,6 +100,10 @@ class Fake:
None,
)
self.fstrim_paths = [{"path": "/", "trimmed": 2**30, "minimum": 0}]
# issue #15 (restore from a backup)
self.nextid, self.nextid_error = 105, None
self.restore_store = ["local-lvm", "zfs-pool"]
self.restore_calls, self.restore_error, self.restore_hook = [], None, None
def list_guests(self):
self.list_calls += 1
@@ -284,6 +289,21 @@ class Fake:
raise self.fstrim_error
return [dict(p) for p in self.fstrim_paths]
def next_id(self):
if self.nextid_error:
raise self.nextid_error
return self.nextid
def restore_storages(self, guest):
return list(self.restore_store)
def restore(self, guest, volid, vmid, storage, force=False, start=False):
self.restore_calls.append((guest["vmid"], volid, vmid, storage, force, start))
if self.restore_hook:
self.restore_hook()
if self.restore_error:
raise self.restore_error
def task_log(self, upid):
self.log_calls.append(upid)
if self.log_error:
@@ -351,6 +371,9 @@ def fake(monkeypatch):
"package_updates",
"agent_ping",
"fstrim",
"next_id",
"restore_storages",
"restore",
):
monkeypatch.setattr(cli, name, getattr(f, name), raising=False)
for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
@@ -4461,3 +4484,391 @@ async def test_fstrim_before_the_info_screen_loaded_says_still_loading(
finally:
release.set()
await settle(app, pilot)
# --- #15 restore a guest from a backup ----------------------------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Restore a guest from a backup (issue #15)" > app.py)
NEW_VOLID = BACKUPS[0]["volid"]
OLD_VOLID = BACKUPS[1]["volid"]
def vm_disk_on(fake, storage="zfs-pool"):
# the guest's first disk lives on `storage` (config value "<storage>:<volume>,size=..")
fake.net_configs[108] = {"scsi0": f"{storage}:vm-108-disk-0,size=10G"}
async def open_restore(app, pilot, down=0, row=0):
await open_info(app, pilot, down=down)
app.screen.query_one("#backups", DataTable).move_cursor(row=row)
await pilot.press("r")
await settle(app, pilot)
assert isinstance(app.screen, appmod.RestoreScreen)
def set_value(app, selector, value):
app.screen.query_one(selector).value = value
async def apply_new(app, pilot, newid=None, start=False):
if newid is not None:
set_value(app, "#newid", newid)
if start:
set_value(app, "#start", True)
await press_button(app, pilot, "#apply")
async def apply_overwrite(app, pilot, confirm="108", start=False):
app.screen.query_one("#overwrite", RadioButton).value = True
set_value(app, "#confirm", confirm)
if start:
set_value(app, "#start", True)
await press_button(app, pilot, "#apply")
def title_of(app):
return str(app.screen.query_one("#dialog").border_title)
async def test_r_on_backup_opens_restore_screen_with_defaults(fake):
# spec: title names date and guest; #new default; #newid from next_id; #storage preselected
vm_disk_on(fake, "zfs-pool")
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
title = title_of(app)
assert "Restore" in title and "docker" in title and "108" in title
assert local_date(T_NEW)[:10] in title
assert app.screen.query_one("#new", RadioButton).value is True
assert app.screen.query_one("#overwrite", RadioButton).value is False
assert app.screen.query_one("#newid", Input).value == "105"
assert app.screen.query_one("#storage", Select).value == "zfs-pool"
assert app.screen.query_one("#start", Checkbox).value is False
assert fake.restore_calls == []
async def test_restore_storage_falls_back_to_a_listed_one_when_disk_storage_unlisted(
fake,
):
# spec: preselected "if listed"; otherwise the select must still hold a listed storage
vm_disk_on(fake, "not-offered")
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
assert app.screen.query_one("#storage", Select).value in fake.restore_store
async def test_restore_ct_preselects_rootfs_storage(fake):
# spec: "storage of the guest's rootfs/first disk"
fake.net_configs[100] = {"rootfs": "zfs-pool:subvol-100-disk-0,size=8G"}
fake.disk_map = {"rootfs": "8G"}
fake.backup_list = [
{**BACKUPS[0], "volid": "truenas-backup:backup/vzdump-lxc-100-new.tar.zst",
"notes": "backrest (manual)"} # the CT's own backup (other notes = earlier guest, hidden)
] # fmt: skip
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot, down=1)
assert "backrest" in title_of(app)
assert app.screen.query_one("#storage", Select).value == "zfs-pool"
async def test_r_without_backups_notifies_info_and_opens_nothing(fake):
# spec: no backup -> info notify
fake.backup_list = []
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot)
await pilot.press("r")
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
assert any(sev == "information" for _, sev in app.seen)
assert fake.restore_calls == []
async def test_new_apply_asks_confirm_titled_restore_then_restores_busy_and_reloads(
fake,
):
# spec: apply new -> ConfirmScreen "Restore"; yes -> busy guard, restore(..., force=False, start=False)
vm_disk_on(fake, "zfs-pool")
app = make_app()
busy_seen = []
fake.restore_hook = lambda: busy_seen.append(set(app.busy))
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_new(app, pilot)
assert isinstance(app.screen, ConfirmScreen)
assert app.screen.title_text == "Restore"
text = text_of(app.screen)
assert "105" in text and "zfs-pool" in text
assert fake.restore_calls == []
before = fake.list_calls
await press_button(app, pilot, "#yes")
assert fake.restore_calls == [(108, NEW_VOLID, 105, "zfs-pool", False, False)]
assert busy_seen and busy_seen[0], (
"source guest must be busy during the restore"
)
assert app.busy == set() and errors(app) == []
assert fake.list_calls > before, "main list must reload"
async def test_new_with_edited_id_and_start_checked_passes_them(fake):
# spec: Input#newid editable; Checkbox#start -> start=True
vm_disk_on(fake, "zfs-pool")
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_new(app, pilot, newid="777", start=True)
await press_button(app, pilot, "#yes")
assert fake.restore_calls == [(108, NEW_VOLID, 777, "zfs-pool", False, True)]
async def test_new_with_another_storage_selected_passes_it(fake):
vm_disk_on(fake, "zfs-pool")
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
set_value(app, "#storage", "local-lvm")
await apply_new(app, pilot)
await press_button(app, pilot, "#yes")
assert fake.restore_calls[0][3] == "local-lvm"
@pytest.mark.parametrize("bad", ["108", "100", "abc", "0", "-5", "", "10.5", "1e3"])
async def test_new_with_unusable_id_errors_without_call(fake, bad):
# spec: newid positive int not used by any guest in app.guests (108, 100 exist)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_new(app, pilot, newid=bad)
assert errors(app)
assert fake.restore_calls == []
assert not isinstance(app.screen, ConfirmScreen)
@pytest.mark.parametrize("how", ["no", "escape"])
async def test_new_confirm_declined_makes_no_call(fake, how):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_new(app, pilot)
if how == "no":
await press_button(app, pilot, "#no")
else:
await pilot.press("escape")
await settle(app, pilot)
assert fake.restore_calls == [] and app.busy == set()
@pytest.mark.parametrize("how", ["cancel", "escape"])
async def test_restore_screen_cancel_or_escape_makes_no_call(fake, how):
# spec: #cancel / escape = cancel
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
if how == "cancel":
await press_button(app, pilot, "#cancel")
else:
await pilot.press("escape")
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
assert fake.restore_calls == [] and errors(app) == []
async def test_overwrite_on_running_guest_errors_without_call(fake):
# spec: overwrite needs a stopped guest (live cli.status at apply time); listing says stopped, live says running
fake.guests = [dict(VM, status="stopped"), CT]
fake.live_status = {108: "running"}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_overwrite(app, pilot)
assert errors(app)
assert fake.restore_calls == []
assert not isinstance(app.screen, ConfirmScreen)
@pytest.mark.parametrize("wrong", ["", "100", "docker", "1080", " 108x"])
async def test_overwrite_with_wrong_typed_confirm_errors_without_call(fake, wrong):
# spec: Input#confirm must equal the VMID
fake.live_status = {108: "stopped"}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_overwrite(app, pilot, confirm=wrong)
assert errors(app)
assert fake.restore_calls == []
assert not isinstance(app.screen, ConfirmScreen)
async def test_overwrite_stopped_guest_right_confirm_asks_overwrite_then_restores_with_force(
fake,
):
# spec: ConfirmScreen "Overwrite" (disks and config replaced, cannot be undone); yes -> force=True on the guest's own vmid
vm_disk_on(fake, "zfs-pool")
fake.live_status = {108: "stopped"}
app = make_app()
busy_seen = []
fake.restore_hook = lambda: busy_seen.append(set(app.busy))
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_overwrite(app, pilot, confirm="108", start=True)
assert isinstance(app.screen, ConfirmScreen)
assert app.screen.title_text == "Overwrite"
text = text_of(app.screen).lower()
assert "docker" in text and "108" in text and "undone" in text
assert fake.restore_calls == []
before = fake.list_calls
await press_button(app, pilot, "#yes")
assert fake.restore_calls == [(108, NEW_VOLID, 108, "zfs-pool", True, True)]
assert busy_seen and busy_seen[0]
assert app.busy == set() and errors(app) == []
assert fake.list_calls > before
async def test_overwrite_ignores_a_stale_newid_value(fake):
# spec: #newid is "used for new" only; overwrite targets the guest's own vmid
fake.live_status = {108: "stopped"}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
set_value(app, "#newid", "999")
await apply_overwrite(app, pilot)
await press_button(app, pilot, "#yes")
assert [(c[2], c[4]) for c in fake.restore_calls] == [(108, True)]
@pytest.mark.parametrize("how", ["no", "escape"])
async def test_overwrite_confirm_declined_makes_no_call(fake, how):
fake.live_status = {108: "stopped"}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
await apply_overwrite(app, pilot)
if how == "no":
await press_button(app, pilot, "#no")
else:
await pilot.press("escape")
await settle(app, pilot)
assert fake.restore_calls == [] and app.busy == set()
async def test_restore_uses_the_backup_under_the_cursor(fake):
# spec: the backup under the cursor of #backups (2nd row = the old one)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot, row=1)
assert local_date(T_OLD)[:10] in title_of(app)
await apply_new(app, pilot)
await press_button(app, pilot, "#yes")
assert [c[1] for c in fake.restore_calls] == [OLD_VOLID]
async def test_hidden_earlier_guest_backup_is_not_restorable_by_default(fake):
# spec: hidden earlier-guest backups are not restorable unless shown with `a`
fake.backup_list = [*BACKUPS, EARLIER]
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot, row=99) # cursor clamps to the last visible row
await apply_new(app, pilot)
await press_button(app, pilot, "#yes")
assert [c[1] for c in fake.restore_calls] == [OLD_VOLID]
async def test_earlier_guest_backup_is_restorable_after_a(fake):
fake.backup_list = [*BACKUPS, EARLIER]
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot)
await pilot.press("a")
await pilot.pause()
app.screen.query_one("#backups", DataTable).move_cursor(row=2)
await pilot.press("r")
await settle(app, pilot)
assert isinstance(app.screen, appmod.RestoreScreen)
await apply_new(app, pilot)
await press_button(app, pilot, "#yes")
assert [c[1] for c in fake.restore_calls] == [EARLIER["volid"]]
@pytest.mark.parametrize("overwrite", [False, True])
async def test_restore_failure_notifies_error_and_releases_busy(fake, overwrite):
# spec: all CliErrors -> error notify (markup-safe), busy released
fake.restore_error = cli_error("Error: restore failed: [bold]no space[/bold]")
fake.live_status = {108: "stopped"}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
if overwrite:
await apply_overwrite(app, pilot)
else:
await apply_new(app, pilot)
await press_button(app, pilot, "#yes")
assert len(fake.restore_calls) == 1
assert any("no space" in m for m in errors(app))
assert app.busy == set() and app.is_running
async def test_next_id_failure_is_notified_not_fatal(fake):
# spec: all CliErrors -> error notify; the app keeps running
fake.nextid_error = cli_error("Error: cluster not quorate")
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot)
await pilot.press("r")
await settle(app, pilot)
assert any("quorate" in m for m in errors(app))
assert app.is_running and fake.restore_calls == []
async def test_r_on_busy_guest_makes_no_restore_call(fake):
# spec: busy guard on the source guest
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_restore(app, pilot)
app.busy.add(108)
await apply_new(app, pilot)
if isinstance(app.screen, ConfirmScreen):
await press_button(app, pilot, "#yes")
assert fake.restore_calls == []
assert any(sev == "warning" for _, sev in app.seen)
app.busy.discard(108)
# --- #15 review findings ----------------------------------------------------------------
async def test_restoring_an_earlier_guests_backup_says_so(fake):
# shown only with "a": its notes name a deleted guest that had the same VMID
fake.backup_list = [*BACKUPS, EARLIER]
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot)
await pilot.press("a")
await pilot.pause()
app.screen.query_one("#backups", DataTable).move_cursor(row=2) # the planka row
await pilot.press("r")
await settle(app, pilot)
assert isinstance(app.screen, appmod.RestoreScreen)
assert "planka" in title_of(app) and "earlier guest" in title_of(app)
async def test_restore_before_the_info_screen_loaded_says_still_loading(
fake, monkeypatch
):
release = threading.Event()
real = cli.guest_config
monkeypatch.setattr(cli, "guest_config", lambda g: (release.wait(10), real(g))[1])
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await pilot.pause()
await pilot.press("enter")
await pilot.pause()
try:
await pilot.press("r")
await pilot.pause()
assert any("loading" in m.lower() for m, _ in app.seen)
assert not isinstance(app.screen, appmod.RestoreScreen)
finally:
release.set()
await settle(app, pilot)
tests/test_cli.py
+146
View File
@@ -1621,3 +1621,149 @@ def test_fstrim_gets_a_longer_timeout_than_other_calls(fake):
fake.result = (json.dumps([{"key": "result", "value": "{}"}]), "", 0)
cli.fstrim(VM19)
assert fake.calls[0][1]["timeout"] >= 1800 # a big disk can take longer than 330 s
# --- restore a guest from a backup (issue #15) ------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Restore a guest from a backup (issue #15)" > cli.py)
CT_VOLID = "truenas-backup:backup/vzdump-lxc-104-2026_11_02-04_11_47.tar.zst"
VM_VOLID = "local:backup/vzdump-qemu-108-2026_11_02-04_11_47.vma.zst"
def test_next_id_parses_json_number_and_runs_get_cluster_nextid(fake):
# spec: next_id() -> int; `get cluster nextid -o json` -> a JSON number
fake.result = ("105\n", "", 0)
assert cli.next_id() == 105
assert fake.argv == ["cv4pve-cli", "get", "cluster", "nextid", "-o", "json"]
@pytest.mark.parametrize("volid", [CT_VOLID, VM_VOLID])
def test_valid_volid_accepts_real_volids_and_returns_them(volid):
# spec: valid_volid(volid) -> str
assert cli.valid_volid(volid) == volid
@pytest.mark.parametrize(
"volid",
[
"-y",
"",
"local:iso/x.iso",
"a b",
"x:backup/vzdump-lxc-104-",
"--force:backup/vzdump-lxc-104-x", # option-looking storage part
"local:backup/vzdump-lxc-104-x y", # whitespace in the file part
"local:backup/vzdump-vm-104-x.tar", # kind is lxc|qemu only
],
)
def test_valid_volid_rejects_everything_else(volid):
with pytest.raises(ValueError):
cli.valid_volid(volid)
def test_restore_storages_ct_wants_rootdir_vm_wants_images(fake):
# spec: enabled+active storages whose content has rootdir (CT) or images (VM), API order
fake.result = (
json.dumps([
storage("zeta", "iso,rootdir,images"),
storage("only-backup", "backup"),
storage("vm-only", "images"),
storage("ct-only", "rootdir,vztmpl"),
storage("off", "rootdir,images", enabled=0),
storage("inactive", "rootdir,images", active=0),
storage("alpha", "images,rootdir"),
]),
"",
0,
) # fmt: skip
assert cli.restore_storages(CT) == ["zeta", "ct-only", "alpha"]
assert cli.restore_storages(VM) == ["zeta", "vm-only", "alpha"]
assert fake.calls[0][0] == [
"cv4pve-cli", "get", "node", "storages", "mainframe", "-o", "json",
] # fmt: skip
def test_restore_new_vm_exact_argv_without_force_or_start(fake):
# spec: do restore <kind> <node> <vmid> <volid> --storage <storage> -y --wait
cli.restore(VM, VM_VOLID, 109, "local-lvm")
assert fake.argv == [
"cv4pve-cli", "do", "restore", "vm", "mainframe", "109", VM_VOLID,
"--storage", "local-lvm", "-y", "--wait",
] # fmt: skip
def test_restore_new_ct_with_start_exact_argv(fake):
cli.restore(CT, CT_VOLID, 105, "zfs", start=True)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "ct", "mainframe", "105", CT_VOLID,
"--storage", "zfs", "--start", "1", "-y", "--wait",
] # fmt: skip
def test_restore_overwrite_vm_exact_argv_with_force(fake):
cli.restore(VM, VM_VOLID, 108, "local-lvm", force=True)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "vm", "mainframe", "108", VM_VOLID,
"--storage", "local-lvm", "--force", "1", "-y", "--wait",
] # fmt: skip
def test_restore_overwrite_ct_with_start_exact_argv(fake):
cli.restore(CT, CT_VOLID, 100, "zfs", force=True, start=True)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "ct", "mainframe", "100", CT_VOLID,
"--storage", "zfs", "--force", "1", "--start", "1", "-y", "--wait",
] # fmt: skip
def test_restore_force_on_another_vmid_raises_and_never_runs(fake):
# spec: force=True only when vmid == guest["vmid"] - never overwrite some other guest
with pytest.raises(ValueError):
cli.restore(VM, VM_VOLID, 100, "local-lvm", force=True)
assert fake.calls == []
@pytest.mark.parametrize("vmid", [0, -1, -108])
@pytest.mark.parametrize("force", [False, True])
def test_restore_non_positive_vmid_raises_and_never_runs(fake, vmid, force):
# spec: vmid a positive int
with pytest.raises(ValueError):
cli.restore(VM, VM_VOLID, vmid, "local-lvm", force=force)
assert fake.calls == []
@pytest.mark.parametrize("volid", ["-y", "", "local:iso/x.iso", "a b"])
def test_restore_rejected_volid_raises_and_never_runs(fake, volid):
# spec: volid validated (it is a positional argument)
with pytest.raises(ValueError):
cli.restore(VM, volid, 109, "local-lvm")
assert fake.calls == []
def test_restore_failure_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: storage is full", returncode=5)
with pytest.raises(cli.CliError):
cli.restore(VM, VM_VOLID, 109, "local-lvm")
@given(vmid=st.integers(min_value=-10, max_value=10**9).filter(lambda n: n != 108))
def test_restore_force_never_runs_for_a_vmid_other_than_the_guests(vmid):
# DANGER property: --force 1 on any other guest would destroy it
with pytest.MonkeyPatch.context() as m:
run = FakeRun(m)
with pytest.raises(ValueError):
cli.restore(VM, VM_VOLID, vmid, "local-lvm", force=True)
assert run.calls == []
# --- #15 review findings ----------------------------------------------------------------
@pytest.mark.parametrize("force", [False, True])
def test_restore_refuses_a_backup_of_the_other_guest_type(fake, force):
ct = {"vmid": 100, "node": "mainframe", "type": "lxc"}
with pytest.raises(ValueError):
cli.restore(
ct, "local:backup/vzdump-qemu-100-x.vma.zst", 100, "local-lvm", force=force
)
assert fake.calls == []