Duckboot/cv4pve-cli-tui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9ff00e425facf65a80ccc518863eaf946e2da190
cv4pve-cli-tui/tests/test_app.py
T
DuckbootandClaude Opus 5.5 9ff00e425f Show all dialogs and the guest table as titled boxes 
The info screen's style now applies everywhere. The guest table sits
in a "Guests" frame with the active context in its bottom border.
Every dialog is a centred box with a title, labelled fields and the
buttons bottom right: confirm (titled by the action), edit resources,
switch context, back up and edit network.

Fixed:
- The network dialog hid most of its fields. In Horizontal rows
  Textual gave Select a width of 1 and Input 100 %, which pushed
  bridge, firewall, the IP mode selects and the gateways off-screen.
  Every field in a row now has an explicit width.
- A guest name or note containing markup, such as "a[/b]", crashed the
  guest table and the dialogs, because DataTable cells and border
  titles were parsed as markup. They are now plain text.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 22:18:10 +01:00

1989 lines
70 KiB
Python
Raw Blame History

"""Textual pilot tests for cv4pve_tui.app; spec: docs/plans/guest-tui-mvp.md ("UI contract").
All cli functions are monkeypatched: cv4pve-cli is never executed.
"""
import contextlib
import subprocess
import threading
from datetime import datetime
import pytest
from textual.screen import ModalScreen
from textual.widgets import (
Button,
DataTable,
Input,
OptionList,
Select,
Static,
)
from cv4pve_tui import app as appmod
from cv4pve_tui import cli
from cv4pve_tui.app import ConfirmScreen, ContextScreen, EditScreen, GuestApp
VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu", "status": "running",
"maxcpu": 4, "maxmem": 8192 * 2**20, "maxdisk": 10 * 2**30} # fmt: skip
CT = {"vmid": 100, "name": "backrest", "node": "mainframe", "type": "lxc", "status": "stopped",
"maxcpu": 1, "maxmem": 512 * 2**20, "maxdisk": 8 * 2**30} # fmt: skip
class Fake:
"""Recorder standing in for the cli module functions the app calls."""
def __init__(self):
self.guests = [VM, CT]
self.pending = []
self.disk_map = {"scsi0": "10G"}
self.list_calls = 0
self.actions, self.sets, self.resizes, self.pending_calls = [], [], [], []
self.action_error = None
self.action_hook = None
self.live_status, self.status_calls = {}, []
self.current = "mainframe"
self.names = ["mainframe", "mainframe-edge"]
self.contexts_error = None
self.contexts_calls = 0
self.use_calls = []
self.use_error = None
self.use_hook = None
self.guests_by_context = {}
self.list_hook = None
# delivery A (guest info screen)
self.net_configs = {} # vmid -> extra config keys (netN, ipconfigN)
self.storages = ["truenas-backup", "local"]
self.jobs = [BACKUP_JOB]
self.backup_list = list(BACKUPS) # newest first, as cli.backups returns them
self.iface_list, self.iface_error = [], None
self.iface_calls, self.backups_calls, self.backup_runs = [], [], []
self.backup_error = None
self.backup_hook = None
# delivery B (network editing)
self.bridge_list = ["vmbr0"]
self.set_calls, self.set_error, self.set_hook = [], None, None
def list_guests(self):
self.list_calls += 1
# copies: the app writes live status into these dicts; shared module constants must not change
snapshot = [
dict(g) for g in self.guests_by_context.get(self.current, self.guests)
]
if (
self.list_hook
): # runs after reading: simulates a slow reply from that cluster
self.list_hook()
return snapshot
def guest_config(self, guest):
return {"cores": guest["maxcpu"], **self.net_configs.get(guest["vmid"], {})}
def disks(self, config, kind):
return dict(self.disk_map)
def action(self, name, guest):
self.actions.append((name, guest["vmid"]))
if self.action_hook:
self.action_hook()
if self.action_error:
raise self.action_error
def set_resources(self, guest, cores=None, memory_mb=None):
given = {"cores": cores, "memory_mb": memory_mb}
self.sets.append({k: v for k, v in given.items() if v is not None})
def resize_disk(self, guest, disk, grow_gb):
self.resizes.append((guest["vmid"], disk, grow_gb))
def pending_keys(self, guest):
self.pending_calls.append(guest["vmid"])
return list(self.pending)
def status(self, guest):
self.status_calls.append(guest["vmid"])
return self.live_status.get(guest["vmid"], guest["status"])
def contexts(self):
self.contexts_calls += 1
if self.contexts_error:
raise self.contexts_error
return [
{"name": n, "host": f"{n}.home.arpa:8006", "auth": "api-token",
"current": n == self.current}
for n in self.names
] # fmt: skip
def use_context(self, name):
self.use_calls.append(name)
if self.use_hook:
self.use_hook()
if self.use_error:
raise self.use_error
self.current = name
def interfaces(self, guest):
self.iface_calls.append(guest["vmid"])
if self.iface_error:
raise self.iface_error
return list(self.iface_list)
def backup_storages(self, guest):
return list(self.storages)
def backup_jobs(self):
return list(self.jobs)
def backups(self, guest, storages):
self.backups_calls.append((guest["vmid"], list(storages)))
return list(self.backup_list)
def backup_now(self, guest, storage):
self.backup_runs.append((guest["vmid"], storage))
if self.backup_hook:
self.backup_hook()
if self.backup_error:
raise self.backup_error
def bridges(self, guest):
return list(self.bridge_list)
def set_net(self, guest, changes):
self.set_calls.append((guest["vmid"], dict(changes)))
if self.set_hook:
self.set_hook()
if self.set_error:
raise self.set_error
config = self.net_configs.setdefault(guest["vmid"], {})
for (
key,
value,
) in changes.items(): # mirror the server so reloads see the change
if value is None:
config.pop(key, None)
else:
config[key] = value
BACKUP_JOB = {"id": "backup-1", "enabled": 1, "storage": "truenas-backup", "all": 1,
"vmid": None, "exclude": None, "node": None, "schedule": "mon,wed,fri 04:00",
"mode": "snapshot", "compress": "zstd", "prune-backups": {"keep-last": "3"}} # fmt: skip
T_OLD, T_NEW = 1_762_000_000, 1_762_500_000 # epoch seconds
BACKUPS = [
{"volid": "truenas-backup:backup/vzdump-qemu-108-new.vma.zst", "ctime": T_NEW,
"size": 3 * 2**30, "format": "vma.zst", "notes": "docker (manual)", "storage": "truenas-backup"},
{"volid": "local:backup/vzdump-qemu-108-old.vma.zst", "ctime": T_OLD,
"size": int(1.5 * 2**30), "format": "vma.zst", "notes": "", "storage": "local"},
] # fmt: skip
CT_CONFIG = {
"net0": "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=auto,type=veth"
}
VM_CONFIG = {
"net0": "virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,firewall=1",
"ipconfig0": "ip=dhcp",
}
@pytest.fixture
def fake(monkeypatch):
f = Fake()
# raising=False: the new names do not exist before the implementation (red run)
for name in (
"interfaces",
"backup_storages",
"backup_jobs",
"backups",
"backup_now",
"bridges",
"set_net",
):
monkeypatch.setattr(cli, name, getattr(f, name), raising=False)
for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
"resize_disk", "pending_keys", "status", "contexts", "use_context"): # fmt: skip
monkeypatch.setattr(cli, name, getattr(f, name))
return f
def make_app():
app = GuestApp()
app.seen = [] # (message, severity) of every notify
original = app.notify
def spy(message, **kwargs):
app.seen.append((str(message), kwargs.get("severity", "information")))
return original(message, **kwargs)
app.notify = spy
return app
async def settle(app, pilot):
await pilot.pause()
await app.workers.wait_for_complete()
await pilot.pause()
def text_of(screen):
return " ".join(str(w.render()) for w in screen.query(Static))
def errors(app):
return [m for m, sev in app.seen if sev == "error"]
async def press_button(app, pilot, selector):
app.screen.query_one(selector, Button).press()
await settle(app, pilot)
# --- table ------------------------------------------------------------------
async def test_table_loads_one_row_per_guest_in_list_order(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
table = app.query_one("#guests", DataTable)
assert table.row_count == 2
assert [k.value for k in table.rows] == ["108", "100"]
cells = [str(c) for c in table.get_row("108")]
assert "docker" in cells and "mainframe" in cells and "running" in cells
assert "8192" in cells # memory in MiB
async def test_refresh_key_reloads_guests(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
before = fake.list_calls
fake.guests = [CT]
await pilot.press("g")
await settle(app, pilot)
assert fake.list_calls == before + 1
assert app.query_one("#guests", DataTable).row_count == 1
# --- lifecycle actions ------------------------------------------------------
async def test_start_runs_without_confirm_and_refreshes_table(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
before = fake.list_calls
await pilot.press("s")
await settle(app, pilot)
assert fake.actions == [("start", 108)]
assert not isinstance(app.screen, ModalScreen)
assert fake.list_calls > before
async def test_action_targets_the_cursor_row(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press("down", "s")
await settle(app, pilot)
assert fake.actions == [("start", 100)]
@pytest.mark.parametrize(
"key,name", [("S", "shutdown"), ("x", "stop"), ("r", "reboot")]
)
async def test_confirmed_action_calls_cli_only_after_yes(fake, key, name):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press(key)
await pilot.pause()
assert isinstance(app.screen, ConfirmScreen)
assert "docker" in text_of(app.screen) and "108" in text_of(app.screen)
assert fake.actions == []
await press_button(app, pilot, "#yes")
assert fake.actions == [(name, 108)]
@pytest.mark.parametrize("key", ["S", "x", "r"])
async def test_declined_action_does_not_call_cli(fake, key):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press(key)
await pilot.pause()
await press_button(app, pilot, "#no")
assert fake.actions == []
assert not isinstance(app.screen, ModalScreen)
async def test_stop_confirm_says_hard_power_off(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press("x")
await pilot.pause()
assert "hard power-off" in text_of(app.screen).lower()
@pytest.mark.parametrize("key", ["s", "S", "x", "r", "e"])
async def test_keys_on_empty_table_do_nothing(fake, key):
fake.guests = []
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press(key)
await settle(app, pilot)
assert fake.actions == [] and fake.sets == [] and fake.resizes == []
assert not isinstance(app.screen, ModalScreen)
async def test_cli_error_is_notified_with_summary(fake, monkeypatch):
with monkeypatch.context() as m: # build a real CliError through cli.run
m.setattr(cli.subprocess, "run", lambda argv, **kw: subprocess.CompletedProcess(
argv, 5, "UPID:mainframe:0:0:0:qmstart:108:root@pam:\nstart failed: no quorum\n", "")) # fmt: skip
with pytest.raises(cli.CliError) as exc:
cli.run("do", "start")
fake.action_error = exc.value
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press("s")
await settle(app, pilot)
assert any(exc.value.summary in m for m in errors(app))
assert not any("UPID" in m for m in errors(app))
async def test_busy_guest_refuses_second_action(fake):
started, release = threading.Event(), threading.Event()
fake.action_hook = lambda: (started.set(), release.wait(10))
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
try:
await pilot.press("s")
for _ in range(200):
if started.is_set():
break
await pilot.pause(0.01)
assert started.is_set()
assert app.seen, "a working notice is expected while the action runs"
await pilot.press("s")
await pilot.pause()
assert fake.actions == [("start", 108)]
finally:
release.set()
await settle(app, pilot)
assert fake.actions == [("start", 108)]
# --- edit flow --------------------------------------------------------------
async def open_edit(app, pilot):
await settle(app, pilot)
await pilot.press("e")
await settle(app, pilot)
assert isinstance(app.screen, EditScreen)
def fill(app, **values):
for key, value in values.items():
w = app.screen.query_one(f"#{key}", Select if key == "disk" else Input)
w.value = value
async def test_edit_applies_only_given_fields_as_ints(fake):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, cores="8")
await press_button(app, pilot, "#apply")
assert fake.sets == [{"cores": 8}]
await pilot.press("e")
await settle(app, pilot)
fill(app, cores="2", memory="4096")
await press_button(app, pilot, "#apply")
assert fake.sets[-1] == {"cores": 2, "memory_mb": 4096}
assert all(isinstance(v, int) for v in fake.sets[-1].values())
@pytest.mark.parametrize("field", ["cores", "memory"])
@pytest.mark.parametrize("bad", ["abc", "0", "-4", "2.5"])
async def test_edit_invalid_resource_input_errors_without_cli_call(fake, field, bad):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, **{field: bad})
await press_button(app, pilot, "#apply")
assert errors(app)
assert fake.sets == [] and fake.resizes == []
@pytest.mark.parametrize("bad", ["abc", "0", "-1"])
async def test_edit_invalid_grow_errors_without_cli_call(fake, bad):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, disk="scsi0", grow=bad)
await press_button(app, pilot, "#apply")
assert errors(app)
assert fake.resizes == []
async def test_edit_with_empty_fields_changes_nothing(fake):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
await press_button(app, pilot, "#apply")
assert fake.sets == [] and fake.resizes == []
async def test_edit_cancel_changes_nothing(fake):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, cores="8")
await press_button(app, pilot, "#cancel")
assert fake.sets == [] and fake.resizes == []
assert not isinstance(app.screen, ModalScreen)
async def test_grow_needs_confirm_then_resizes(fake):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, disk="scsi0", grow="5")
await press_button(app, pilot, "#apply")
assert isinstance(app.screen, ConfirmScreen)
assert fake.resizes == []
await press_button(app, pilot, "#yes")
assert fake.resizes == [(108, "scsi0", 5)]
async def test_grow_declined_does_not_resize(fake):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, disk="scsi0", grow="5")
await press_button(app, pilot, "#apply")
await press_button(app, pilot, "#no")
assert fake.resizes == []
@pytest.mark.parametrize(
"pending,expect_notice", [(["memory"], True), (["cores"], True), ([], False)]
)
async def test_pending_cores_or_memory_notifies_reboot(fake, pending, expect_notice):
fake.pending = pending
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, memory="4096")
await press_button(app, pilot, "#apply")
assert fake.pending_calls == [108]
assert any("reboot" in m.lower() for m, _ in app.seen) is expect_notice
# --- review findings: robustness --------------------------------------------
async def test_guest_without_size_fields_still_listed(fake):
# e.g. a guest on an unreachable node: the API omits maxcpu/maxmem/maxdisk
fake.guests = [
{
"vmid": 999,
"name": "ghost",
"node": "down",
"type": "qemu",
"status": "unknown",
}
]
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
assert app.query_one("#guests", DataTable).row_count == 1
async def test_unexpected_error_in_action_is_notified_not_fatal(fake):
fake.action_error = KeyError("key")
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press("s")
await settle(app, pilot)
assert errors(app)
assert app.is_running and app.busy == set()
async def test_unexpected_error_in_listing_is_notified_not_fatal(fake, monkeypatch):
def broken():
raise ValueError("bad json")
monkeypatch.setattr(cli, "list_guests", broken)
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
assert errors(app) and app.is_running
async def test_pending_check_failure_does_not_skip_confirmed_grow(fake, monkeypatch):
def broken(guest):
raise cli.CliError("Error: boom")
monkeypatch.setattr(cli, "pending_keys", broken)
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, memory="4096", disk="scsi0", grow="5")
await press_button(app, pilot, "#apply")
await press_button(app, pilot, "#yes")
assert fake.sets == [{"memory_mb": 4096}]
assert fake.resizes == [(108, "scsi0", 5)]
async def test_declining_grow_cancels_whole_edit(fake):
app = make_app()
async with app.run_test() as pilot:
await open_edit(app, pilot)
fill(app, cores="8", disk="scsi0", grow="5")
await press_button(app, pilot, "#apply")
await press_button(app, pilot, "#no")
assert fake.sets == [] and fake.resizes == []
@pytest.mark.parametrize("key", ["S", "e"])
async def test_escape_closes_modal_without_cli_call(fake, key):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press(key)
await settle(app, pilot)
assert isinstance(app.screen, ModalScreen)
await pilot.press("escape")
await settle(app, pilot)
assert not isinstance(app.screen, ModalScreen)
assert fake.actions == [] and fake.sets == []
async def test_table_shows_live_status_after_action_not_lagging_list(fake):
# /cluster/resources lags a few seconds behind; the acted-on guest must show live status
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
fake.live_status = {108: "stopped"} # list_guests still says "running"
await pilot.press("S")
await settle(app, pilot)
await press_button(app, pilot, "#yes")
assert fake.status_calls == [108]
assert "stopped" in [
str(c) for c in app.query_one("#guests", DataTable).get_row("108")
]
# --- context switching ------------------------------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Context switching" > "app.py (UI contract)")
async def open_contexts(app, pilot):
await settle(app, pilot)
await pilot.press("c")
await settle(app, pilot)
assert isinstance(app.screen, ContextScreen)
return app.screen.query_one("#contexts", OptionList)
async def choose(app, pilot, index):
ol = app.screen.query_one("#contexts", OptionList)
ol.focus()
ol.highlighted = index
await pilot.pause()
await pilot.press("enter")
await settle(app, pilot)
async def test_sub_title_shows_active_context_after_mount(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
assert app.sub_title == "context: mainframe"
async def test_contexts_failure_notifies_and_sub_title_is_unknown(fake):
fake.contexts_error = cli.CliError("Error: no config")
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
assert errors(app)
assert app.sub_title == "context: ?"
async def test_c_opens_context_screen_listing_contexts_with_current_marked(fake):
app = make_app()
async with app.run_test() as pilot:
ol = await open_contexts(app, pilot)
assert ol.option_count == 2
options = [ol.get_option_at_index(i) for i in range(2)]
assert [o.id for o in options] == ["mainframe", "mainframe-edge"]
assert "*" in str(options[0].prompt)
assert "*" not in str(options[1].prompt)
assert "mainframe-edge" in str(options[1].prompt)
async def test_selecting_other_context_switches_and_reloads(fake):
app = make_app()
async with app.run_test() as pilot:
await open_contexts(app, pilot)
before = fake.list_calls
await choose(app, pilot, 1)
assert fake.use_calls == ["mainframe-edge"]
assert app.sub_title == "context: mainframe-edge"
assert fake.list_calls > before
assert not isinstance(app.screen, ModalScreen)
assert app.query_one("#guests", DataTable).row_count == 2
async def test_table_is_cleared_while_switch_is_in_progress(fake):
started, release = threading.Event(), threading.Event()
fake.use_hook = lambda: (started.set(), release.wait(10))
app = make_app()
async with app.run_test() as pilot:
await open_contexts(app, pilot)
assert app.query_one("#guests", DataTable).row_count == 2
try:
ol = app.screen.query_one("#contexts", OptionList)
ol.focus()
ol.highlighted = 1
await pilot.pause()
await pilot.press("enter")
for _ in range(200):
if started.is_set():
break
await pilot.pause(0.01)
assert started.is_set()
await pilot.pause()
assert app.query_one("#guests", DataTable).row_count == 0
finally:
release.set()
await settle(app, pilot)
assert app.query_one("#guests", DataTable).row_count == 2
async def test_selecting_current_context_makes_no_cli_call(fake):
app = make_app()
async with app.run_test() as pilot:
await open_contexts(app, pilot)
before = fake.list_calls
await choose(app, pilot, 0)
assert fake.use_calls == []
assert not isinstance(app.screen, ModalScreen)
assert app.sub_title == "context: mainframe"
assert fake.list_calls == before
async def test_escape_closes_context_screen_without_cli_call(fake):
app = make_app()
async with app.run_test() as pilot:
await open_contexts(app, pilot)
await pilot.press("escape")
await settle(app, pilot)
assert not isinstance(app.screen, ModalScreen)
assert fake.use_calls == []
assert app.sub_title == "context: mainframe"
async def test_c_while_guest_busy_warns_and_opens_nothing(fake):
started, release = threading.Event(), threading.Event()
fake.action_hook = lambda: (started.set(), release.wait(10))
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
try:
await pilot.press("s")
for _ in range(200):
if started.is_set():
break
await pilot.pause(0.01)
assert started.is_set()
await pilot.press("c")
await pilot.pause()
assert not isinstance(app.screen, ContextScreen)
assert any(sev == "warning" for _, sev in app.seen)
finally:
release.set()
await settle(app, pilot)
assert fake.use_calls == []
async def test_use_context_failure_notifies_rereads_context_and_reloads(fake):
err = cli.CliError("Error: Context 'mainframe-edge' not found.")
def hook(): # the CLI state differs from what the app assumed
fake.current = "mainframe-edge"
fake.use_hook, fake.use_error = hook, err
app = make_app()
async with app.run_test() as pilot:
await open_contexts(app, pilot)
contexts_before, list_before = fake.contexts_calls, fake.list_calls
await choose(app, pilot, 1)
assert any(err.summary in m for m in errors(app))
assert fake.contexts_calls > contexts_before
assert app.sub_title == "context: mainframe-edge" # whatever contexts() says
assert fake.list_calls > list_before
assert app.query_one("#guests", DataTable).row_count == 2
# --- review finding: no old-cluster rows after a switch ----------------------
EDGE = {"vmid": 300, "name": "edge-vm", "node": "edge", "type": "qemu", "status": "running",
"maxcpu": 2, "maxmem": 2**30, "maxdisk": 2**31} # fmt: skip
async def until(pilot, cond):
for _ in range(300):
if cond():
return
await pilot.pause(0.01)
raise AssertionError("condition not reached")
def blocking_once():
started, release = threading.Event(), threading.Event()
def hook():
if not started.is_set():
started.set()
release.wait(10)
return hook, started, release
def rows(app):
return [k.value for k in app.query_one("#guests", DataTable).rows]
async def switch_to_edge(app, pilot):
await pilot.press("c")
await until(pilot, lambda: isinstance(app.screen, ContextScreen))
ol = app.screen.query_one("#contexts", OptionList)
ol.focus()
ol.highlighted = 1
await pilot.pause()
await pilot.press("enter")
async def test_reload_in_flight_before_switch_cannot_restore_old_rows(fake):
fake.guests_by_context = {"mainframe-edge": [EDGE]}
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
hook, started, release = blocking_once()
fake.list_hook = hook
try:
await pilot.press("g") # reads mainframe, then stalls
await until(pilot, started.is_set)
await switch_to_edge(app, pilot)
await until(pilot, lambda: rows(app) == ["300"])
finally:
release.set()
await settle(app, pilot)
assert rows(app) == ["300"]
assert app.sub_title == "context: mainframe-edge"
async def test_refresh_refused_while_switching(fake):
started, release = threading.Event(), threading.Event()
fake.use_hook = lambda: (started.set(), release.wait(10))
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
try:
await switch_to_edge(app, pilot)
await until(pilot, started.is_set)
before = fake.list_calls
await pilot.press("g")
await pilot.pause()
assert fake.list_calls == before
assert any(sev == "warning" for _, sev in app.seen)
finally:
release.set()
await settle(app, pilot)
async def test_context_refused_while_post_action_reload_runs(fake):
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
hook, started, release = blocking_once()
fake.list_hook = hook
try:
await pilot.press("s") # start is instant; the reload after it stalls
await until(pilot, started.is_set)
await pilot.press("c")
await pilot.pause(0.1)
assert not isinstance(app.screen, ContextScreen)
assert any(sev == "warning" for _, sev in app.seen)
finally:
release.set()
await settle(app, pilot)
# --- LXC shell --------------------------------------------------------------
# spec: docs/plans/guest-tui-mvp.md ("LXC shell" > "app.py (UI contract)")
RUNNING_CT = {"vmid": 104, "name": "debian", "node": "mainframe", "type": "lxc", "status": "running",
"maxcpu": 1, "maxmem": 512 * 2**20, "maxdisk": 8 * 2**30} # fmt: skip
SSH_ARGV = ["ssh", "-t", "root@mainframe", "pct", "enter", "104"]
def patch_shell(monkeypatch, app, rc=0, error=None):
"""Replace subprocess.run and app.suspend with recorders; returns the event log."""
events = []
def run(argv, **kwargs):
events.append(("run", list(argv)))
if error:
raise error
return subprocess.CompletedProcess(argv, rc)
@contextlib.contextmanager
def suspend():
events.append(("enter", None))
yield
events.append(("exit", None))
monkeypatch.setattr("cv4pve_tui.app.subprocess.run", run)
app.suspend = suspend
return events
def ssh_runs(events):
return [argv for kind, argv in events if kind == "run"]
async def press_t(app, pilot):
await settle(app, pilot)
await pilot.press("t")
await settle(app, pilot)
async def test_t_on_running_lxc_runs_ssh_pct_enter_inside_suspend(fake, monkeypatch):
# spec: runs the ssh argv once, on the main thread, inside `with self.suspend()`
fake.guests = [RUNNING_CT]
app = make_app()
events = patch_shell(monkeypatch, app)
async with app.run_test() as pilot:
await press_t(app, pilot)
assert events == [("enter", None), ("run", SSH_ARGV), ("exit", None)]
assert not errors(app)
assert app.is_running
async def test_shell_exit_nonzero_notifies_error_with_guest_and_code(fake, monkeypatch):
# spec: non-zero exit -> error notify naming the guest and the exit code
fake.guests = [RUNNING_CT]
app = make_app()
events = patch_shell(monkeypatch, app, rc=255)
async with app.run_test() as pilot:
await press_t(app, pilot)
assert ssh_runs(events) == [SSH_ARGV]
assert any(("debian" in m or "104" in m) and "255" in m for m in errors(app))
assert app.is_running
@pytest.mark.parametrize("guest,why", [(VM, "not a container"), (CT, "not running")])
async def test_t_on_ineligible_guest_warns_and_runs_nothing(
fake, monkeypatch, guest, why
):
# spec: only type lxc and status running; otherwise warning (says why), nothing runs
fake.guests = [guest]
app = make_app()
events = patch_shell(monkeypatch, app)
async with app.run_test() as pilot:
await press_t(app, pilot)
assert events == []
assert any(sev == "warning" for _, sev in app.seen)
assert not errors(app)
async def test_t_on_empty_table_does_nothing(fake, monkeypatch):
# spec: empty table -> nothing
fake.guests = []
app = make_app()
events = patch_shell(monkeypatch, app)
async with app.run_test() as pilot:
await press_t(app, pilot)
assert events == []
assert app.seen == []
assert app.is_running
async def test_t_on_busy_guest_warns_and_runs_nothing(fake, monkeypatch):
# spec: busy guest (action running) -> warning notify, nothing runs
started, release = threading.Event(), threading.Event()
fake.guests = [RUNNING_CT]
fake.action_hook = lambda: (started.set(), release.wait(10))
app = make_app()
events = patch_shell(monkeypatch, app)
async with app.run_test() as pilot:
await settle(app, pilot)
try:
await pilot.press("s")
await until(pilot, started.is_set)
warnings_before = sum(sev == "warning" for _, sev in app.seen)
await pilot.press("t")
await pilot.pause()
assert events == []
assert sum(sev == "warning" for _, sev in app.seen) > warnings_before
finally:
release.set()
await settle(app, pilot)
assert events == []
async def test_suspend_not_supported_notifies_error_and_app_keeps_running(
fake, monkeypatch
):
# spec: SuspendNotSupported (headless driver, suspend left unpatched) -> error notify
fake.guests = [RUNNING_CT]
app = make_app()
calls = []
monkeypatch.setattr(
"cv4pve_tui.app.subprocess.run", lambda argv, **kw: calls.append(argv)
)
async with app.run_test() as pilot:
await press_t(app, pilot)
assert calls == []
assert errors(app)
assert app.is_running
async def test_missing_ssh_binary_notifies_error_and_app_keeps_running(
fake, monkeypatch
):
# spec: OSError (no ssh binary) -> error notify, app keeps running
fake.guests = [RUNNING_CT]
app = make_app()
events = patch_shell(monkeypatch, app, error=FileNotFoundError("ssh"))
async with app.run_test() as pilot:
await press_t(app, pilot)
assert ssh_runs(events) == [SSH_ARGV]
assert errors(app)
assert app.is_running
# --- review findings: LXC shell ----------------------------------------------
async def test_missing_ssh_binary_still_resumes_app(fake, monkeypatch):
# Textual's suspend() has no try/finally: an exception inside it skips the resume
fake.guests = [RUNNING_CT]
app = make_app()
events = patch_shell(monkeypatch, app, error=FileNotFoundError("ssh"))
async with app.run_test() as pilot:
await press_t(app, pilot)
assert events[-1] == ("exit", None)
async def test_shell_exit_code_from_inside_container_is_not_an_error(fake, monkeypatch):
# pct enter returns the shell's last status; only 255 means ssh itself failed
fake.guests = [RUNNING_CT]
app = make_app()
patch_shell(monkeypatch, app, rc=1)
async with app.run_test() as pilot:
await press_t(app, pilot)
assert errors(app) == []
async def test_table_reloads_after_shell(fake, monkeypatch):
fake.guests = [RUNNING_CT]
app = make_app()
patch_shell(monkeypatch, app)
async with app.run_test() as pilot:
await settle(app, pilot)
before = fake.list_calls
await press_t(app, pilot)
await settle(app, pilot)
assert fake.list_calls > before
async def test_terminal_is_reset_inside_suspend_before_ssh(fake, monkeypatch):
# user request: a clean screen (and no leftover TUI terminal modes) in the container shell
fake.guests = [RUNNING_CT]
app = make_app()
events = patch_shell(monkeypatch, app)
resets = []
monkeypatch.setattr(
"cv4pve_tui.app._reset_terminal", lambda: resets.append(list(events))
)
async with app.run_test() as pilot:
await press_t(app, pilot)
assert resets == [[("enter", None)]] # inside suspend, before ssh ran
# --- delivery A: guest info screen / backup now ------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Guest info screen ... delivery A" > app.py)
def local_date(epoch):
return datetime.fromtimestamp(epoch).strftime("%Y-%m-%d %H:%M")
def cells(table, row):
return [str(c) for c in table.get_row_at(row)]
def cli_error(text):
with pytest.MonkeyPatch.context() as m:
m.setattr(
cli.subprocess,
"run",
lambda argv, **kw: subprocess.CompletedProcess(argv, 5, text, ""),
)
with pytest.raises(cli.CliError) as exc:
cli.run("do", "x")
return exc.value
async def open_info(app, pilot, down=0):
await settle(app, pilot)
for _ in range(down):
await pilot.press("down")
await pilot.press("i")
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
async def open_backup(app, pilot):
await pilot.press("b")
await settle(app, pilot)
assert isinstance(app.screen, appmod.BackupScreen)
async def test_i_opens_info_screen_with_backups_newest_first(fake):
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
table = app.screen.query_one("#backups", DataTable)
assert table.row_count == 2
new, old = cells(table, 0), cells(table, 1)
assert local_date(T_NEW) in new and "truenas-backup" in new
assert any("3.0" in c for c in new) and "docker (manual)" in new
assert local_date(T_OLD) in old and "local" in old
assert any("1.5" in c for c in old)
assert fake.backups_calls == [(108, ["truenas-backup", "local"])]
async def test_info_jobs_line_names_schedule_storage_and_keep_last(fake):
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
text = str(app.screen.query_one("#jobs", Static).render())
assert (
"mon,wed,fri 04:00" in text
and "truenas-backup" in text
and "keep-last" in text
)
async def test_info_jobs_line_when_no_job_covers_guest(fake):
fake.jobs = []
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
text = str(app.screen.query_one("#jobs", Static).render())
assert "No backup job covers this guest" in text
async def test_info_nets_row_for_ct_sample(fake):
fake.net_configs[100] = CT_CONFIG
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot, down=1) # CT 100 (stopped)
table = app.screen.query_one("#nets", DataTable)
assert table.row_count == 1
assert cells(table, 0) == [
"net0",
"vmbr0",
"-",
"BC:24:11:36:DE:78",
"dhcp",
"auto",
"no",
]
async def test_info_nets_row_for_vm_uses_ipconfig_tag_and_firewall(fake):
fake.net_configs[108] = VM_CONFIG
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
table = app.screen.query_one("#nets", DataTable)
assert cells(table, 0) == [
"net0",
"vmbr0",
"20",
"02:24:A6:49:A8:25",
"dhcp",
"-",
"yes",
]
async def test_info_addresses_show_runtime_ips_for_running_guest(fake):
fake.iface_list = [{"name": "ens18", "mac": "02:24:A6:49:A8:25",
"ipv4": ["172.16.41.50/23"], "ipv6": ["fe80::1/64"]}] # fmt: skip
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
row = cells(app.screen.query_one("#addresses", DataTable), 0)
assert row[0] == "ens18" and row[1] == "02:24:A6:49:A8:25"
assert "172.16.41.50/23" in row[2] and "fe80::1/64" in row[3]
assert fake.iface_calls == [108]
async def test_info_does_not_query_interfaces_for_stopped_guest(fake):
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot, down=1) # CT 100 is stopped
assert fake.iface_calls == []
async def test_info_interfaces_failure_is_inline_unavailable_not_fatal(fake):
fake.iface_error = cli_error("Error: QEMU guest agent is not running")
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
text = text_of(app.screen).lower()
assert "unavailable" in text and "guest agent is not running" in text
assert errors(app) == [] and app.is_running
assert (
app.screen.query_one("#backups", DataTable).row_count == 2
) # rest still loaded
async def test_i_on_empty_table_does_nothing(fake):
fake.guests = []
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
await pilot.press("i")
await settle(app, pilot)
assert not isinstance(app.screen, appmod.InfoScreen)
assert fake.backups_calls == []
async def test_escape_closes_info_and_g_reloads_it(fake):
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
before = len(fake.backups_calls)
await pilot.press("g")
await settle(app, pilot)
assert len(fake.backups_calls) == before + 1
await pilot.press("escape")
await settle(app, pilot)
assert not isinstance(app.screen, appmod.InfoScreen)
async def test_b_opens_backup_screen_preselecting_job_storage(fake):
fake.storages = ["local", "truenas-backup"] # job storage is not the first
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
await open_backup(app, pilot)
assert app.screen.query_one("#storage", Select).value == "truenas-backup"
text = text_of(app.screen).lower()
assert "snapshot" in text and "zstd" in text and "docker" in text
assert fake.backup_runs == []
async def test_backup_storage_falls_back_to_first_without_job(fake):
fake.jobs = []
fake.storages = ["local", "truenas-backup"]
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
await open_backup(app, pilot)
assert app.screen.query_one("#storage", Select).value == "local"
async def test_backup_yes_runs_backup_now_once_busy_then_done_and_reloads(fake):
app = make_app()
busy_seen = []
fake.backup_hook = lambda: busy_seen.append(set(app.busy))
async with app.run_test() as pilot:
await open_info(app, pilot)
before = len(fake.backups_calls)
await open_backup(app, pilot)
await press_button(app, pilot, "#yes")
assert fake.backup_runs == [(108, "truenas-backup")]
assert busy_seen and busy_seen[0], (
"guest must be marked busy while the backup runs"
)
assert app.busy == set()
msgs = [m.lower() for m, _ in app.seen]
assert any("backup started" in m for m in msgs)
assert any("backup done" in m for m in msgs)
assert errors(app) == []
assert len(fake.backups_calls) > before # list reloaded
assert isinstance(app.screen, appmod.InfoScreen)
@pytest.mark.parametrize("how", ["no", "escape"])
async def test_backup_declined_makes_no_call(fake, how):
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
await open_backup(app, pilot)
if how == "no":
await press_button(app, pilot, "#no")
else:
await pilot.press("escape")
await settle(app, pilot)
assert fake.backup_runs == []
assert not isinstance(app.screen, appmod.BackupScreen)
assert app.busy == set()
async def test_backup_while_guest_busy_warns_and_makes_no_second_call(fake):
started, release = threading.Event(), threading.Event()
fake.backup_hook = lambda: (started.set(), release.wait(10))
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
try:
await open_backup(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("b") # second attempt while the first still runs
await pilot.pause()
if isinstance(app.screen, appmod.BackupScreen): # 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.backup_runs == [(108, "truenas-backup")]
finally:
release.set()
await settle(app, pilot)
assert fake.backup_runs == [(108, "truenas-backup")]
async def test_backup_now_failure_notifies_error_and_releases_busy(fake):
fake.backup_error = cli_error("Error: storage 'truenas-backup' is not online")
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
await open_backup(app, pilot)
await press_button(app, pilot, "#yes")
assert any("is not online" in m for m in errors(app))
assert app.busy == set() and app.is_running
assert not any("backup done" in m.lower() for m, _ in app.seen)
# --- review findings: info screen robustness ---------------------------------
async def test_info_backup_with_missing_size_shows_dash_and_app_survives(fake):
fake.backup_list = [dict(BACKUPS[0], size=None)]
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
assert "-" in cells(app.screen.query_one("#backups", DataTable), 0)
assert app.is_running
async def test_info_uses_live_status_not_table_status(fake):
# the table status can be stale (listing lags); runtime IPs follow the live status
fake.live_status = {100: "running"} # CT 100 is "stopped" in the listing
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot, down=1)
assert fake.iface_calls == [100]
# --- backups of earlier guests with a reused VMID ------------------------------
EARLIER = {"volid": "truenas-backup:backup/vzdump-qemu-108-planka.vma.zst", "ctime": T_OLD - 60,
"size": 2**30, "format": "vma.zst", "notes": "planka", "storage": "truenas-backup"} # fmt: skip
def backup_notes(app):
t = app.screen.query_one("#backups", DataTable)
return [cells(t, i)[4] for i in range(t.row_count)]
async def test_backups_of_earlier_guest_hidden_by_default_with_hint(fake):
# VMIDs are reused: a deleted guest's backups (notes = its name) are not this guest's
fake.backup_list = [*BACKUPS, EARLIER]
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
assert "planka" not in backup_notes(app)
assert backup_notes(app) == [
"docker (manual)",
"",
] # own name and no notes stay
assert "1" in str(app.screen.query_one("#hidden", Static).render())
async def test_a_toggles_showing_earlier_guest_backups(fake):
fake.backup_list = [*BACKUPS, EARLIER]
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
await pilot.press("a")
await pilot.pause()
assert any("planka" in n for n in backup_notes(app))
await pilot.press("a")
await pilot.pause()
assert not any("planka" in n for n in backup_notes(app))
async def test_no_hint_when_nothing_hidden(fake):
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
assert str(app.screen.query_one("#hidden", Static).render()).strip() == ""
# --- delivery B: network editing ----------------------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Network editing - delivery B" > app.py)
CT_OLD = CT_CONFIG["net0"]
VM_CI = {
**VM_CONFIG,
"ide2": "local-lvm:vm-108-cloudinit,media=cdrom",
} # VM with cloud-init
CT_STATIC6 = ("name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,"
"ip6=2001:db8::5/64,gw6=2001:db8::1,type=veth") # fmt: skip
async def open_net(app, pilot, down=1):
"""Open NetScreen; default row is CT 100 (stopped), down=0 is VM 108 (running)."""
await open_info(app, pilot, down=down)
await pilot.press("n")
await settle(app, pilot)
assert isinstance(app.screen, appmod.NetScreen)
def val(app, sel):
return app.screen.query_one(f"#{sel}").value
async def setv(app, pilot, **values):
for key, value in values.items():
app.screen.query_one(f"#{key}").value = value
await pilot.pause()
async def to_info(app, pilot):
for _ in range(4):
if isinstance(app.screen, appmod.InfoScreen):
return
await pilot.press("escape")
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
async def apply_to_confirm(app, pilot):
await press_button(app, pilot, "#apply")
assert isinstance(app.screen, ConfirmScreen)
return text_of(app.screen)
async def change_and_confirm(app, pilot, **values):
await setv(app, pilot, **values)
text = await apply_to_confirm(app, pilot)
await press_button(app, pilot, "#yes")
return text
def keys_of(changes):
return {k: cli.parse_net(v) if v is not None else None for k, v in changes.items()}
async def test_n_opens_net_screen_prefilled_for_ct_sample(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
assert val(app, "iface") == "net0"
assert val(app, "bridge") == "vmbr0"
assert val(app, "tag") == ""
assert val(app, "firewall") is False
assert val(app, "ipv4") == "dhcp"
assert val(app, "ipv6") == "auto"
async def test_net_screen_prefilled_for_vm_from_ipconfig(fake):
fake.net_configs[108] = dict(VM_CI)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
assert val(app, "bridge") == "vmbr0"
assert val(app, "tag") == "20"
assert val(app, "firewall") is True
assert val(app, "ipv4") == "dhcp"
assert val(app, "ipv6") == "none"
assert not app.screen.query_one("#ipv4", Select).disabled
async def test_n_on_guest_without_nets_opens_nothing(fake):
app = make_app() # default Fake config has no netN
async with app.run_test() as pilot:
await open_info(app, pilot)
await pilot.press("n")
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
async def test_changing_iface_prefills_again(fake):
fake.bridge_list = ["vmbr0", "vmbr1"]
fake.net_configs[100] = {
**CT_CONFIG,
"net1": "name=eth1,bridge=vmbr1,hwaddr=BC:24:11:36:DE:79,tag=7,firewall=1,"
"ip=10.1.1.5/24,gw=10.1.1.1,ip6=manual,type=veth",
} # fmt: skip
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await setv(app, pilot, iface="net1")
assert val(app, "bridge") == "vmbr1" and val(app, "tag") == "7"
assert val(app, "firewall") is True
assert val(app, "ipv4") == "static"
assert (
val(app, "ipv4_cidr") == "10.1.1.5/24" and val(app, "ipv4_gw") == "10.1.1.1"
)
assert val(app, "ipv6") == "none"
async def test_vm_without_cloudinit_disables_ip_widgets_and_explains(fake):
fake.net_configs[108] = dict(VM_CONFIG) # no cloud-init drive
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
for wid in ("ipv4", "ipv4_cidr", "ipv4_gw", "ipv6", "ipv6_cidr", "ipv6_gw"):
assert app.screen.query_one(f"#{wid}").disabled, wid
assert str(app.screen.query_one("#ip_note", Static).render()).strip() != ""
assert not app.screen.query_one("#tag").disabled
async def test_vm_without_cloudinit_link_change_keeps_key_position_and_no_ipconfig(
fake,
):
fake.net_configs[108] = dict(VM_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await change_and_confirm(app, pilot, tag="30")
assert fake.set_calls == [
(108, {"net0": "virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=30,firewall=1"})
]
async def test_set_vlan_99_confirm_shows_old_and_new_then_sets_once_busy_and_reloads(
fake,
):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
busy_seen = []
fake.set_hook = lambda: busy_seen.append(set(app.busy))
new = CT_OLD + ",tag=99"
async with app.run_test() as pilot:
await open_net(app, pilot)
before = len(fake.backups_calls)
await setv(app, pilot, tag="99")
text = await apply_to_confirm(app, pilot)
assert CT_OLD in text and new in text
assert fake.set_calls == [] # nothing before "yes"
await press_button(app, pilot, "#yes")
assert fake.set_calls == [(100, {"net0": new})]
assert busy_seen and busy_seen[0]
assert app.busy == set() and errors(app) == []
assert len(fake.backups_calls) > before # InfoScreen reloaded
assert isinstance(app.screen, appmod.InfoScreen)
assert cells(app.screen.query_one("#nets", DataTable), 0)[2] == "99"
@pytest.mark.parametrize("bad", ["5000", "0", "abc"])
async def test_invalid_tag_errors_without_call_and_screen_stays(fake, bad):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await setv(app, pilot, tag=bad)
await press_button(app, pilot, "#apply")
assert errors(app) and fake.set_calls == []
assert isinstance(app.screen, appmod.NetScreen)
async def test_static_ipv4_without_prefix_errors_without_call(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await setv(app, pilot, ipv4="static", ipv4_cidr="10.0.0.5")
await press_button(app, pilot, "#apply")
assert errors(app) and fake.set_calls == []
assert isinstance(app.screen, appmod.NetScreen)
async def test_no_change_gives_info_notify_no_confirm_no_call(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await press_button(app, pilot, "#apply")
assert not isinstance(app.screen, ConfirmScreen)
assert fake.set_calls == [] and errors(app) == []
assert any(sev == "information" for _, sev in app.seen)
async def test_static_ipv4_with_gateway_keeps_hwaddr_and_other_keys(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await change_and_confirm(
app,
pilot,
ipv4="static",
ipv4_cidr="172.16.41.200/23",
ipv4_gw="172.16.40.1",
)
assert len(fake.set_calls) == 1
vmid, changes = fake.set_calls[0]
assert vmid == 100 and set(changes) == {"net0"}
assert cli.parse_net(changes["net0"]) == {
"name": "eth0", "bridge": "vmbr0", "hwaddr": "BC:24:11:36:DE:78",
"ip": "172.16.41.200/23", "gw": "172.16.40.1", "ip6": "auto", "type": "veth",
} # fmt: skip
async def test_ipv6_static_to_dhcp_removes_gw6(fake):
fake.net_configs[100] = {"net0": CT_STATIC6}
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
assert val(app, "ipv6") == "static" and val(app, "ipv6_gw") == "2001:db8::1"
await change_and_confirm(app, pilot, ipv6="dhcp")
new = cli.parse_net(fake.set_calls[0][1]["net0"])
assert new["ip6"] == "dhcp" and "gw6" not in new
assert new["hwaddr"] == "BC:24:11:36:DE:78" and new["ip"] == "dhcp"
async def test_vm_ipconfig_change_goes_into_ipconfig0_and_confirm_mentions_next_boot(
fake,
):
fake.net_configs[108] = dict(VM_CI)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await setv(
app, pilot, ipv4="static", ipv4_cidr="10.0.0.5/24", ipv4_gw="10.0.0.1"
)
text = await apply_to_confirm(app, pilot)
assert "next boot" in text.lower()
assert "ip=dhcp" in text and "10.0.0.5/24" in text # old and new raw strings
await press_button(app, pilot, "#yes")
assert [v for v, _ in fake.set_calls] == [108]
changes = fake.set_calls[0][1]
assert set(changes) == {"ipconfig0"} # net0 is unchanged and not sent
assert cli.parse_net(changes["ipconfig0"]) == {
"ip": "10.0.0.5/24",
"gw": "10.0.0.1",
}
async def test_vm_ipconfig_none_none_deletes_ipconfig0(fake):
fake.net_configs[108] = dict(VM_CI)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await change_and_confirm(app, pilot, ipv4="none") # ipv6 already none
assert fake.set_calls == [(108, {"ipconfig0": None})]
async def test_link_confirm_for_running_guest_says_applies_immediately(fake):
fake.net_configs[108] = dict(VM_CI)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await setv(app, pilot, firewall=False)
text = await apply_to_confirm(app, pilot)
assert "immediately" in text.lower()
@pytest.mark.parametrize("how", ["cancel", "escape"])
async def test_cancel_or_escape_in_net_screen_makes_no_call(fake, how):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await setv(app, pilot, tag="99")
if how == "cancel":
await press_button(app, pilot, "#cancel")
else:
await pilot.press("escape")
await settle(app, pilot)
assert not isinstance(app.screen, appmod.NetScreen)
assert fake.set_calls == [] and app.busy == set()
async def test_declining_confirm_makes_no_call(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await setv(app, pilot, tag="99")
await apply_to_confirm(app, pilot)
await press_button(app, pilot, "#no")
assert fake.set_calls == [] and app.busy == set()
async def test_apply_while_guest_busy_warns_and_makes_no_second_call(fake):
fake.net_configs[100] = dict(CT_CONFIG)
started, release = threading.Event(), threading.Event()
fake.set_hook = lambda: (started.set(), release.wait(10))
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
try:
await setv(app, pilot, tag="99")
await apply_to_confirm(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(
"escape"
) # leave whatever is on top; the guest is still busy
await pilot.pause()
await to_info(app, pilot)
await pilot.press("n")
await pilot.pause()
if isinstance(
app.screen, appmod.NetScreen
): # refusal may come at apply/yes
await setv(app, pilot, tag="55")
app.screen.query_one("#apply", Button).press()
await pilot.pause()
if isinstance(app.screen, ConfirmScreen):
app.screen.query_one("#yes", Button).press()
await pilot.pause()
assert any(sev == "warning" for _, sev in app.seen)
assert len(fake.set_calls) == 1
finally:
release.set()
await settle(app, pilot)
assert len(fake.set_calls) == 1
async def test_set_net_failure_notifies_error_releases_busy_and_stores_no_undo(fake):
fake.net_configs[100] = dict(CT_CONFIG)
fake.set_error = cli_error("Error: unable to apply net0")
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await change_and_confirm(app, pilot, tag="99")
assert any("unable to apply net0" in m for m in errors(app))
assert app.busy == set() and app.is_running
await to_info(app, pilot)
calls = len(fake.set_calls)
await pilot.press("u")
await settle(app, pilot)
assert not isinstance(app.screen, ConfirmScreen) # no undo entry was stored
assert len(fake.set_calls) == calls
# --- undo ---------------------------------------------------------------------
async def test_u_without_undo_entry_gives_info_notify_and_no_call(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot, down=1)
await pilot.press("u")
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
assert any(sev == "information" for _, sev in app.seen)
assert fake.set_calls == []
async def test_u_after_change_confirms_then_restores_old_net_string_and_clears_entry(
fake,
):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await change_and_confirm(app, pilot, tag="99")
await to_info(app, pilot)
await pilot.press("u")
await settle(app, pilot)
assert isinstance(app.screen, ConfirmScreen)
assert CT_OLD in text_of(app.screen) # what will be restored
assert len(fake.set_calls) == 1 # nothing before "yes"
await press_button(app, pilot, "#yes")
assert fake.set_calls[1] == (100, {"net0": CT_OLD})
assert app.busy == set() and errors(app) == []
assert isinstance(app.screen, appmod.InfoScreen)
assert cells(app.screen.query_one("#nets", DataTable), 0)[2] == "-"
await pilot.press("u") # entry cleared after success
await settle(app, pilot)
assert not isinstance(app.screen, ConfirmScreen) and len(fake.set_calls) == 2
async def test_undo_declined_makes_no_call_and_keeps_entry(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await change_and_confirm(app, pilot, tag="99")
await to_info(app, pilot)
await pilot.press("u")
await settle(app, pilot)
await press_button(app, pilot, "#no")
assert len(fake.set_calls) == 1
await pilot.press("u")
await settle(app, pilot)
assert isinstance(app.screen, ConfirmScreen) # entry still there
async def test_undo_of_key_that_did_not_exist_deletes_it(fake):
ci_no_ipconfig = {k: v for k, v in VM_CI.items() if k != "ipconfig0"}
fake.net_configs[108] = ci_no_ipconfig
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await change_and_confirm(app, pilot, ipv4="dhcp")
assert fake.set_calls == [(108, {"ipconfig0": "ip=dhcp"})]
await to_info(app, pilot)
await pilot.press("u")
await settle(app, pilot)
await press_button(app, pilot, "#yes")
assert fake.set_calls[1] == (108, {"ipconfig0": None})
async def test_undo_restores_both_net_and_ipconfig_of_one_change(fake):
fake.net_configs[108] = dict(VM_CI)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await change_and_confirm(app, pilot, tag="30", ipv4="none")
assert set(fake.set_calls[0][1]) == {"net0", "ipconfig0"}
await to_info(app, pilot)
await pilot.press("u")
await settle(app, pilot)
await press_button(app, pilot, "#yes")
assert fake.set_calls[1] == (
108,
{"net0": VM_CONFIG["net0"], "ipconfig0": VM_CONFIG["ipconfig0"]},
)
async def test_undo_entry_is_per_guest(fake):
fake.net_configs[100] = dict(CT_CONFIG)
fake.net_configs[108] = dict(VM_CI)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot) # CT 100
await change_and_confirm(app, pilot, tag="99")
await pilot.press("escape") # close InfoScreen
await settle(app, pilot)
await pilot.press("up")
await pilot.press("i") # VM 108: no change was made there
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
await pilot.press("u")
await settle(app, pilot)
assert not isinstance(app.screen, ConfirmScreen)
# --- review findings: network editing -------------------------------------------
async def test_context_switch_clears_undo_entries(fake):
# VMIDs repeat across clusters: an undo from one cluster must never hit the other
app = make_app()
async with app.run_test() as pilot:
await settle(app, pilot)
app.undo[108] = {"net0": "virtio=02:24:A6:49:A8:25,bridge=vmbr0"}
await switch_to_edge(app, pilot)
await settle(app, pilot)
assert app.undo == {}
async def test_untouched_fields_are_not_rewritten(fake):
# ip6=manual is left alone when only the VLAN changes; firewall=0 == absent (spec: off removes it)
fake.net_configs[100] = {
"net0": "name=eth0,bridge=vmbr0,firewall=0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=manual,type=veth"
}
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot)
await press_button(app, pilot, "#apply")
assert not isinstance(
app.screen, ConfirmScreen
) # nothing touched -> "no change"
await setv(app, pilot, tag="99")
await press_button(app, pilot, "#apply")
await press_button(app, pilot, "#yes")
assert fake.set_calls[-1][1] == {
"net0": "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=manual,type=veth,tag=99"
}
# --- live finding: a running CT does not follow IP mode changes -------------------
async def test_ip_change_on_running_ct_warns_reboot_may_be_needed(fake):
# live on CT 104: static -> dhcp left the CT without IPv4 until a reboot
fake.guests = [RUNNING_CT]
fake.net_configs[104] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await setv(app, pilot, ipv4="static", ipv4_cidr="172.16.41.200/23")
text = await apply_to_confirm(app, pilot)
assert "reboot" in text.lower()
async def test_link_only_change_on_running_ct_does_not_mention_reboot(fake):
fake.guests = [RUNNING_CT]
fake.net_configs[104] = dict(CT_CONFIG)
app = make_app()
async with app.run_test() as pilot:
await open_net(app, pilot, down=0)
await setv(app, pilot, tag="99")
text = await apply_to_confirm(app, pilot)
assert "reboot" not in text.lower()
# --- info screen layout: backups and network clearly separated --------------------
async def test_info_has_separate_titled_backup_and_network_sections(fake):
app = make_app()
async with app.run_test() as pilot:
await open_info(app, pilot)
backup = app.screen.query_one("#backup_section")
network = app.screen.query_one("#network_section")
assert "Backups" in str(backup.border_title)
assert "Network" in str(network.border_title)
assert {w.id for w in backup.query("*")} >= {"jobs", "backups", "hidden"}
assert {w.id for w in network.query("*")} >= {
"nets",
"unavailable",
"addresses",
}
# --- boxed design: every dialog a titled box, main table in a titled frame ----------
from textual.content import Content
from textual.widgets import Label
SIZE = (100, 40)
def visible(app, widget):
r = widget.region
return (
r.width >= 3
and r.height >= 1
and r.right <= app.size.width
and r.bottom <= app.size.height
)
def label_texts(screen):
return {str(w.render()).strip() for w in screen.query(Label)}
async def test_net_screen_shows_every_field_inside_the_screen(fake):
# bug: in the old rows Select got width 1 and Input 100 %, pushing fields off-screen
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_net(app, pilot)
for wid in ("iface", "bridge", "tag", "firewall", "ipv4", "ipv4_cidr", "ipv4_gw",
"ipv6", "ipv6_cidr", "ipv6_gw", "apply", "cancel"): # fmt: skip
assert visible(app, app.screen.query_one(f"#{wid}")), wid
async def test_net_screen_labels_and_ip_sections(fake):
fake.net_configs[100] = dict(CT_CONFIG)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_net(app, pilot)
assert {"Interface", "Bridge", "VLAN"} <= label_texts(app.screen)
assert "IPv4" in str(app.screen.query_one("#ipv4_section").border_title)
assert "IPv6" in str(app.screen.query_one("#ipv6_section").border_title)
async def test_edit_screen_fields_have_labels(fake):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_edit(app, pilot)
assert {"Cores", "Memory (MiB)", "Disk", "Grow by (GiB)"} <= label_texts(
app.screen
)
async def open_each_dialog(app, pilot):
"""Yields after each dialog is open; closes it with escape afterwards."""
await settle(app, pilot)
for keys in (["S"], ["e"], ["c"], ["i", "b"], ["i", "n"]):
for k in keys:
await pilot.press(k)
await settle(app, pilot)
yield keys
for _ in keys:
await pilot.press("escape")
await settle(app, pilot)
async def test_every_dialog_is_a_titled_centred_box(fake):
fake.net_configs[108] = dict(VM_CONFIG)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
async for keys in open_each_dialog(app, pilot):
box = app.screen.query_one("#dialog")
assert str(box.border_title).strip(), keys
assert box.region.width < app.size.width, keys # a box, not full screen
assert box.region.x > 0, keys # centred
async def test_main_table_is_framed_with_title_and_context(fake):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await settle(app, pilot)
frame = app.query_one("#guests_section")
assert "Guests" in str(frame.border_title)
assert "mainframe" in str(frame.border_subtitle)
@pytest.mark.parametrize("keys", [["e"], ["i", "b"], ["i", "n"]])
async def test_guest_name_with_markup_characters_does_not_break_dialogs(fake, keys):
# border titles are parsed as markup; a name like "a[/b]" must be shown literally
fake.guests = [dict(VM, name="a[/b]")]
fake.net_configs[108] = dict(VM_CONFIG)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await settle(app, pilot)
for k in keys:
await pilot.press(k)
await settle(app, pilot)
title = app.screen.query_one("#dialog").border_title
assert (
"a[/b]" in Content.from_markup(title).plain
) # what is shown, not how it is stored
assert app.is_running
Reference in new issue View Git Blame Copy Permalink