Show live usage: CPU % and Mem % in the table, in-use on the info screen
CI / pytest and ruff (pull_request) Successful in 3m49s
CI / pytest and ruff (push) Has been cancelled

The guest table gets two last columns, CPU % (cpu * 100) and Mem %
(mem / maxmem), each with one decimal. They show "-" for a guest that
is not running or lacks the field.

The info screen's Resources table gets a header and an "In use"
column: CPU %, memory MiB with %, swap for containers, and rootfs used
GiB with % for containers. VM disks show "-", because the API reports
0 without guest-agent disk info. The info screen reads status and
usage in one status/current call (new cli.current).

After an action, the acted guest's status and usage both come from
status/current, so a guest that has just started no longer shows the
lagging listing's 0.0 %.

One blind-written test expected a VM's CPU cell to read "4 cores". VMs
have always shown sockets ("4 cores × 1 socket", as the resources
spec and its tests define), so that expectation was corrected.

Closes #12

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit was merged in pull request #24.
This commit is contained in:
DuckbootandClaude Opus 5.5 committed 2026-11-02 23:52:32 +01:00
1 parent 8c2b48a04c
commit eec65e0585
6 files changed
+356 -12

No files matched your search

+204
View File
@@ -71,6 +71,9 @@ class Fake:
self.take_calls, self.rollback_calls, self.delete_calls = [], [], []
self.snap_error_op = None
self.snap_hook = None
# issue #12 (live usage); NB the name "current" is taken by the context-name attribute above
self.usage = {} # vmid -> overrides merged into the status/current dict
self.current_error, self.current_calls = None, []
def list_guests(self):
self.list_calls += 1
@@ -112,6 +115,17 @@ class Fake:
self.status_calls.append(guest["vmid"])
return self.live_status.get(guest["vmid"], guest["status"])
def current_of(self, guest):
"""Stands in for cli.current: status/current-like dict built from the guest dict."""
self.status_calls.append(guest["vmid"])
self.current_calls.append(guest["vmid"])
if self.current_error:
raise self.current_error
data = {k: v for k, v in guest.items()
if k in ("cpu", "mem", "maxmem", "disk", "maxdisk")} # fmt: skip
data["status"] = self.live_status.get(guest["vmid"], guest["status"])
return data | self.usage.get(guest["vmid"], {})
def not_backed_up(self):
if self.unprotected_error:
raise self.unprotected_error
@@ -227,6 +241,7 @@ VM_CONFIG = {
def fake(monkeypatch):
f = Fake()
# raising=False: the new names do not exist before the implementation (red run)
monkeypatch.setattr(cli, "current", f.current_of, raising=False)
for name in (
"interfaces",
"backup_storages",
@@ -2646,3 +2661,192 @@ async def test_info_screen_scrolls_on_short_terminals(fake):
async with app.run_test(size=(100, 24)) as pilot:
await open_info(app, pilot)
assert isinstance(app.screen.query_one("#info_body"), VerticalScroll)
# --- live usage: CPU %, memory used, disk used (issue #12) ---------------------------------
# spec: docs/plans/guest-tui-mvp.md, "Live usage" > app.py
GIB, MIB = 2**30, 2**20
def column_labels(table):
return [str(c.label) for c in table.columns.values()]
async def table_cells(app, pilot, guests):
"""Open the app with the given guests; return {vmid: row cells as str}."""
fake_rows = {}
await settle(app, pilot)
table = app.query_one("#guests", DataTable)
for key in table.rows:
fake_rows[key.value] = [str(c) for c in table.get_row(key.value)]
return fake_rows
async def test_table_has_cpu_and_mem_percent_as_last_two_columns(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
assert column_labels(app.query_one("#guests", DataTable))[-2:] == [
"CPU %",
"Mem %",
]
async def test_table_running_guest_shows_cpu_and_mem_percent_one_decimal(fake):
# cpu is a 0..1 fraction -> x100; mem 932048896 of 8 GiB = 10.85 %
fake.guests = [dict(VM, cpu=0.0538, mem=932048896), CT]
app = make_app()
async with app.run_test() as pilot:
rows = await table_cells(app, pilot, fake.guests)
assert rows["108"][-2:] == ["5.4", "10.9"]
async def test_table_idle_running_guest_shows_zero_not_dash(fake):
fake.guests = [dict(VM, cpu=0.0, mem=0), CT]
app = make_app()
async with app.run_test() as pilot:
rows = await table_cells(app, pilot, fake.guests)
assert rows["108"][-2:] == ["0.0", "0.0"]
async def test_table_stopped_guest_shows_dashes(fake):
# stopped guests report 0s (or stale values); both columns are "-"
fake.guests = [VM, dict(CT, cpu=0.5, mem=256 * MIB)]
app = make_app()
async with app.run_test() as pilot:
rows = await table_cells(app, pilot, fake.guests)
assert rows["100"][-2:] == ["-", "-"]
async def test_table_running_guest_missing_fields_shows_dashes(fake):
app = make_app() # VM 108 is running but has no cpu/mem in the fixture
async with app.run_test() as pilot:
rows = await table_cells(app, pilot, fake.guests)
assert rows["108"][-2:] == ["-", "-"]
async def test_table_running_guest_with_only_cpu_shows_dash_for_mem(fake):
fake.guests = [dict(VM, cpu=0.25), CT]
app = make_app()
async with app.run_test() as pilot:
rows = await table_cells(app, pilot, fake.guests)
assert rows["108"][-2:] == ["25.0", "-"]
def in_use_rows(app):
t = app.screen.query_one("#resources", DataTable)
return {cells(t, i)[0]: cells(t, i)[2] for i in range(t.row_count)}
CT_RESOURCES = {"memory": 512, "swap": 256, "rootfs": "local-lvm:vm-100-disk-0,size=8G",
"mp0": "local-lvm:vm-100-disk-1,mp=/data,size=32G"} # fmt: skip
CT_USAGE = {"cpu": 0.0538, "cpus": 1, "mem": 256 * MIB, "maxmem": 512 * MIB, "swap": 64 * MIB,
"maxswap": 256 * MIB, "disk": 2 * GIB, "maxdisk": 8 * GIB} # fmt: skip
def running_ct(fake):
fake.net_configs[100] = CT_RESOURCES
fake.disk_map = {"rootfs": "8G", "mp0": "32G"}
fake.live_status = {100: "running"}
fake.usage[100] = CT_USAGE
async def test_info_resources_table_has_header_and_three_columns(fake):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot)
t = app.screen.query_one("#resources", DataTable)
assert t.show_header is True
assert column_labels(t) == ["Resource", "Configured", "In use"]
async def test_info_configured_column_unchanged_with_usage(fake):
running_ct(fake)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=1)
rows = resource_rows(app)
assert rows["CPU"] == "1 core" and rows["Memory"] == "512 MiB"
assert rows["Swap"] == "256 MiB"
assert rows["Disk rootfs"] == "8G" and rows["Disk mp0"] == "32G"
async def test_info_running_ct_in_use_column(fake):
running_ct(fake)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=1)
use = in_use_rows(app)
assert use["CPU"] == "5.4 %"
assert use["Memory"] == "256 MiB (50 %)"
assert use["Swap"] == "64 MiB"
assert use["Disk rootfs"] == "2.0 GiB (25 %)"
assert use["Disk mp0"] == "-" # only rootfs usage is known
async def test_info_running_vm_cpu_memory_in_use_and_disks_dash(fake):
# VMs have no guest-agent disk info: the disk In use is "-" even when the API sends a value
fake.net_configs[108] = {
"memory": 8192,
"scsi0": "local-lvm:vm-108-disk-0,size=10G",
}
fake.usage[108] = {"cpu": 0.1, "cpus": 4, "mem": 2 * GIB, "maxmem": 8 * GIB,
"disk": 3 * GIB, "maxdisk": 10 * GIB} # fmt: skip
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot)
use = in_use_rows(app)
assert use["CPU"] == "10.0 %"
assert use["Memory"] == "2048 MiB (25 %)"
assert use["Disk scsi0"] == "-"
async def test_info_stopped_guest_in_use_all_dashes(fake):
fake.net_configs[100] = CT_RESOURCES
fake.disk_map = {"rootfs": "8G", "mp0": "32G"}
fake.usage[100] = CT_USAGE # stale non-zero values; the guest is stopped
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=1)
use = in_use_rows(app)
assert {"CPU", "Memory", "Swap", "Disk rootfs"} <= set(use)
assert set(use.values()) == {"-"}
async def test_info_load_uses_one_current_call_for_status_and_usage(fake):
running_ct(fake)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=1)
assert fake.current_calls == [100]
assert fake.iface_calls == [100] # live status from the same call -> running
async def test_info_current_failure_is_inline_unavailable_and_in_use_dashes(fake):
fake.current_error = cli_error("Error: usage-probe exploded")
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen) and app.is_running
text = text_of(app.screen).lower()
assert "unavailable" in text and "usage-probe exploded" in text
assert set(in_use_rows(app).values()) == {"-"}
# configured column still loads (a VM always shows sockets, see the #resources tests)
assert resource_rows(app)["CPU"] == "4 cores × 1 socket"
assert fake.iface_calls == [] # status unknown -> treated as not running
async def test_table_usage_of_acted_guest_comes_from_live_status(fake):
# the listing lags after an action: CPU %/Mem % of the acted guest must be live too
fake.guests = [dict(CT, status="stopped", cpu=0, mem=0)]
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await settle(app, pilot)
fake.live_status = {100: "running"}
fake.usage = {100: {"cpu": 0.5, "mem": 256 * 2**20, "maxmem": 512 * 2**20}}
await pilot.press("s") # start CT 100; its listing still says stopped, 0, 0
await settle(app, pilot)
row = [str(c) for c in app.query_one("#guests", DataTable).get_row("100")]
assert "running" in row
assert row[-2:] == ["50.0", "50.0"]