From 0df928ab11e39daad216bc5008c173fe8086511a Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Tue, 3 Nov 2026 01:32:05 +0100 Subject: [PATCH] Restore a guest from a backup in the info screen r on the info screen restores the backup under the cursor of the backup list. - New guest (default): the next free VMID from `get cluster nextid`, editable. The original is left untouched. An id already used by a listed guest is refused. - Overwrite: replaces the guest itself. The guest must be stopped; its live status is read at apply and again right before the call. The VMID must be typed to confirm, then an "Overwrite" confirm follows. --storage is always passed. It is preselected to the storage of the guest's rootfs or first disk, and the choices are the node's storages that can hold CT (rootdir) or VM (images) disks. Start after restore is off by default, because a copy keeps the original's MAC address and IP. Safety: - cli.restore refuses force (overwrite) for any VMID other than the guest's own (explicit raise). A hypothesis property checks that no other VMID ever reaches the CLI with force. - The volid and storage are validated before they become arguments. - A backup of the other guest type is refused. A reused VMID can carry a qemu backup for a CT or the other way round, and whether PVE rejects a mismatched archive before removing the old guest is not verified. - An earlier guest's backup (visible with a) is labelled with that guest's name in the dialog and the confirm. - r before the info screen has loaded says "still loading". - Restore gets a 1-hour CLI timeout. Not run live: a live test would create a guest that then has to be deleted, which the issue reserves for the user's approval. A read-only check opened the dialog for CT 104 (next id 105, local-lvm preselected) and closed it without any call. Closes #15 Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + docs/plans/guest-tui-mvp.md | 60 ++++++ src/cv4pve_tui/app.py | 186 +++++++++++++++- src/cv4pve_tui/cli.py | 61 ++++++ tests/test_app.py | 411 ++++++++++++++++++++++++++++++++++++ tests/test_cli.py | 146 +++++++++++++ 6 files changed, 862 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index bb97190..34c8011 100644 --- a/README.md +++ b/README.md @@ -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@ pct enter `) | | `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 | diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index f97efc3..a928796 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -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 [--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 --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 backup of ()"): + `RadioSet#target` with `RadioButton#new` (default, "New guest") and + `RadioButton#overwrite` ("Overwrite ()"); `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. diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index 984d314..d594d82 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -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: + """' backup of ()', 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 "" diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index 79e30c6..b50ae6c 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -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 diff --git a/tests/test_app.py b/tests/test_app.py index 522ece8..6c34d48 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -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 ":,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) diff --git a/tests/test_cli.py b/tests/test_cli.py index 756e90e..1def5e3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -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 --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 == []