Flag guests that no backup job covers

The guest table marks with "!" each guest that Proxmox's not-backed-up
list returns. The frame's bottom border shows the count, e.g.
"mainframe · 2 without backup job". This uses Proxmox's own answer, so
it also counts jobs that select guests by pool.

If the coverage call fails, the guests are still listed and a warning
says the coverage is unknown.

Closes #9

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

README.md

@@ -35,6 +35,9 @@ The shell key needs key-based root SSH to the PVE node (LXC only).
 Switching context runs `cv4pve-cli config use`, which changes cv4pve-cli's global current
 context: other shells and scripts follow, and it stays after the app exits.
 
+A `!` in the first column marks a guest that no backup job covers (Proxmox's
+`not-backed-up` list); the frame's bottom border shows how many.
+
 Empty fields in the edit dialog are left unchanged. If a CPU/memory change cannot be
 hot-plugged, the app tells you it applies at the next reboot.
 

src/cv4pve_tui/app.py

@@ -659,6 +659,8 @@ class GuestApp(App):
         # bumped on every context switch; replies started earlier come from the old cluster
         self.generation = 0
         self.switching = False
+        self.context_name = ""
+        self.unprotected: set[int] = set()  # VMIDs no backup job covers
 
     CSS = """
     #guests_section {
@@ -679,6 +681,7 @@ class GuestApp(App):
         self.watch(self, "sub_title", self._show_context_frame)
         table = self.query_one("#guests", DataTable)
         table.add_columns(
+            "",  # "!" = no backup job covers the guest
             "VMID",
             "Name",
             "Type",
@@ -692,9 +695,13 @@ class GuestApp(App):
         self.action_reload()
 
     def _show_context_frame(self, sub_title: str) -> None:
-        self.query_one("#guests_section").border_subtitle = sub_title.removeprefix(
+        self.context_name = sub_title.removeprefix("context: ")
-            "context: "
+        self._update_frame()
-        )
+
+    def _update_frame(self) -> None:
+        n = len(self.unprotected & {g["vmid"] for g in self.guests.values()})
+        count = f" · {n} without backup job" if n else ""
+        self.query_one("#guests_section").border_subtitle = self.context_name + count
 
     @work(thread=True)
     def _show_context(self) -> None:
@@ -724,21 +731,26 @@ class GuestApp(App):
             table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value
         ]
 
-    def _fill(self, guests: list[dict], gen: int) -> None:
+    def _fill(
+        self, guests: list[dict], gen: int, unprotected: set[int] | None = None
+    ) -> None:
         if gen != self.generation:  # listed before a context switch: old cluster
             return
+        self.unprotected = unprotected or set()
         table = self.query_one("#guests", DataTable)
         current = self._selected()
         table.clear()
         self.guests = {str(g["vmid"]): g for g in guests}
         for g in guests:
-            _add_plain(table, 
+            _add_plain(table,
+                "!" if g["vmid"] in self.unprotected else "",
                 g["vmid"], g["name"], g["type"], g["node"], g["status"],
                 g.get("maxcpu", "-"), _scaled(g.get("maxmem"), 2**20), _scaled(g.get("maxdisk"), 2**30),
                 key=str(g["vmid"]),
             )  # fmt: skip
         if current and str(current["vmid"]) in self.guests:
             table.move_cursor(row=table.get_row_index(str(current["vmid"])))
+        self._update_frame()
 
     def _reload(self, acted: dict | None = None) -> None:
         """Runs in a worker thread. `acted` gets its live status: the listing lags."""
@@ -751,7 +763,16 @@ class GuestApp(App):
         except Exception as exc:  # noqa: BLE001 - any failure must reach the user, not kill the app
             self.call_from_thread(self.notify, _message(exc), severity="error")
             return
-        self.call_from_thread(self._fill, guests, gen)
+        unprotected = None
+        try:
+            unprotected = cli.not_backed_up()
+        except Exception as exc:  # noqa: BLE001 - the listing still works without it
+            self.call_from_thread(
+                self.notify,
+                f"backup coverage unknown: {_message(exc)}",
+                severity="warning",
+            )
+        self.call_from_thread(self._fill, guests, gen, unprotected)
 
     def action_reload(self) -> None:
         if self.switching:

src/cv4pve_tui/cli.py

@@ -379,3 +379,9 @@ def set_net(guest: dict, changes: dict[str, str | None]) -> None:
     run(
         "set", kind(guest), "config", guest["node"], str(guest["vmid"]), *opts, "--wait"
     )
+
+
+def not_backed_up() -> set[int]:
+    """VMIDs no backup job covers (Proxmox's own answer, incl. pool-based jobs)."""
+    items = json.loads(run("get", "cluster", "not-backed-up", "-o", "json"))
+    return {int(i["vmid"]) for i in items if i.get("vmid") is not None}

tests/test_app.py

@@ -49,6 +49,7 @@ class Fake:
         self.use_error = None
         self.use_hook = None
         self.guests_by_context = {}
+        self.unprotected, self.unprotected_error = set(), None
         self.list_hook = None
         # delivery A (guest info screen)
         self.net_configs = {}  # vmid -> extra config keys (netN, ipconfigN)
@@ -103,6 +104,11 @@ class Fake:
         self.status_calls.append(guest["vmid"])
         return self.live_status.get(guest["vmid"], guest["status"])
 
+    def not_backed_up(self):
+        if self.unprotected_error:
+            raise self.unprotected_error
+        return set(self.unprotected)
+
     def contexts(self):
         self.contexts_calls += 1
         if self.contexts_error:
@@ -195,6 +201,7 @@ def fake(monkeypatch):
         "backup_now",
         "bridges",
         "set_net",
+        "not_backed_up",
     ):
         monkeypatch.setattr(cli, name, getattr(f, name), raising=False)
     for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
