Add Textual TUI for managing Proxmox guests via cv4pve-cli
List VMs and LXC containers from the current cv4pve-cli context and start, shutdown, stop and reboot them, change cores and memory, and grow disks. The app shells out to `cv4pve-cli -o json` and stores no credentials. - Every mutating call uses --wait: without it cv4pve-cli exits 0 even when the task fails. - shutdown, stop and reboot need confirmation; disk growth is increment-only and asks again because it cannot be undone. - After an action the acted-on guest's status is read from status/current, because /cluster/resources lags a few seconds. - Notifies when a cores/memory change is pending until reboot. Tests mock cv4pve-cli (85 tests). Live-tested edit, shutdown, start and reboot on CT 104 (mainframe), restored afterwards; disk resize was not run live. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DuckbootandClaude Opus 5.5 committed
1 parent
6f009c6782
commit
f51aff3db5
10 files changed
+1612
No files matched your search
@@ -0,0 +1,449 @@
|
||||
"""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 subprocess
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
from textual.screen import ModalScreen
|
||||
from textual.widgets import Button, DataTable, Input, Select, Static
|
||||
|
||||
from cv4pve_tui import cli
|
||||
from cv4pve_tui.app import ConfirmScreen, 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 = {}, []
|
||||
|
||||
def list_guests(self):
|
||||
self.list_calls += 1
|
||||
return list(self.guests)
|
||||
|
||||
def guest_config(self, guest):
|
||||
return {"cores": guest["maxcpu"]}
|
||||
|
||||
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"])
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake(monkeypatch):
|
||||
f = Fake()
|
||||
for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
|
||||
"resize_disk", "pending_keys", "status"): # 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")
|
||||
]
|
||||
@@ -0,0 +1,372 @@
|
||||
"""Tests for cv4pve_tui.cli; spec: docs/plans/guest-tui-mvp.md ("Module cv4pve_tui/cli.py").
|
||||
|
||||
subprocess.run is always mocked: cv4pve-cli is never executed.
|
||||
"""
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
from cv4pve_tui import cli
|
||||
|
||||
VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu"}
|
||||
CT = {"vmid": 100, "name": "backrest", "node": "mainframe", "type": "lxc"}
|
||||
|
||||
|
||||
class FakeRun:
|
||||
def __init__(self, monkeypatch, stdout="", stderr="", returncode=0):
|
||||
self.calls = []
|
||||
self.result = (stdout, stderr, returncode)
|
||||
monkeypatch.setattr(cli.subprocess, "run", self)
|
||||
|
||||
def __call__(self, argv, **kwargs):
|
||||
self.calls.append((list(argv), kwargs))
|
||||
out, err, rc = self.result
|
||||
return subprocess.CompletedProcess(argv, rc, out, err)
|
||||
|
||||
@property
|
||||
def argv(self):
|
||||
assert len(self.calls) == 1
|
||||
return self.calls[0][0]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake(monkeypatch):
|
||||
return FakeRun(monkeypatch)
|
||||
|
||||
|
||||
def error_for(monkeypatch, stdout, stderr="", returncode=1):
|
||||
FakeRun(monkeypatch, stdout, stderr, returncode)
|
||||
with pytest.raises(cli.CliError) as exc:
|
||||
cli.run("do", "start")
|
||||
return exc.value
|
||||
|
||||
|
||||
# --- run / CliError ---------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_passes_exact_subprocess_arguments(fake):
|
||||
fake.result = ("ok\n", "", 0)
|
||||
assert cli.run("get", "guests").strip() == "ok"
|
||||
argv, kwargs = fake.calls[0]
|
||||
assert argv == ["cv4pve-cli", "get", "guests"]
|
||||
assert kwargs["capture_output"] is True
|
||||
assert kwargs["text"] is True
|
||||
assert kwargs["stdin"] == subprocess.DEVNULL
|
||||
assert kwargs["timeout"] == 330
|
||||
|
||||
|
||||
def test_run_nonzero_exit_raises_clierror_with_stripped_output(monkeypatch):
|
||||
err = error_for(monkeypatch, " Error: boom\n\n", returncode=5)
|
||||
assert str(err) == "Error: boom"
|
||||
|
||||
|
||||
def test_clierror_includes_stderr_when_stdout_is_empty(monkeypatch):
|
||||
err = error_for(monkeypatch, "", stderr="Error: only on stderr\n")
|
||||
assert "only on stderr" in str(err)
|
||||
|
||||
|
||||
def test_summary_multiline_sync_failure(monkeypatch):
|
||||
out = (
|
||||
"Error: Proxmox VE refused the call (HTTP 500):\n"
|
||||
" Configuration file 'nodes/mainframe/qemu-server/999.conf' does not exist\n"
|
||||
"\n"
|
||||
"Call: POST /nodes/mainframe/qemu/999/status/start\n"
|
||||
"Alias: prod\n"
|
||||
)
|
||||
summary = error_for(monkeypatch, out).summary
|
||||
assert "\n" not in summary
|
||||
assert summary.startswith("Proxmox VE refused the call (HTTP 500):")
|
||||
assert "Error:" not in summary
|
||||
assert "Configuration file" in summary and "does not exist" in summary
|
||||
assert "Call:" not in summary and "Alias:" not in summary
|
||||
|
||||
|
||||
def test_summary_skips_upid_line_exit_5(monkeypatch):
|
||||
out = "UPID:mainframe:0001:0002:0003:qmstart:999:root@pam:\nConfiguration file 'x' does not exist\n"
|
||||
summary = error_for(monkeypatch, out, returncode=5).summary
|
||||
assert summary == "Configuration file 'x' does not exist"
|
||||
|
||||
|
||||
def test_summary_single_line_strips_error_prefix(monkeypatch):
|
||||
assert error_for(monkeypatch, "Error: no such guest\n").summary == "no such guest"
|
||||
|
||||
|
||||
# --- kind / list_guests / guest_config --------------------------------------
|
||||
|
||||
|
||||
def test_kind_maps_type():
|
||||
assert cli.kind({"type": "qemu"}) == "vm"
|
||||
assert cli.kind({"type": "lxc"}) == "ct"
|
||||
|
||||
|
||||
def test_list_guests_argv_and_sorted_by_vmid(fake):
|
||||
fake.result = (json.dumps([CT | {"vmid": 200}, VM, CT]), "", 0)
|
||||
guests = cli.list_guests()
|
||||
assert fake.argv == ["cv4pve-cli", "get", "guests", "-o", "json"]
|
||||
assert [g["vmid"] for g in guests] == [100, 108, 200]
|
||||
|
||||
|
||||
def test_guest_config_flattens_key_value_list(fake):
|
||||
fake.result = (
|
||||
json.dumps([{"key": "cores", "value": 4}, {"key": "memory", "value": 8192}]),
|
||||
"",
|
||||
0,
|
||||
)
|
||||
assert cli.guest_config(VM) == {"cores": 4, "memory": 8192}
|
||||
assert fake.argv == ["cv4pve-cli", "show", "vm", "mainframe", "108", "-o", "json"]
|
||||
|
||||
|
||||
def test_guest_config_uses_ct_for_lxc(fake):
|
||||
fake.result = ("[]", "", 0)
|
||||
cli.guest_config(CT)
|
||||
assert fake.argv == ["cv4pve-cli", "show", "ct", "mainframe", "100", "-o", "json"]
|
||||
|
||||
|
||||
# --- disks ------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_disks_vm_keeps_sized_disks_only():
|
||||
config = {
|
||||
"cores": 4,
|
||||
"memory": 8192,
|
||||
"scsi0": "local-lvm:vm-108-disk-0,discard=on,size=10G,ssd=1",
|
||||
"virtio1": "local-lvm:vm-108-disk-2,size=32G",
|
||||
"ide2": "local-lvm:vm-108-cloudinit,media=cdrom",
|
||||
"sata0": "local-lvm:vm-108-cloudinit,size=4M", # cloud-init drive without media=cdrom
|
||||
"ide3": "none,media=cdrom",
|
||||
"efidisk0": "local-lvm:vm-108-disk-1,efitype=4m,size=4M",
|
||||
"net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0",
|
||||
}
|
||||
assert cli.disks(config, "vm") == {"scsi0": "10G", "virtio1": "32G"}
|
||||
|
||||
|
||||
def test_disks_ct_rootfs_and_mount_points():
|
||||
config = {
|
||||
"cores": 1,
|
||||
"memory": 512,
|
||||
"rootfs": "local-lvm:vm-100-disk-0,size=8G",
|
||||
"mp0": "local-lvm:vm-100-disk-1,mp=/data,size=2G",
|
||||
"net0": "name=eth0,bridge=vmbr0",
|
||||
}
|
||||
assert cli.disks(config, "ct") == {"rootfs": "8G", "mp0": "2G"}
|
||||
|
||||
|
||||
def test_disks_without_disks_is_empty():
|
||||
assert cli.disks({"cores": 1}, "vm") == {}
|
||||
assert cli.disks({"cores": 1}, "ct") == {}
|
||||
|
||||
|
||||
# --- action -----------------------------------------------------------------
|
||||
|
||||
|
||||
def test_actions_constant():
|
||||
assert cli.ACTIONS == ("start", "shutdown", "stop", "reboot")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", cli.ACTIONS)
|
||||
def test_action_argv_vm(fake, name):
|
||||
cli.action(name, VM)
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"do",
|
||||
name,
|
||||
"vm",
|
||||
"mainframe",
|
||||
"108",
|
||||
"-y",
|
||||
"--wait",
|
||||
]
|
||||
|
||||
|
||||
def test_action_argv_ct(fake):
|
||||
cli.action("stop", CT)
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"do",
|
||||
"stop",
|
||||
"ct",
|
||||
"mainframe",
|
||||
"100",
|
||||
"-y",
|
||||
"--wait",
|
||||
]
|
||||
|
||||
|
||||
def test_action_unknown_name_raises_and_does_not_call(fake):
|
||||
with pytest.raises(ValueError):
|
||||
cli.action("destroy", VM)
|
||||
assert fake.calls == []
|
||||
|
||||
|
||||
def test_action_failure_raises_clierror(monkeypatch):
|
||||
FakeRun(monkeypatch, "UPID:x:\nstart failed\n", returncode=5)
|
||||
with pytest.raises(cli.CliError):
|
||||
cli.action("start", VM)
|
||||
|
||||
|
||||
# --- set_resources ----------------------------------------------------------
|
||||
|
||||
|
||||
def test_set_resources_cores_and_memory(fake):
|
||||
cli.set_resources(VM, cores=4, memory_mb=8192)
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli", "set", "vm", "config", "mainframe", "108",
|
||||
"--cores", "4", "--memory", "8192", "--wait",
|
||||
] # fmt: skip
|
||||
|
||||
|
||||
def test_set_resources_only_given_fields(fake):
|
||||
cli.set_resources(CT, memory_mb=1024)
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"set",
|
||||
"ct",
|
||||
"config",
|
||||
"mainframe",
|
||||
"100",
|
||||
"--memory",
|
||||
"1024",
|
||||
"--wait",
|
||||
]
|
||||
fake.calls.clear()
|
||||
cli.set_resources(CT, cores=2)
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"set",
|
||||
"ct",
|
||||
"config",
|
||||
"mainframe",
|
||||
"100",
|
||||
"--cores",
|
||||
"2",
|
||||
"--wait",
|
||||
]
|
||||
|
||||
|
||||
def test_set_resources_nothing_to_set(fake):
|
||||
with pytest.raises(ValueError):
|
||||
cli.set_resources(VM)
|
||||
assert fake.calls == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"kwargs", [{"cores": 0}, {"cores": -1}, {"memory_mb": 0}, {"memory_mb": -512}]
|
||||
)
|
||||
def test_set_resources_rejects_non_positive(fake, kwargs):
|
||||
with pytest.raises(ValueError):
|
||||
cli.set_resources(VM, **kwargs)
|
||||
assert fake.calls == []
|
||||
|
||||
|
||||
# --- resize_disk ------------------------------------------------------------
|
||||
|
||||
|
||||
def test_resize_disk_argv(fake):
|
||||
cli.resize_disk(VM, "scsi0", 5)
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"do",
|
||||
"resize",
|
||||
"vm",
|
||||
"disk",
|
||||
"mainframe",
|
||||
"108",
|
||||
"scsi0",
|
||||
"+5G",
|
||||
"--wait",
|
||||
]
|
||||
|
||||
|
||||
def test_resize_disk_ct_argv(fake):
|
||||
cli.resize_disk(CT, "rootfs", 2)
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"do",
|
||||
"resize",
|
||||
"ct",
|
||||
"disk",
|
||||
"mainframe",
|
||||
"100",
|
||||
"rootfs",
|
||||
"+2G",
|
||||
"--wait",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("grow", [0, -1, -10])
|
||||
def test_resize_disk_growth_must_be_positive(fake, grow):
|
||||
with pytest.raises(ValueError):
|
||||
cli.resize_disk(VM, "scsi0", grow)
|
||||
assert fake.calls == []
|
||||
|
||||
|
||||
# --- pending_keys -----------------------------------------------------------
|
||||
|
||||
|
||||
def test_pending_keys_only_non_null_pending(fake):
|
||||
fake.result = (
|
||||
json.dumps([
|
||||
{"key": "cores", "value": 4, "pending": None, "delete": None},
|
||||
{"key": "memory", "value": 8192, "pending": 16384},
|
||||
{"key": "balloon", "value": 0, "pending": 0},
|
||||
]),
|
||||
"",
|
||||
0,
|
||||
) # fmt: skip
|
||||
assert sorted(cli.pending_keys(VM)) == ["balloon", "memory"]
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"api",
|
||||
"get",
|
||||
"/nodes/mainframe/qemu/108/pending",
|
||||
"-o",
|
||||
"json",
|
||||
]
|
||||
|
||||
|
||||
def test_pending_keys_ct_path_and_none_pending(fake):
|
||||
fake.result = (json.dumps([{"key": "cores", "value": 1, "pending": None}]), "", 0)
|
||||
assert cli.pending_keys(CT) == []
|
||||
assert fake.argv[3] == "/nodes/mainframe/lxc/100/pending"
|
||||
|
||||
|
||||
def test_missing_binary_or_timeout_becomes_cli_error(monkeypatch):
|
||||
for exc in (
|
||||
FileNotFoundError("cv4pve-cli"),
|
||||
subprocess.TimeoutExpired("cv4pve-cli", 330),
|
||||
):
|
||||
|
||||
def boom(*a, _exc=exc, **k):
|
||||
raise _exc
|
||||
|
||||
monkeypatch.setattr(subprocess, "run", boom)
|
||||
with pytest.raises(cli.CliError):
|
||||
cli.run("get", "version")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", [{}, {"name": None}, {"name": ""}])
|
||||
def test_list_guests_fills_missing_name(fake, raw):
|
||||
guest = {k: v for k, v in VM.items() if k != "name"} | raw
|
||||
fake.result = (json.dumps([guest]), "", 0)
|
||||
assert cli.list_guests()[0]["name"] == "VM 108"
|
||||
|
||||
|
||||
def test_status_reads_live_status_current(fake):
|
||||
fake.result = (
|
||||
json.dumps(
|
||||
[{"key": "cpus", "value": 1}, {"key": "status", "value": "stopped"}]
|
||||
),
|
||||
"",
|
||||
0,
|
||||
)
|
||||
assert cli.status(CT) == "stopped"
|
||||
assert fake.argv == [
|
||||
"cv4pve-cli",
|
||||
"api",
|
||||
"get",
|
||||
"/nodes/mainframe/lxc/100/status/current",
|
||||
"-o",
|
||||
"json",
|
||||
]
|
||||
Reference in new issue
Block a user