From 520de16cc630005ba261c97f8959346ba4e32f83 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Mon, 2 Nov 2026 23:02:35 +0100 Subject: [PATCH] Flag guests that no backup job covers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- README.md | 3 +++ src/cv4pve_tui/app.py | 33 +++++++++++++++++++++++++++------ src/cv4pve_tui/cli.py | 6 ++++++ tests/test_app.py | 43 +++++++++++++++++++++++++++++++++++++++++++ tests/test_cli.py | 18 ++++++++++++++++++ 5 files changed, 97 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index a27bd86..fe489d6 100644 --- a/README.md +++ b/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. diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index 3260186..e1213db 100644 --- a/src/cv4pve_tui/app.py +++ b/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( - "context: " - ) + self.context_name = sub_title.removeprefix("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: diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index d3cd2c5..7760ad2 100644 --- a/src/cv4pve_tui/cli.py +++ b/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} diff --git a/tests/test_app.py b/tests/test_app.py index 7d0f1c9..909a45d 100644 --- a/tests/test_app.py +++ b/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 diff --git a/tests/test_cli.py b/tests/test_cli.py index a3d8208..6fff074 100644 --- a/tests/test_cli.py +++ b/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()