DuckbootandClaude Opus 5.5
62490af913
Put backups and network in their own bordered sections titled "Backups" and "Network". Inside Network, the configured settings and the addresses in use are labelled "Configured" and "In use". The "unavailable" note is hidden when it is empty, so it no longer leaves a blank line. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1886 lines
66 KiB
Python
1886 lines
66 KiB
Python
"""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",
|
|
}
|