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 == []