Run fstrim in a VM via the guest agent

f on the info screen runs fstrim in a running VM with the guest agent
enabled. A container gets an info note (no guest agent; use pct
fstrim on the node). A stopped VM or one without an agent gets a
warning, and nothing runs. Otherwise a confirm comes first, then
agent ping and fstrim. The result reports the trimmed total
(GiB/MiB), the number of paths and any per-path errors.

Review fixes:
- Odd agent output no longer turns a finished trim into "fstrim
  failed". Unreadable output gives "no results". Odd entries are
  repaired: non-dict entries are dropped, a non-int trimmed becomes 0
  and a missing path becomes "?". Well-formed entries are returned
  unchanged.
- fstrim gets a 30-minute CLI timeout instead of 330 s, because a big
  disk can take longer.
- Notifications are never parsed as markup (GuestApp.notify defaults
  to markup=False). Guest names, paths and CLI or agent errors are
  untrusted text, so a "[/b]" in them could break the toast.
- f before the info screen has loaded says "still loading" instead of
  a wrong "not running".
- "1 path" in the singular.

Not run live: the issue asks for approval before fstrim on VM 108.
Only the harmless agent ping was run there, and it answered.

Closes #19

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

tests/test_app.py

@@ -4,6 +4,7 @@ All cli functions are monkeypatched: cv4pve-cli is never executed.
 """
 
 import contextlib
+import re
 import subprocess
 import threading
 from datetime import datetime
@@ -89,6 +90,15 @@ class Fake:
         self.nodes_error, self.updates_errors = None, {}  # node -> exception
         self.nodes_calls, self.updates_calls = 0, []
         self.updates_hook = None
+        # issue #19 (fstrim via the guest agent)
+        self.events = []  # "ping"/"fstrim" in call order
+        self.ping_error, self.fstrim_error, self.fstrim_hook, self.ping_hook = (
+            None,
+            None,
+            None,
+            None,
+        )
+        self.fstrim_paths = [{"path": "/", "trimmed": 2**30, "minimum": 0}]
 
     def list_guests(self):
         self.list_calls += 1
@@ -259,6 +269,21 @@ class Fake:
             raise self.updates_errors[node]
         return rows
 
+    def agent_ping(self, guest):
+        self.events.append(("ping", guest["vmid"]))
+        if self.ping_hook:
+            self.ping_hook()
+        if self.ping_error:
+            raise self.ping_error
+
+    def fstrim(self, guest):
+        self.events.append(("fstrim", guest["vmid"]))
+        if self.fstrim_hook:
+            self.fstrim_hook()
+        if self.fstrim_error:
+            raise self.fstrim_error
+        return [dict(p) for p in self.fstrim_paths]
+
     def task_log(self, upid):
         self.log_calls.append(upid)
         if self.log_error:
@@ -324,6 +349,8 @@ def fake(monkeypatch):
         "task_log",
         "nodes",
         "package_updates",
+        "agent_ping",
+        "fstrim",
     ):
         monkeypatch.setattr(cli, name, getattr(f, name), raising=False)
     for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
@@ -4195,3 +4222,242 @@ async def test_boot_set_net_failure_notifies_error_and_releases_busy(fake):
         assert any("unable to apply onboot" in m for m in errors(app))
         assert app.busy == set() and app.is_running
         assert len(fake.set_calls) == 1
+
+
+# --- issue #19: fstrim via the guest agent (spec: "fstrim in a VM via the guest agent") -----
+
+
+def agent_vm(fake, config="enabled=1", status="running"):
+    fake.guests = [dict(VM, status=status)]
+    fake.live_status = {108: status}
+    if config is not None:
+        fake.net_configs[108] = {"agent": config}
+
+
+def notes(app):
+    return " | ".join(m for m, _ in app.seen).lower()
+
+
+async def press_f(app, pilot):
+    await pilot.press("f")
+    await settle(app, pilot)
+
+
+@pytest.mark.parametrize("live", [None, "running"])
+async def test_fstrim_on_container_notifies_pct_fstrim_and_runs_nothing(fake, live):
+    # spec: container -> info notify (pct fstrim on the node instead); nothing runs
+    if live:
+        fake.live_status = {100: live}
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot, down=1)
+        await press_f(app, pilot)
+        assert any("pct fstrim" in m and sev == "information" for m, sev in app.seen)
+        assert fake.events == []
+        assert isinstance(app.screen, appmod.InfoScreen)
+
+
+async def test_fstrim_on_stopped_vm_warns_and_runs_nothing(fake):
+    # spec: VM not running -> warning notify saying why; nothing runs
+    agent_vm(fake, status="stopped")
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        assert any("running" in m.lower() and sev == "warning" for m, sev in app.seen)
+        assert fake.events == []
+        assert isinstance(app.screen, appmod.InfoScreen)
+
+
+@pytest.mark.parametrize("config", [None, "enabled=0", "0"])
+async def test_fstrim_on_vm_without_agent_warns_and_runs_nothing(fake, config):
+    # spec: agent_enabled false -> warning notify saying why; nothing runs
+    agent_vm(fake, config=config)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        assert any("agent" in m.lower() and sev == "warning" for m, sev in app.seen)
+        assert fake.events == []
+        assert isinstance(app.screen, appmod.InfoScreen)
+
+
+async def test_fstrim_asks_confirm_titled_fstrim_naming_vm_before_any_call(fake):
+    # spec: ConfirmScreen titled "fstrim" naming the VM; nothing runs before yes
+    agent_vm(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        assert isinstance(app.screen, ConfirmScreen)
+        assert app.screen.title_text == "fstrim"
+        assert "docker" in text_of(app.screen)
+        assert fake.events == []
+
+
+async def test_fstrim_yes_pings_then_trims_once_busy_then_done_with_size_and_count(
+    fake,
+):
+    # spec: yes -> busy guard, agent_ping then fstrim; done notify with GiB total and path count
+    agent_vm(fake)
+    fake.fstrim_paths = [
+        {"path": "/", "trimmed": 2**30, "minimum": 0},
+        {"path": "/boot", "error": "not supported"},
+    ]
+    busy_seen = []
+    fake.fstrim_hook = lambda: busy_seen.append(set(app.busy))
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        await press_button(app, pilot, "#yes")
+        assert fake.events == [("ping", 108), ("fstrim", 108)]
+        assert busy_seen and busy_seen[0], "guest must be busy while fstrim runs"
+        assert app.busy == set()
+        done = [m for m, sev in app.seen if "1.0 GiB" in m]
+        assert done, app.seen
+        assert any(re.search(r"\b2\b", m) for m in done), "path count missing"
+        assert "not supported" in notes(app), "per-path error must be mentioned"
+        assert errors(app) == []
+
+
+async def test_fstrim_small_total_is_shown_in_mib(fake):
+    # spec: trimmed total in GiB/MiB
+    agent_vm(fake)
+    fake.fstrim_paths = [{"path": "/", "trimmed": 5 * 2**20, "minimum": 0}]
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        await press_button(app, pilot, "#yes")
+        assert "mib" in notes(app) and "gib" not in notes(app)
+        assert errors(app) == []
+
+
+@pytest.mark.parametrize("how", ["no", "escape"])
+async def test_fstrim_declined_makes_no_call(fake, how):
+    # spec: ConfirmScreen; declining runs nothing
+    agent_vm(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        assert isinstance(app.screen, ConfirmScreen)
+        if how == "no":
+            await press_button(app, pilot, "#no")
+        else:
+            await pilot.press("escape")
+            await settle(app, pilot)
+        assert fake.events == []
+        assert not isinstance(app.screen, ConfirmScreen)
+        assert app.busy == set()
+
+
+async def test_fstrim_ping_failure_errors_agent_did_not_answer_and_skips_fstrim(fake):
+    # spec: CliError -> error notify (ping failure says the agent did not answer)
+    agent_vm(fake)
+    fake.ping_error = cli_error("Error: VM 108 qmp command failed - timeout")
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        await press_button(app, pilot, "#yes")
+        assert fake.events == [("ping", 108)]
+        assert any("agent" in m.lower() and "answer" in m.lower() for m in errors(app))
+        assert app.busy == set() and app.is_running
+
+
+async def test_fstrim_failure_notifies_error_and_releases_busy(fake):
+    # spec: CliError -> error notify
+    agent_vm(fake)
+    fake.fstrim_error = cli_error("Error: fstrim command failed")
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        await press_button(app, pilot, "#yes")
+        assert fake.events == [("ping", 108), ("fstrim", 108)]
+        assert any("fstrim command failed" in m for m in errors(app))
+        assert app.busy == set() and app.is_running
+        assert not any("GiB" in m or "MiB" in m for m, _ in app.seen)
+
+
+async def test_fstrim_while_guest_busy_warns_and_makes_no_second_call(fake):
+    # spec: busy guard
+    agent_vm(fake)
+    started, release = threading.Event(), threading.Event()
+    fake.ping_hook = lambda: (started.set(), release.wait(10))
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await open_info(app, pilot)
+        try:
+            await press_f(app, pilot)
+            app.screen.query_one("#yes", Button).press()
+            for _ in range(200):
+                if started.is_set():
+                    break
+                await pilot.pause(0.01)
+            assert started.is_set()
+            await pilot.press("f")
+            await pilot.pause()
+            if isinstance(app.screen, ConfirmScreen):  # refusal may come at "yes"
+                app.screen.query_one("#yes", Button).press()
+                await pilot.pause()
+            assert any(sev == "warning" for _, sev in app.seen)
+            assert fake.events.count(("ping", 108)) == 1
+        finally:
+            release.set()
+        await settle(app, pilot)
+        assert fake.events.count(("ping", 108)) == 1
+        assert fake.events.count(("fstrim", 108)) == 1
+
+
+# --- #19 review findings ----------------------------------------------------------------
+
+
+async def test_notifications_never_parse_markup(fake):
+    # guest names, paths and agent errors are untrusted text; notify must show them literally
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        await settle(app, pilot)
+        app.notify("path [/b] failed")
+        await pilot.pause()
+        assert app.is_running
+        assert all(n.markup is False for n in app._notifications)
+
+
+async def test_fstrim_without_results_says_so_and_singular_path(fake):
+    agent_vm(fake)
+    app = make_app()
+    async with app.run_test(size=SIZE) as pilot:
+        fake.fstrim_paths = []
+        await open_info(app, pilot)
+        await press_f(app, pilot)
+        await press_button(app, pilot, "#yes")
+        assert any("no result" in m.lower() for m, _ in app.seen)
+        fake.fstrim_paths = [{"path": "/", "trimmed": 2**20, "error": None}]
+        await press_f(app, pilot)
+        await press_button(app, pilot, "#yes")
+        assert any("in 1 path" in m and "paths" not in m for m, _ in app.seen)
+
+
+async def test_fstrim_before_the_info_screen_loaded_says_still_loading(
+    fake, monkeypatch
+):
+    agent_vm(fake)
+    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("f")
+            await pilot.pause()
+            assert any("loading" in m.lower() for m, _ in app.seen)
+            assert fake.events == []
+        finally:
+            release.set()
+        await settle(app, pilot)