Duckboot/cv4pve-cli-tui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
32809b6743cfd7292b25a13ffa87011be2323995
cv4pve-cli-tui/tests/test_cli.py
T
DuckbootandClaude Opus 5.5 32809b6743 Add guest info screen with backups, backup now and network 
Press i on a guest to see its backups and network. Backups from every
backup-capable storage are listed newest first, with the backup jobs
that cover the guest. The network part shows the netN settings
(bridge, VLAN, MAC, IPv4/IPv6 setting, firewall) and the addresses in
use.

Press b in the info screen to back the guest up now: snapshot mode and
zstd, like the scheduled job, to a storage you pick (the job's storage
is preselected). It asks for confirmation, marks the guest busy and
uses --wait.

Fixed after review and a read-only live check:
- prune-backups arrives as a JSON string; parsing it as a dict
  crashed the app.
- CT addresses come from ip-addresses. inet6 holds only the
  link-local address, so the global SLAAC address was missing.
- The info screen uses the guest's live status, and odd API data
  (null size, unknown address types) no longer kills the app.
- The test fake now copies guest dicts; a shared constant was being
  mutated across tests.

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

686 lines
21 KiB
Python
Raw Blame History

"""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)
# --- delivery A: backups / jobs / net / interfaces ---------------------------
# spec: docs/plans/guest-tui-mvp.md ("Guest info screen ... delivery A" > cli.py)
CT_NET = "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=auto,type=veth"
VM_NET = "virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,firewall=1"
VMID108 = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu"}
def storage(name, content="backup", enabled=1, active=1):
return {"storage": name, "content": content, "enabled": enabled, "active": active,
"type": "dir", "avail": 10**9} # fmt: skip
def test_backup_storages_filters_content_enabled_active_keeps_order(fake):
fake.result = (
json.dumps([
storage("zeta", "iso,backup,vztmpl"),
storage("nobackup", "iso,snippets,import"),
storage("off", enabled=0),
storage("inactive", active=0),
storage("alpha", "backup"),
]),
"",
0,
) # fmt: skip
assert cli.backup_storages(VM) == ["zeta", "alpha"]
assert fake.argv == [
"cv4pve-cli",
"get",
"node",
"storages",
"mainframe",
"-o",
"json",
]
def test_backup_jobs_argv_and_returns_list(fake):
jobs = [{"id": "backup-1", "enabled": 1, "storage": "truenas-backup", "all": 1}]
fake.result = (json.dumps(jobs), "", 0)
assert cli.backup_jobs() == jobs
assert fake.argv == ["cv4pve-cli", "get", "cluster", "backup-jobs", "-o", "json"]
def job(storage="truenas-backup", enabled=1, all=0, vmid=None, exclude=None, node=None):
return {"id": "j", "enabled": enabled, "storage": storage, "all": all, "vmid": vmid,
"exclude": exclude, "node": node} # fmt: skip
@pytest.mark.parametrize(
"jobs,expected",
[
([job(all=1)], "truenas-backup"), # all=1
([job(all=1, exclude="108")], None), # excluded
([job(all=1, exclude="100,108,200")], None), # excluded in a list
([job(all=1, exclude="1080")], "truenas-backup"), # not a substring match
([job(vmid="100,108,200")], "truenas-backup"), # vmid comma string
([job(vmid="1080,200")], None), # not a substring match
([job(vmid="100,200")], None), # not listed
([job(all=1, node="othernode")], None), # node mismatch
([job(all=1, node="mainframe")], "truenas-backup"), # node match
([job(all=1, enabled=0)], None), # disabled
([job(vmid=None, all=0)], None), # covers nothing
([], None), # no job
(
[job("first", enabled=0, all=1), job("second", all=1), job("third", all=1)],
"second",
),
],
)
def test_job_storage(jobs, expected):
assert cli.job_storage(jobs, VMID108) == expected
def test_backups_merges_storages_newest_first_and_adds_storage(monkeypatch):
per_storage = {
"truenas-backup": [
{
"volid": "truenas-backup:backup/vzdump-qemu-108-a.vma.zst",
"ctime": 100,
"size": 1,
},
{
"volid": "truenas-backup:backup/vzdump-qemu-108-b.vma.zst",
"ctime": 300,
"size": 3,
},
],
"local": [
{"volid": "local:backup/vzdump-qemu-108-c.vma.zst", "ctime": 200, "size": 2}
],
}
calls = []
def run(argv, **kwargs):
calls.append(list(argv))
return subprocess.CompletedProcess(
argv, 0, json.dumps(per_storage[argv[5]]), ""
)
monkeypatch.setattr(cli.subprocess, "run", run)
result = cli.backups(VMID108, ["truenas-backup", "local"])
assert [(b["ctime"], b["storage"]) for b in result] == [
(300, "truenas-backup"), (200, "local"), (100, "truenas-backup"),
] # fmt: skip
assert calls == [
["cv4pve-cli", "get", "guest", "backups", "mainframe", s, "108", "-o", "json"]
for s in ("truenas-backup", "local")
]
def test_backup_now_exact_argv_without_yes(fake):
cli.backup_now(VMID108, "truenas-backup")
assert fake.argv == [
"cv4pve-cli", "do", "backup", "guest", "mainframe", "108", "truenas-backup",
"--mode", "snapshot", "--compress", "zstd",
"--notes-template", "{{guestname}} (manual)", "--wait",
] # fmt: skip
assert "-y" not in fake.argv and "--yes" not in fake.argv
def test_backup_now_failure_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: no space", returncode=5)
with pytest.raises(cli.CliError):
cli.backup_now(VMID108, "local")
def test_parse_net_ct_sample_keeps_order():
net = cli.parse_net(CT_NET)
assert list(net.items()) == [
("name", "eth0"), ("bridge", "vmbr0"), ("hwaddr", "BC:24:11:36:DE:78"),
("ip", "dhcp"), ("ip6", "auto"), ("type", "veth"),
] # fmt: skip
def test_parse_net_vm_sample_first_item_is_model_mac():
net = cli.parse_net(VM_NET)
assert list(net.items()) == [
("virtio", "02:24:A6:49:A8:25"), ("bridge", "vmbr0"), ("tag", "20"), ("firewall", "1"),
] # fmt: skip
def test_nets_only_netN_keys_sorted_numerically():
config = {
"cores": 4,
"net10": VM_NET,
"net2": CT_NET,
"ipconfig0": "ip=dhcp",
"net0": VM_NET,
}
result = cli.nets(config)
assert list(result) == ["net0", "net2", "net10"]
assert result["net2"] == cli.parse_net(CT_NET)
def test_interfaces_ct_inet_inet6_and_excludes_lo(fake):
fake.result = (
json.dumps([
{"name": "lo", "hwaddr": "00:00:00:00:00:00", "inet": "127.0.0.1/8", "inet6": "::1/128"},
{"name": "eth0", "hwaddr": "BC:24:11:36:DE:78", "inet": "172.16.41.200/23",
"inet6": "fe80::be24:11ff:fe36:de78/64", "ip-addresses": "[]"},
{"name": "eth1", "hwaddr": "BC:24:11:36:DE:79", "inet": "10.0.0.5/24", "ip-addresses": "[]"},
]),
"",
0,
) # fmt: skip
assert cli.interfaces(CT) == [
{"name": "eth0", "mac": "BC:24:11:36:DE:78", "ipv4": ["172.16.41.200/23"],
"ipv6": ["fe80::be24:11ff:fe36:de78/64"]},
{"name": "eth1", "mac": "BC:24:11:36:DE:79", "ipv4": ["10.0.0.5/24"], "ipv6": []},
] # fmt: skip
assert fake.argv == [
"cv4pve-cli",
"get",
"ct",
"interfaces",
"mainframe",
"100",
"-o",
"json",
]
def test_interfaces_vm_agent_result_json_string_and_excludes_lo(fake):
agent = [
{"name": "lo", "hardware-address": "00:00:00:00:00:00",
"ip-addresses": [{"ip-address": "127.0.0.1", "prefix": 8, "ip-address-type": "ipv4"}]},
{"name": "ens18", "hardware-address": "02:24:A6:49:A8:25",
"ip-addresses": [
{"ip-address": "172.16.41.50", "prefix": 23, "ip-address-type": "ipv4"},
{"ip-address": "fe80::24:a6ff:fe49:a825", "prefix": 64, "ip-address-type": "ipv6"},
{"ip-address": "2001:db8::5", "prefix": 64, "ip-address-type": "ipv6"},
]},
] # fmt: skip
fake.result = (json.dumps([{"key": "result", "value": json.dumps(agent)}]), "", 0)
assert cli.interfaces(VMID108) == [
{"name": "ens18", "mac": "02:24:A6:49:A8:25", "ipv4": ["172.16.41.50/23"],
"ipv6": ["fe80::24:a6ff:fe49:a825/64", "2001:db8::5/64"]},
] # fmt: skip
assert fake.argv == [
"cv4pve-cli",
"get",
"vm",
"agent",
"network",
"mainframe",
"108",
"-o",
"json",
]
def test_interfaces_vm_agent_failure_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: QEMU guest agent is not running", returncode=5)
with pytest.raises(cli.CliError):
cli.interfaces(VMID108)
def test_backup_jobs_parses_prune_backups_json_string(fake):
# real cv4pve-cli output: prune-backups is a JSON *string*
fake.result = (
json.dumps(
[
{"id": "j", "prune-backups": '{"keep-last":"3"}'},
{"id": "k", "prune-backups": None},
]
),
"",
0,
)
jobs = cli.backup_jobs()
assert jobs[0]["prune-backups"] == {"keep-last": "3"}
assert jobs[1]["prune-backups"] is None
def test_interfaces_vm_tolerates_null_addresses_and_unknown_types(fake):
result = [
{"name": "eth0", "hardware-address": "aa", "ip-addresses": None},
{"name": "eth1", "ip-addresses": [{"ip-address-type": "ipv5", "ip-address": "x", "prefix": 1},
{"ip-address-type": "ipv4", "ip-address": "10.0.0.2", "prefix": 24}]},
] # fmt: skip
fake.result = (json.dumps([{"key": "result", "value": json.dumps(result)}]), "", 0)
found = cli.interfaces({"vmid": 108, "node": "mainframe", "type": "qemu"})
assert found == [
{"name": "eth0", "mac": "aa", "ipv4": [], "ipv6": []},
{"name": "eth1", "mac": None, "ipv4": ["10.0.0.2/24"], "ipv6": []},
]
def test_interfaces_ct_lists_all_addresses_incl_global_ipv6(fake):
# real output: inet6 holds only one address (link-local); ip-addresses (JSON string) has all
addrs = [
{"prefix": "23", "ip-address": "172.16.41.200", "ip-address-type": "inet"},
{"ip-address-type": "inet6", "ip-address": "2a02:fe1:e77e:6004:be24:11ff:fe36:de78", "prefix": "64"},
{"prefix": "64", "ip-address": "fe80::be24:11ff:fe36:de78", "ip-address-type": "inet6"},
] # fmt: skip
eth0 = {"name": "eth0", "hwaddr": "bc:24:11:36:de:78", "inet": "172.16.41.200/23",
"inet6": "fe80::be24:11ff:fe36:de78/64", "ip-addresses": json.dumps(addrs)} # fmt: skip
fake.result = (json.dumps([eth0]), "", 0)
found = cli.interfaces({"vmid": 104, "node": "mainframe", "type": "lxc"})
assert found == [{"name": "eth0", "mac": "bc:24:11:36:de:78", "ipv4": ["172.16.41.200/23"],
"ipv6": ["2a02:fe1:e77e:6004:be24:11ff:fe36:de78/64", "fe80::be24:11ff:fe36:de78/64"]}] # fmt: skip
Reference in new issue View Git Blame Copy Permalink