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@<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

@@ -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

@@ -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 = [
+        shown = self._visible_backups()
-            b for b in self.backups if self.show_all or not _earlier_guest(b, name)
-        ]
         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

@@ -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

@@ -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

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