"""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()