DuckbootandClaude Opus 5.5
62b79ab29e
Press c to list cv4pve-cli contexts and switch between them. The active context is always shown in the header. - Switching runs `cv4pve-cli config use`, which changes the global current context. cv4pve-cli 2.4.0 has no per-call context option. - `config list` has no JSON output, so its text lines are parsed. - The guest table is cleared as soon as another context is chosen, so the old cluster's guests cannot be acted on. - A generation counter drops guest listings that started before a switch, and g and c are refused while a switch runs. Without this, a reload already in flight could refill the table with the old cluster's guests, and VMIDs collide across clusters. 103 tests pass. The switch was tested live from mainframe to mainframe-edge and back. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
422 lines
12 KiB
Python
422 lines
12 KiB
Python
"""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",
|
|
]
|
|
|
|
|
|
# --- contexts / use_context (spec: "Context switching" > cli.py) -------------
|
|
|
|
CONTEXT_LIST = (
|
|
"* mainframe mainframe.iot.home.arpa:8006 (api-token)\n"
|
|
" mainframe-edge mainframe-edge.home.arpa:8006 (api-token)\n"
|
|
)
|
|
|
|
|
|
def test_contexts_parses_current_name_host_auth(fake):
|
|
fake.result = (CONTEXT_LIST, "", 0)
|
|
assert cli.contexts() == [
|
|
{"name": "mainframe", "host": "mainframe.iot.home.arpa:8006",
|
|
"auth": "api-token", "current": True},
|
|
{"name": "mainframe-edge", "host": "mainframe-edge.home.arpa:8006",
|
|
"auth": "api-token", "current": False},
|
|
] # fmt: skip
|
|
|
|
|
|
def test_contexts_runs_config_list(fake):
|
|
fake.result = (CONTEXT_LIST, "", 0)
|
|
cli.contexts()
|
|
assert fake.argv == ["cv4pve-cli", "config", "list"]
|
|
|
|
|
|
def test_contexts_ignores_junk_and_empty_lines(fake):
|
|
fake.result = (
|
|
"\nContexts:\n" + CONTEXT_LIST + "\n \nnot a context line\n",
|
|
"",
|
|
0,
|
|
)
|
|
assert [c["name"] for c in cli.contexts()] == ["mainframe", "mainframe-edge"]
|
|
|
|
|
|
def test_contexts_empty_output_gives_empty_list(fake):
|
|
assert cli.contexts() == []
|
|
|
|
|
|
def test_use_context_runs_config_use_with_name(fake):
|
|
cli.use_context("mainframe-edge")
|
|
assert fake.argv == ["cv4pve-cli", "config", "use", "mainframe-edge"]
|
|
|
|
|
|
def test_use_context_unknown_name_raises_clierror(monkeypatch):
|
|
FakeRun(monkeypatch, "Error: Context 'x' not found.\n", returncode=3)
|
|
with pytest.raises(cli.CliError) as exc:
|
|
cli.use_context("x")
|
|
assert "Context 'x' not found" in str(exc.value)
|