@@ -2108,3 +2115,39 @@ async def test_info_resources_reload_after_edit(fake, monkeypatch):
         await settle(app, pilot)
         assert isinstance(app.screen, appmod.InfoScreen)
         assert resource_rows(app)["CPU"].startswith("8 cores")
+
+
+# --- #9: flag guests that no backup job covers --------------------------------------
+
+
+def first_cell(app, vmid):
+    return str(app.query_one("#guests", DataTable).get_row(str(vmid))[0])
+
+
+async def test_unprotected_guests_are_marked_and_counted(fake):
+    fake.unprotected = {100}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert first_cell(app, 100) == "!"
+        assert first_cell(app, 108) == ""
+        subtitle = str(app.query_one("#guests_section").border_subtitle)
+        assert "mainframe" in subtitle and "1 without backup job" in subtitle
+
+
+async def test_no_count_when_every_guest_is_covered(fake):
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert str(app.query_one("#guests_section").border_subtitle) == "mainframe"
+        assert first_cell(app, 100) == "" and first_cell(app, 108) == ""
+
+
+async def test_backup_coverage_failure_still_lists_guests(fake):
+    fake.unprotected_error = cli.CliError("Error: boom")
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        assert app.query_one("#guests", DataTable).row_count == 2
+        assert any(sev == "warning" for _, sev in app.seen)  # coverage unknown, said so
+        assert app.is_running

tests/test_cli.py

@@ -985,3 +985,21 @@ def test_apply_ip_rejects_netmask_form_and_zero_prefix(cidr):
     family = 6 if ":" in cidr else 4
     with pytest.raises(ValueError):
         cli.apply_ip({}, family, "static", cidr)
+
+
+# --- #9: guests not covered by any backup job -----------------------------------------
+
+
+def test_not_backed_up_returns_vmids(fake):
+    fake.result = (
+        json.dumps([{"vmid": 104, "name": "debian", "type": "lxc"}, {"vmid": "110"}]),
+        "",
+        0,
+    )
+    assert cli.not_backed_up() == {104, 110}
+    assert fake.argv == ["cv4pve-cli", "get", "cluster", "not-backed-up", "-o", "json"]
+
+
+def test_not_backed_up_empty(fake):
+    fake.result = ("[]", "", 0)
+    assert cli.not_backed_up() == set()