Duckboot/cv4pve-cli-tui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
cv4pve-cli-tui/tests/test_cli.py
T
DuckbootandClaude Opus 5.5 82902edfc8 Restore a container backup with a smaller root disk 
Proxmox only grows disks. The restore dialog gets an optional "Root
disk GiB" field for container backups; with a size, the backup is
restored as a new guest with --rootfs <storage>:<N> (Proxmox's advanced
restore mode).

- A smaller disk is only allowed for a new VMID, checked in the dialog
  and again in cli.restore(): with force, Proxmox deletes the old disks
  before it unpacks, and a failed unpack then removes the guest.
- The size must be a positive integer (0 gives an unlimited subvol on
  ZFS/dir storage) and above the used root disk space of a running
  guest.
- The confirm screen is a separate DANGER screen with a red border: it
  says mount points are dropped and their backed-up files land in the
  root disk, raw lxc.* lines can come back under root@pam, root disk
  options are not carried over, and the used space is no fit check.
- Never --unique or ignore-unpack-errors.

Design: docs/plans/lxc-shrink.md (section A).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-04 19:58:02 +01:00

1822 lines
58 KiB
Python
Raw Permalink 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 hypothesis import given
from hypothesis import strategies as st
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",
]
# --- current (issue #12; spec: "Live usage" > cli.py) ------------------------------
def test_current_runs_status_current_for_lxc(fake):
fake.result = (json.dumps([{"key": "status", "value": "running"}]), "", 0)
cli.current(CT)
assert fake.argv == [
"cv4pve-cli", "api", "get", "/nodes/mainframe/lxc/100/status/current", "-o", "json",
] # fmt: skip
def test_current_runs_status_current_for_qemu(fake):
fake.result = (json.dumps([{"key": "status", "value": "running"}]), "", 0)
cli.current(VM)
assert fake.argv[3] == "/nodes/mainframe/qemu/108/status/current"
def test_current_flattens_key_value_list_to_dict(fake):
fake.result = (
json.dumps([
{"key": "status", "value": "running"}, {"key": "cpu", "value": 0.0538},
{"key": "mem", "value": 932048896}, {"key": "maxmem", "value": 8589934592},
{"key": "swap", "value": 0},
]),
"", 0,
) # fmt: skip
assert cli.current(CT) == {
"status": "running", "cpu": 0.0538, "mem": 932048896,
"maxmem": 8589934592, "swap": 0,
} # fmt: skip
def test_current_cli_failure_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: nope", "", 5)
with pytest.raises(cli.CliError):
cli.current(CT)
def test_status_still_returns_status_from_current_payload(fake):
fake.result = (
json.dumps([{"key": "cpu", "value": 0.5}, {"key": "status", "value": "running"}]),
"", 0,
) # fmt: skip
assert cli.status(VM) == "running"
assert fake.argv[3] == "/nodes/mainframe/qemu/108/status/current"
# --- 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
# --- delivery B: network editing ----------------------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Network editing - delivery B" > cli.py)
CT_NET = "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=auto,type=veth"
VM_NET_LINK = "virtio=02:24:A6:49:A8:25,bridge=vmbr0"
def test_bridges_filters_bridge_types_sorted_and_uses_node_network_argv(fake):
fake.result = (
json.dumps(
[
{"iface": "vmbr1", "type": "bridge", "active": 1},
{"iface": "eno1", "type": "eth", "active": 1},
{"iface": "ovs0", "type": "OVSBridge", "active": 1},
{"iface": "bond0", "type": "bond", "active": 1},
{"iface": "vmbr0", "type": "bridge", "active": 1},
]
),
"",
0,
)
assert cli.bridges(VM) == ["ovs0", "vmbr0", "vmbr1"]
assert fake.argv == [
"cv4pve-cli",
"get",
"node",
"network",
"mainframe",
"-o",
"json",
]
def test_bridges_none_present_gives_empty_list(fake):
fake.result = (json.dumps([{"iface": "eno1", "type": "eth", "active": 1}]), "", 0)
assert cli.bridges(CT) == []
@pytest.mark.parametrize("sample", [CT_NET, VM_NET_LINK, "ip=dhcp"])
def test_format_net_round_trips_real_samples(sample):
assert cli.format_net(cli.parse_net(sample)) == sample
_token = st.text(
alphabet=st.characters(
blacklist_characters=",=", blacklist_categories=("Cs", "Cc")
),
min_size=1,
)
@given(
st.lists(
st.tuples(_token, _token), min_size=1, max_size=8, unique_by=lambda kv: kv[0]
)
)
def test_format_net_parse_net_round_trip_property(pairs):
s = ",".join(f"{k}={v}" for k, v in pairs)
assert cli.format_net(cli.parse_net(s)) == s
def test_format_net_keeps_dict_order():
assert cli.format_net({"b": "1", "a": "2"}) == "b=1,a=2"
def test_has_cloudinit_true_when_a_disk_value_mentions_cloudinit():
assert cli.has_cloudinit(
{"ide2": "local-lvm:vm-108-cloudinit,media=cdrom", "cores": 2}
)
def test_has_cloudinit_false_without_cloudinit_drive():
assert not cli.has_cloudinit(
{"scsi0": "local-lvm:vm-108-disk-0,size=10G", "cores": 2}
)
# apply_link
def test_apply_link_sets_bridge_and_returns_a_copy():
net = cli.parse_net(CT_NET)
out = cli.apply_link(net, "vmbr1", None, False)
assert out["bridge"] == "vmbr1"
assert net == cli.parse_net(CT_NET) # input untouched
def test_apply_link_new_tag_and_firewall_are_appended_after_existing_keys():
out = cli.apply_link(cli.parse_net(CT_NET), "vmbr0", 99, True)
assert list(out)[-2:] == ["tag", "firewall"] or list(out)[-2:] == [
"firewall",
"tag",
]
assert out["tag"] == "99" and out["firewall"] == "1"
assert list(out)[:6] == ["name", "bridge", "hwaddr", "ip", "ip6", "type"]
def test_apply_link_none_tag_and_false_firewall_remove_the_keys():
net = cli.parse_net("virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,firewall=1")
out = cli.apply_link(net, "vmbr0", None, False)
assert out == {"virtio": "02:24:A6:49:A8:25", "bridge": "vmbr0"}
def test_apply_link_false_firewall_removes_firewall_zero_too():
out = cli.apply_link(cli.parse_net(CT_NET + ",firewall=0"), "vmbr0", None, False)
assert "firewall" not in out
def test_apply_link_existing_keys_keep_position_and_unrelated_keys_untouched():
src = "e1000=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,mtu=1400,rate=10,trunks=10;20,firewall=1"
out = cli.apply_link(cli.parse_net(src), "vmbr2", 30, True)
assert cli.format_net(out) == (
"e1000=02:24:A6:49:A8:25,bridge=vmbr2,tag=30,mtu=1400,rate=10,trunks=10;20,firewall=1"
)
def test_apply_link_keeps_lxc_hwaddr_and_ip_keys():
out = cli.apply_link(cli.parse_net(CT_NET), "vmbr1", 5, False)
for key in ("hwaddr", "ip", "ip6", "name", "type"):
assert out[key] == cli.parse_net(CT_NET)[key]
@pytest.mark.parametrize("tag", [1, 4094])
def test_apply_link_accepts_tag_boundaries(tag):
assert cli.apply_link({"bridge": "vmbr0"}, "vmbr0", tag, False)["tag"] == str(tag)
@pytest.mark.parametrize("tag", [0, -1, 4095, 5000])
def test_apply_link_rejects_tag_outside_1_to_4094(tag):
with pytest.raises(ValueError):
cli.apply_link({"bridge": "vmbr0"}, "vmbr0", tag, False)
# ip_mode
@pytest.mark.parametrize(
("cfg", "family", "expected"),
[
({"ip": "dhcp", "ip6": "auto"}, 4, ("dhcp", "", "")),
({"ip": "dhcp", "ip6": "auto"}, 6, ("auto", "", "")),
(
{"ip": "10.0.0.5/24", "gw": "10.0.0.1"},
4,
("static", "10.0.0.5/24", "10.0.0.1"),
),
({"ip": "10.0.0.5/24"}, 4, ("static", "10.0.0.5/24", "")),
(
{"ip6": "2001:db8::5/64", "gw6": "2001:db8::1"},
6,
("static", "2001:db8::5/64", "2001:db8::1"),
),
({"ip6": "dhcp"}, 6, ("dhcp", "", "")),
({"ip": "manual", "ip6": "manual"}, 4, ("none", "", "")),
({"ip": "manual", "ip6": "manual"}, 6, ("none", "", "")),
({"ip": "dhcp"}, 6, ("none", "", "")), # key absent reads as none
({"ip6": "auto"}, 4, ("none", "", "")), # families are independent
({"ip": "10.0.0.5/24", "gw": "10.0.0.1"}, 6, ("none", "", "")),
],
)
def test_ip_mode_reads_mode_cidr_and_gateway(cfg, family, expected):
assert cli.ip_mode(cfg, family) == expected
# apply_ip
def test_apply_ip_dhcp_sets_dhcp_and_removes_gateway_keeping_other_keys():
cfg = {"name": "eth0", "ip": "10.0.0.5/24", "gw": "10.0.0.1", "ip6": "auto"}
out = cli.apply_ip(cfg, 4, "dhcp")
assert out == {"name": "eth0", "ip": "dhcp", "ip6": "auto"}
assert cfg["ip"] == "10.0.0.5/24" # input untouched
def test_apply_ip_auto_for_family_6_removes_gw6():
out = cli.apply_ip(
{"ip": "dhcp", "ip6": "2001:db8::5/64", "gw6": "2001:db8::1"}, 6, "auto"
)
assert out == {"ip": "dhcp", "ip6": "auto"}
def test_apply_ip_dhcp_for_family_6():
assert cli.apply_ip({"ip6": "auto"}, 6, "dhcp") == {"ip6": "dhcp"}
def test_apply_ip_static_v4_with_gateway():
out = cli.apply_ip(
{"ip": "dhcp", "ip6": "auto"}, 4, "static", "172.16.41.200/23", "172.16.40.1"
)
assert out == {"ip": "172.16.41.200/23", "gw": "172.16.40.1", "ip6": "auto"}
def test_apply_ip_static_v4_without_gateway_removes_old_gateway():
out = cli.apply_ip(
{"ip": "10.0.0.5/24", "gw": "10.0.0.1"}, 4, "static", "10.0.0.9/24"
)
assert out == {"ip": "10.0.0.9/24"}
def test_apply_ip_static_v6_with_and_without_gateway():
with_gw = cli.apply_ip(
{"ip6": "auto"}, 6, "static", "2001:db8::5/64", "2001:db8::1"
)
assert with_gw == {"ip6": "2001:db8::5/64", "gw6": "2001:db8::1"}
no_gw = cli.apply_ip(with_gw, 6, "static", "2001:db8::6/64")
assert no_gw == {"ip6": "2001:db8::6/64"}
def test_apply_ip_none_removes_ip_and_gateway_of_that_family_only():
cfg = {
"ip": "10.0.0.5/24",
"gw": "10.0.0.1",
"ip6": "2001:db8::5/64",
"gw6": "2001:db8::1",
}
assert cli.apply_ip(cfg, 4, "none") == {
"ip6": "2001:db8::5/64",
"gw6": "2001:db8::1",
}
assert cli.apply_ip(cfg, 6, "none") == {"ip": "10.0.0.5/24", "gw": "10.0.0.1"}
def test_apply_ip_none_on_vm_ipconfig_can_leave_empty_dict():
assert cli.apply_ip({"ip": "dhcp"}, 4, "none") == {}
@pytest.mark.parametrize(
("family", "mode", "cidr", "gw"),
[
(4, "static", "2001:db8::5/64", ""), # wrong family
(6, "static", "10.0.0.5/24", ""), # wrong family
(4, "static", "10.0.0.5", ""), # missing prefix
(6, "static", "2001:db8::5", ""), # missing prefix
(4, "static", "", ""), # missing CIDR
(4, "static", "not-an-ip/24", ""),
(4, "static", "10.0.0.5/24", "nonsense"), # bad gateway
(4, "static", "10.0.0.5/24", "2001:db8::1"), # gateway family mismatch
(6, "static", "2001:db8::5/64", "10.0.0.1"), # gateway family mismatch
(4, "auto", "", ""), # SLAAC exists only for IPv6
],
)
def test_apply_ip_validation_errors(family, mode, cidr, gw):
with pytest.raises(ValueError):
cli.apply_ip({"ip": "dhcp"}, family, mode, cidr, gw)
# set_net
def test_set_net_sets_only(fake):
cli.set_net(CT, {"net0": CT_NET + ",tag=99"})
assert fake.argv == ["cv4pve-cli", "set", "ct", "config", "mainframe", "100",
"--net0", CT_NET + ",tag=99", "--wait"] # fmt: skip
def test_set_net_deletes_only(fake):
cli.set_net(VM, {"ipconfig0": None})
assert fake.argv == ["cv4pve-cli", "set", "vm", "config", "mainframe", "108",
"--delete", "ipconfig0", "--wait"] # fmt: skip
def test_set_net_sets_then_deletes_joined_with_comma_and_wait_last(fake):
cli.set_net(
VM, {"net0": VM_NET_LINK + ",tag=5", "ipconfig0": None, "ipconfig1": None}
)
assert fake.argv == ["cv4pve-cli", "set", "vm", "config", "mainframe", "108",
"--net0", VM_NET_LINK + ",tag=5", "--delete", "ipconfig0,ipconfig1", "--wait"] # fmt: skip
def test_set_net_empty_changes_is_value_error_without_call(fake):
with pytest.raises(ValueError):
cli.set_net(VM, {})
assert fake.calls == []
def test_set_net_failure_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: boom", returncode=1)
with pytest.raises(cli.CliError):
cli.set_net(CT, {"net0": CT_NET})
# --- review findings: network editing -----------------------------------------
def test_has_cloudinit_only_looks_at_disks():
assert cli.has_cloudinit({"ide2": "local-lvm:vm-108-cloudinit,media=cdrom"})
assert not cli.has_cloudinit(
{"name": "ubuntu-cloudinit-tpl", "description": "cloudinit base"}
)
@pytest.mark.parametrize(
"cidr", ["10.0.0.5/255.255.255.0", "10.0.0.5/0", "2001:db8::5/0"]
)
def test_apply_ip_rejects_netmask_form_and_zero_prefix(cidr):
family = 6 if ":" in cidr else 4
with pytest.raises(ValueError):
cli.apply_ip({}, family, "static", cidr)
# --- #9: guests not covered by any backup job -----------------------------------------
def test_not_backed_up_returns_vmids(fake):
fake.result = (
json.dumps([{"vmid": 104, "name": "debian", "type": "lxc"}, {"vmid": "110"}]),
"",
0,
)
assert cli.not_backed_up() == {104, 110}
assert fake.argv == ["cv4pve-cli", "get", "cluster", "not-backed-up", "-o", "json"]
def test_not_backed_up_empty(fake):
fake.result = ("[]", "", 0)
assert cli.not_backed_up() == set()
# --- snapshots (issue #8); spec: docs/plans/guest-tui-mvp.md ("Snapshots in the info screen") ---
# Sections below name the spec bullet they cover.
GOOD_NAMES = [
"a",
"snap1",
"tui_20251102_1430",
"A-b_c",
"x" * 40,
"snapshot",
"currentx",
"curren",
]
BAD_NAMES = [
"",
"current",
"Current",
"CURRENT",
"1snap",
"_snap",
"-snap",
"my snap",
"a,b",
";rm",
"a;rm -rf",
"x" * 41,
"snäp",
"snap\n",
"snap ",
]
@pytest.mark.parametrize("name", GOOD_NAMES)
def test_valid_snapname_accepts_pattern(
name,
): # valid_snapname: [A-Za-z][A-Za-z0-9_-]{0,39}
assert cli.valid_snapname(name) == name
@pytest.mark.parametrize("name", BAD_NAMES)
def test_valid_snapname_rejects(
name,
): # valid_snapname: ValueError outside pattern / "current"
with pytest.raises(ValueError):
cli.valid_snapname(name)
@given(st.text())
def test_delete_snapshot_with_rejected_name_raises_and_never_runs(name):
# delete_snapshot: name validated first; no subprocess call for an invalid name
try:
cli.valid_snapname(name)
except ValueError:
pass
else:
return # accepted names are covered by the argv tests
with pytest.MonkeyPatch.context() as m:
run = FakeRun(m)
with pytest.raises(ValueError):
cli.delete_snapshot(CT, name)
with pytest.raises(ValueError):
cli.delete_snapshot(VM, name)
assert run.calls == []
SNAPS_RAW = [
{"name": "old", "description": "first", "snaptime": 1_000, "parent": None},
{"name": "undated", "description": "", "snaptime": None, "parent": "old"},
{
"name": "current",
"description": "You are here!",
"snaptime": None,
"parent": "new",
},
{"name": "new", "description": "second", "snaptime": 3_000, "parent": "old"},
{"name": "mid", "description": "", "snaptime": 2_000, "parent": "old"},
]
def test_snapshots_drops_current_sorts_newest_first_null_last(fake): # snapshots()
fake.result = (json.dumps(SNAPS_RAW), "", 0)
assert [s["name"] for s in cli.snapshots(CT)] == ["new", "mid", "old", "undated"]
def test_snapshots_drops_current_case_insensitively(
fake,
): # snapshots(): current is not a snapshot
fake.result = (
json.dumps([{"name": "Current", "snaptime": None}, SNAPS_RAW[0]]),
"",
0,
)
assert [s["name"] for s in cli.snapshots(CT)] == ["old"]
def test_snapshots_only_current_gives_empty_list(
fake,
): # snapshots(): "No snapshots" case
fake.result = (json.dumps([SNAPS_RAW[2]]), "", 0)
assert cli.snapshots(CT) == []
@pytest.mark.parametrize(
"guest,kind",
[(CT, "lxc"), (VM, "qemu")],
)
def test_snapshots_argv_uses_qemu_or_lxc(
fake, guest, kind
): # CLI facts: get guest snapshots
fake.result = ("[]", "", 0)
cli.snapshots(guest)
assert fake.argv == [
"cv4pve-cli", "get", "guest", "snapshots", "mainframe", kind, str(guest["vmid"]), "-o", "json",
] # fmt: skip
def test_snapshots_failure_raises_clierror(
monkeypatch,
): # snapshots(): failure surfaces as CliError
FakeRun(monkeypatch, "Error: boom", returncode=1)
with pytest.raises(cli.CliError):
cli.snapshots(CT)
def test_take_snapshot_ct_argv_has_no_vmstate_and_wait_last(fake): # take_snapshot
cli.take_snapshot(CT, "snap1", "hello")
assert fake.argv == [
"cv4pve-cli", "create", "ct", "snapshot", "mainframe", "100", "snap1", "hello", "--wait",
] # fmt: skip
def test_take_snapshot_vm_without_vmstate(fake): # take_snapshot
cli.take_snapshot(VM, "snap1", "hello")
assert fake.argv == [
"cv4pve-cli", "create", "vm", "snapshot", "mainframe", "108", "snap1", "hello", "--wait",
] # fmt: skip
def test_take_snapshot_vm_with_vmstate_before_wait(
fake,
): # take_snapshot: --vmstate 1 (VM only)
cli.take_snapshot(VM, "snap1", "hello", vmstate=True)
assert fake.argv == [
"cv4pve-cli", "create", "vm", "snapshot", "mainframe", "108", "snap1", "hello",
"--vmstate", "1", "--wait",
] # fmt: skip
def test_take_snapshot_ct_ignores_vmstate(fake): # take_snapshot: vmstate only for a VM
cli.take_snapshot(CT, "snap1", "hello", vmstate=True)
assert "--vmstate" not in fake.argv and fake.argv[-1] == "--wait"
def test_take_snapshot_default_description(
fake,
): # take_snapshot: description or "cv4pve-tui"
cli.take_snapshot(CT, "snap1")
assert fake.argv[7] == "cv4pve-tui"
def test_take_snapshot_empty_description_falls_back(
fake,
): # take_snapshot: `description or "cv4pve-tui"`
cli.take_snapshot(VM, "snap1", "")
assert fake.argv[7] == "cv4pve-tui"
@pytest.mark.parametrize("name", ["", "current", "1x", "a b", ";rm"])
def test_take_snapshot_invalid_name_no_call(
fake, name
): # take_snapshot: name validated first
with pytest.raises(ValueError):
cli.take_snapshot(VM, name, "d")
assert fake.calls == []
@pytest.mark.parametrize("guest,kind", [(CT, "ct"), (VM, "vm")])
def test_rollback_argv_without_start(fake, guest, kind): # rollback_snapshot
cli.rollback_snapshot(guest, "snap1")
assert fake.argv == [
"cv4pve-cli", "do", "rollback", kind, "mainframe", str(guest["vmid"]), "snap1", "-y", "--wait",
] # fmt: skip
@pytest.mark.parametrize("guest,kind", [(CT, "ct"), (VM, "vm")])
def test_rollback_argv_with_start(
fake, guest, kind
): # rollback_snapshot: [--start 1] -y --wait
cli.rollback_snapshot(guest, "snap1", start=True)
assert fake.argv == [
"cv4pve-cli", "do", "rollback", kind, "mainframe", str(guest["vmid"]), "snap1",
"--start", "1", "-y", "--wait",
] # fmt: skip
@pytest.mark.parametrize("name", ["", "current", "Current", "9", "a b", "a,b"])
def test_rollback_invalid_name_no_call(fake, name): # rollback_snapshot: name validated
with pytest.raises(ValueError):
cli.rollback_snapshot(CT, name, start=True)
assert fake.calls == []
@pytest.mark.parametrize("guest,kind", [(CT, "ct"), (VM, "vm")])
def test_delete_snapshot_exact_argv(fake, guest, kind): # delete_snapshot
cli.delete_snapshot(guest, "snap1")
assert fake.argv == [
"cv4pve-cli", "delete", kind, "snapshot", "mainframe", str(guest["vmid"]), "snap1", "-y", "--wait",
] # fmt: skip
@given(
name=st.from_regex(r"[A-Za-z][A-Za-z0-9_-]{0,39}", fullmatch=True).filter(
lambda n: n.lower() != "current"
),
guest=st.sampled_from([CT, VM]),
)
def test_delete_never_produces_guest_destroying_form(name, guest):
# DANGER: `delete <kind> <node> <vmid> -y` destroys the guest; "snapshot" must sit at index 3
kind = {"lxc": "ct", "qemu": "vm"}[guest["type"]]
with pytest.MonkeyPatch.context() as m:
run = FakeRun(m)
cli.delete_snapshot(guest, name)
argv = run.argv
assert argv[1:3] == ["delete", kind]
assert argv[3] == "snapshot" # never the node directly after the kind
assert (
argv[4] == guest["node"] and argv[5] == str(guest["vmid"]) and argv[6] == name
)
assert argv[1:3] != ["delete", kind] or argv[3] != guest["node"]
def test_delete_with_snapshot_like_name_still_has_snapshot_at_index_3(fake):
# a valid name equal to the subcommand word must not shift the positions
cli.delete_snapshot(CT, "snapshot")
assert fake.argv[3:7] == ["snapshot", "mainframe", "100", "snapshot"]
def test_take_snapshot_refuses_description_that_looks_like_an_option(fake):
with pytest.raises(ValueError):
cli.take_snapshot(
{"vmid": 104, "node": "mainframe", "type": "lxc"}, "snap1", "-y"
)
assert fake.calls == []
# --- issue #10: task log (spec: "Task log (issue #10)" > cli.py) ------------------------
UPID = "UPID:mainframe:0001:00A:68F:qmstart:108:root@pam:"
def test_tasks_argv_is_cluster_tasks_json(fake): # tasks(): api get /cluster/tasks
fake.result = ("[]", "", 0)
assert cli.tasks() == []
assert fake.argv == ["cv4pve-cli", "api", "get", "/cluster/tasks", "-o", "json"]
def test_tasks_newest_first_including_running_without_endtime(
fake,
): # tasks(): newest first
raw = [
{"upid": "a", "starttime": 100, "endtime": 150, "status": "OK"},
{"upid": "c", "starttime": 300}, # running: no endtime, no status
{"upid": "b", "starttime": 200, "endtime": None},
]
fake.result = (json.dumps(raw), "", 0)
assert [t["upid"] for t in cli.tasks()] == ["c", "b", "a"]
def test_tasks_failure_raises_clierror(monkeypatch): # failure surfaces as CliError
FakeRun(monkeypatch, "Error: boom", returncode=1)
with pytest.raises(cli.CliError):
cli.tasks()
def test_task_log_argv_and_returns_text(fake): # task_log(): task log <upid>
fake.result = ("starting\nTASK OK\n", "", 0)
out = cli.task_log(UPID)
assert fake.argv == ["cv4pve-cli", "task", "log", UPID]
assert "TASK OK" in out
@pytest.mark.parametrize("bad", ["-y", "--help", "", "108", "x UPID:a"])
def test_task_log_rejects_non_upid_without_calling_subprocess(
fake, bad
): # ValueError unless UPID:
with pytest.raises(ValueError):
cli.task_log(bad)
assert fake.calls == []
# --- #13: nodes / package_updates (spec: "Node package updates (issue #13)") ---------------
def test_nodes_argv_only_online_sorted(fake):
fake.result = (json.dumps([{"node": "zeta", "status": "online"},
{"node": "down", "status": "offline"},
{"node": "alpha", "status": "online"}]), "", 0) # fmt: skip
assert cli.nodes() == ["alpha", "zeta"]
assert fake.argv == ["cv4pve-cli", "get", "nodes", "-o", "json"]
def test_package_updates_argv_and_sorted_by_package(fake):
fake.result = (json.dumps([{"Package": "zsh", "Version": "2"},
{"Package": "bash", "Version": "1"},
{"Package": "curl", "Version": "3"}]), "", 0) # fmt: skip
got = cli.package_updates("mainframe")
assert fake.argv == [
"cv4pve-cli",
"get",
"node",
"packages",
"mainframe",
"-o",
"json",
]
assert [p["Package"] for p in got] == ["bash", "curl", "zsh"]
assert got[0] == {"Package": "bash", "Version": "1"}
def test_package_updates_failure_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: node unreachable\n", returncode=1)
with pytest.raises(cli.CliError):
cli.package_updates("mainframe")
# --- #14 ----------------------------------------------------------------------------
def test_version_argv_and_strips(fake):
fake.result = ("2.4.0\n", "", 0)
assert cli.version() == "2.4.0"
assert fake.argv == ["cv4pve-cli", "--version"]
def test_tested_version_constant():
assert cli.TESTED_VERSION == "2.4.0"
# --- #18: startup order and tags (spec: "Autostart, startup order and tags") ----------
@pytest.mark.parametrize(
("args", "expected"),
[
((2, 10, 30), "order=2,up=10,down=30"),
((2, 10, None), "order=2,up=10"),
((2, None, None), "order=2"),
((None, 10, None), "up=10"),
((None, None, 30), "down=30"),
((None, 10, 30), "up=10,down=30"),
((0, 0, 0), "order=0,up=0,down=0"), # 0 is a value, not "unset"
((None, None, None), None),
],
)
def test_startup_value_builds_only_the_given_parts(args, expected):
assert cli.startup_value(*args) == expected
@pytest.mark.parametrize(
"args", [(-1, None, None), (None, -5, None), (None, None, -30), (1, 2, -1)]
)
def test_startup_value_rejects_negatives(args):
with pytest.raises(ValueError):
cli.startup_value(*args)
@pytest.mark.parametrize("bad", ["x", 1.5, [1]])
def test_startup_value_rejects_non_ints(bad):
for args in ((bad, None, None), (None, bad, None), (None, None, bad)):
with pytest.raises(ValueError):
cli.startup_value(*args)
@pytest.mark.parametrize(
("text", "expected"),
[
("order=2,up=10,down=30", (2, 10, 30)),
("3", (3, None, None)), # a bare leading number is the order
("up=5", (None, 5, None)),
("order=2,up=10", (2, 10, None)),
("3,up=5", (3, 5, None)),
("down=7", (None, None, 7)),
(None, (None, None, None)),
("", (None, None, None)),
],
)
def test_parse_startup(text, expected):
assert cli.parse_startup(text) == expected
@given(
st.tuples(*[st.one_of(st.none(), st.integers(min_value=0, max_value=10**6))] * 3)
)
def test_parse_startup_inverts_startup_value_property(parts):
assert cli.parse_startup(cli.startup_value(*parts)) == parts
@pytest.mark.parametrize(
("text", "expected"),
[
("os", "os"),
("a;b;c", "a;b;c"),
("a,b", "a;b"),
("a b", "a;b"), # whitespace separates: two tags, not one with a space
("a;b, c d", "a;b;c;d"),
("a; b; c", "a;b;c"),
("a;b;a;c;b", "a;b;c"), # duplicates dropped, first occurrence wins
("c;a;b", "c;a;b"), # order kept, not sorted
(";;a;;,, b;", "a;b"), # empties dropped
("172.16.41.200", "172.16.41.200"), # IP-like tag is valid
("172.16.41.200;community-script;os", "172.16.41.200;community-script;os"),
("_x;a+b;a.b;a-b;9", "_x;a+b;a.b;a-b;9"),
],
)
def test_tags_value_normalises(text, expected):
assert cli.tags_value(text) == expected
@pytest.mark.parametrize("text", ["", " ", ";", " ; , ;", "\n\t"])
def test_tags_value_empty_is_none(text):
assert cli.tags_value(text) is None
@pytest.mark.parametrize(
"bad", ["-x", "a;b!", "!", "a;-b", ".a", "+a", "a/b", "a;b@c", "ä"]
)
def test_tags_value_rejects_invalid_tags(bad):
with pytest.raises(ValueError):
cli.tags_value(bad)
_TAG = st.from_regex(r"[A-Za-z0-9_][A-Za-z0-9_+.-]*", fullmatch=True)
_SEP = st.sampled_from([";", ",", " ", "; ", ", ", " ", ";;"])
@st.composite
def _valid_tag_text(draw):
tags = draw(st.lists(_TAG, max_size=6))
out = ""
for tag in tags:
out += tag + draw(_SEP)
return out
@given(_valid_tag_text())
def test_tags_value_is_idempotent_property(text):
once = cli.tags_value(text)
if once is None:
assert text.replace(";", "").replace(",", "").strip() == ""
else:
assert cli.tags_value(once) == once
assert len(set(once.split(";"))) == len(once.split(";")) # no duplicates
# --- issue #19: guest-agent fstrim (spec: "fstrim in a VM via the guest agent") ----------
@pytest.mark.parametrize(
"agent,expected",
[
("enabled=1", True),
("1", True),
("1,fstrim_cloned_disks=1", True),
("enabled=0", False),
("0", False),
(None, False),
],
)
def test_agent_enabled(agent, expected):
config = {} if agent is None else {"agent": agent}
assert cli.agent_enabled(config) is expected
def test_agent_ping_exact_argv(fake):
fake.result = (json.dumps([{"key": "result", "value": "{}"}]), "", 0)
cli.agent_ping(VMID108)
assert fake.argv == [
"cv4pve-cli", "do", "vm", "agent", "ping", "mainframe", "108", "-o", "json",
] # fmt: skip
def test_agent_ping_nonzero_exit_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: QEMU guest agent is not running", returncode=5)
with pytest.raises(cli.CliError):
cli.agent_ping(VMID108)
FSTRIM_PATHS = [
{"path": "/", "trimmed": 1073741824, "minimum": 0},
{"path": "/boot", "error": "not supported"},
]
def fstrim_stdout(value):
return json.dumps([{"key": "result", "value": value}])
def test_fstrim_exact_argv_and_parses_paths(fake):
fake.result = (fstrim_stdout(json.dumps({"paths": FSTRIM_PATHS})), "", 0)
assert cli.fstrim(VMID108) == FSTRIM_PATHS
assert fake.argv == [
"cv4pve-cli", "do", "vm", "agent", "fstrim", "mainframe", "108", "-o", "json",
] # fmt: skip
@pytest.mark.parametrize(
"stdout",
[
"[]",
json.dumps([{"key": "other", "value": "{}"}]),
fstrim_stdout("not json at all"),
fstrim_stdout("{}"),
fstrim_stdout(""),
],
)
def test_fstrim_missing_or_garbage_result_is_empty_list(fake, stdout):
fake.result = (stdout, "", 0)
assert cli.fstrim(VMID108) == []
def test_fstrim_nonzero_exit_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: fstrim failed", returncode=5)
with pytest.raises(cli.CliError):
cli.fstrim(VMID108)
# --- #19 review findings ----------------------------------------------------------------
VM19 = {"vmid": 108, "node": "mainframe", "type": "qemu"}
@pytest.mark.parametrize(
"stdout",
[
"garbage",
"",
json.dumps({"key": "result"}), # object instead of list
json.dumps([{"value": "{}"}]), # item without "key"
json.dumps([{"key": "result", "value": {"paths": "x"}}]), # already a dict
],
)
def test_fstrim_unreadable_output_is_empty_not_an_error(fake, stdout):
fake.result = (stdout, "", 0)
assert cli.fstrim(VM19) == []
def test_fstrim_normalises_odd_path_entries(fake):
paths = [
{"path": "/", "trimmed": "12"},
{"error": "x"},
"junk",
{"path": "/var", "trimmed": 5},
]
fake.result = (
json.dumps([{"key": "result", "value": json.dumps({"paths": paths})}]),
"",
0,
)
assert cli.fstrim(VM19) == [
{"path": "/", "trimmed": 0}, # non-int trimmed -> 0
{"error": "x", "path": "?"}, # missing path -> "?"; "junk" dropped
{"path": "/var", "trimmed": 5}, # well-formed: unchanged
]
def test_fstrim_gets_a_longer_timeout_than_other_calls(fake):
fake.result = (json.dumps([{"key": "result", "value": "{}"}]), "", 0)
cli.fstrim(VM19)
assert fake.calls[0][1]["timeout"] >= 1800 # a big disk can take longer than 330 s
# --- restore a guest from a backup (issue #15) ------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Restore a guest from a backup (issue #15)" > cli.py)
CT_VOLID = "truenas-backup:backup/vzdump-lxc-104-2026_11_02-04_11_47.tar.zst"
VM_VOLID = "local:backup/vzdump-qemu-108-2026_11_02-04_11_47.vma.zst"
def test_next_id_parses_json_number_and_runs_get_cluster_nextid(fake):
# spec: next_id() -> int; `get cluster nextid -o json` -> a JSON number
fake.result = ("105\n", "", 0)
assert cli.next_id() == 105
assert fake.argv == ["cv4pve-cli", "get", "cluster", "nextid", "-o", "json"]
@pytest.mark.parametrize("volid", [CT_VOLID, VM_VOLID])
def test_valid_volid_accepts_real_volids_and_returns_them(volid):
# spec: valid_volid(volid) -> str
assert cli.valid_volid(volid) == volid
@pytest.mark.parametrize(
"volid",
[
"-y",
"",
"local:iso/x.iso",
"a b",
"x:backup/vzdump-lxc-104-",
"--force:backup/vzdump-lxc-104-x", # option-looking storage part
"local:backup/vzdump-lxc-104-x y", # whitespace in the file part
"local:backup/vzdump-vm-104-x.tar", # kind is lxc|qemu only
],
)
def test_valid_volid_rejects_everything_else(volid):
with pytest.raises(ValueError):
cli.valid_volid(volid)
def test_restore_storages_ct_wants_rootdir_vm_wants_images(fake):
# spec: enabled+active storages whose content has rootdir (CT) or images (VM), API order
fake.result = (
json.dumps([
storage("zeta", "iso,rootdir,images"),
storage("only-backup", "backup"),
storage("vm-only", "images"),
storage("ct-only", "rootdir,vztmpl"),
storage("off", "rootdir,images", enabled=0),
storage("inactive", "rootdir,images", active=0),
storage("alpha", "images,rootdir"),
]),
"",
0,
) # fmt: skip
assert cli.restore_storages(CT) == ["zeta", "ct-only", "alpha"]
assert cli.restore_storages(VM) == ["zeta", "vm-only", "alpha"]
assert fake.calls[0][0] == [
"cv4pve-cli", "get", "node", "storages", "mainframe", "-o", "json",
] # fmt: skip
def test_restore_new_vm_exact_argv_without_force_or_start(fake):
# spec: do restore <kind> <node> <vmid> <volid> --storage <storage> -y --wait
cli.restore(VM, VM_VOLID, 109, "local-lvm")
assert fake.argv == [
"cv4pve-cli", "do", "restore", "vm", "mainframe", "109", VM_VOLID,
"--storage", "local-lvm", "-y", "--wait",
] # fmt: skip
def test_restore_new_ct_with_start_exact_argv(fake):
cli.restore(CT, CT_VOLID, 105, "zfs", start=True)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "ct", "mainframe", "105", CT_VOLID,
"--storage", "zfs", "--start", "1", "-y", "--wait",
] # fmt: skip
def test_restore_overwrite_vm_exact_argv_with_force(fake):
cli.restore(VM, VM_VOLID, 108, "local-lvm", force=True)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "vm", "mainframe", "108", VM_VOLID,
"--storage", "local-lvm", "--force", "1", "-y", "--wait",
] # fmt: skip
def test_restore_overwrite_ct_with_start_exact_argv(fake):
cli.restore(CT, CT_VOLID, 100, "zfs", force=True, start=True)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "ct", "mainframe", "100", CT_VOLID,
"--storage", "zfs", "--force", "1", "--start", "1", "-y", "--wait",
] # fmt: skip
def test_restore_force_on_another_vmid_raises_and_never_runs(fake):
# spec: force=True only when vmid == guest["vmid"] - never overwrite some other guest
with pytest.raises(ValueError):
cli.restore(VM, VM_VOLID, 100, "local-lvm", force=True)
assert fake.calls == []
@pytest.mark.parametrize("vmid", [0, -1, -108])
@pytest.mark.parametrize("force", [False, True])
def test_restore_non_positive_vmid_raises_and_never_runs(fake, vmid, force):
# spec: vmid a positive int
with pytest.raises(ValueError):
cli.restore(VM, VM_VOLID, vmid, "local-lvm", force=force)
assert fake.calls == []
@pytest.mark.parametrize("volid", ["-y", "", "local:iso/x.iso", "a b"])
def test_restore_rejected_volid_raises_and_never_runs(fake, volid):
# spec: volid validated (it is a positional argument)
with pytest.raises(ValueError):
cli.restore(VM, volid, 109, "local-lvm")
assert fake.calls == []
def test_restore_failure_raises_clierror(monkeypatch):
FakeRun(monkeypatch, "Error: storage is full", returncode=5)
with pytest.raises(cli.CliError):
cli.restore(VM, VM_VOLID, 109, "local-lvm")
@given(vmid=st.integers(min_value=-10, max_value=10**9).filter(lambda n: n != 108))
def test_restore_force_never_runs_for_a_vmid_other_than_the_guests(vmid):
# DANGER property: --force 1 on any other guest would destroy it
with pytest.MonkeyPatch.context() as m:
run = FakeRun(m)
with pytest.raises(ValueError):
cli.restore(VM, VM_VOLID, vmid, "local-lvm", force=True)
assert run.calls == []
# --- #15 review findings ----------------------------------------------------------------
@pytest.mark.parametrize("force", [False, True])
def test_restore_refuses_a_backup_of_the_other_guest_type(fake, force):
ct = {"vmid": 100, "node": "mainframe", "type": "lxc"}
with pytest.raises(ValueError):
cli.restore(
ct, "local:backup/vzdump-qemu-100-x.vma.zst", 100, "local-lvm", force=force
)
assert fake.calls == []
# --- shrink restore: smaller rootfs (docs/plans/lxc-shrink.md, A) ------------
def test_restore_ct_with_rootfs_gb_adds_rootfs_on_the_chosen_storage(fake):
cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=4)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "ct", "mainframe", "105", CT_VOLID,
"--storage", "zfs", "--rootfs", "zfs:4", "-y", "--wait",
] # fmt: skip
def test_restore_ct_with_rootfs_gb_and_start_exact_argv(fake):
cli.restore(CT, CT_VOLID, 105, "zfs", start=True, rootfs_gb=4)
assert fake.argv == [
"cv4pve-cli", "do", "restore", "ct", "mainframe", "105", CT_VOLID,
"--storage", "zfs", "--start", "1", "--rootfs", "zfs:4", "-y", "--wait",
] # fmt: skip
def test_restore_without_rootfs_gb_argv_is_unchanged(fake):
cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=None)
assert "--rootfs" not in fake.argv
assert fake.argv[-4:] == ["--storage", "zfs", "-y", "--wait"]
def test_restore_with_rootfs_gb_never_adds_unique_or_ignore_unpack_errors(fake):
cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=4)
joined = " ".join(fake.argv)
assert "--unique" not in joined and "ignore-unpack-errors" not in joined
@pytest.mark.parametrize("bad", [0, -1, True, "4", 1.5])
def test_restore_rootfs_gb_must_be_a_positive_int_and_never_runs(fake, bad):
# a size of 0 makes PVE allocate an unlimited subvol on ZFS/dir
with pytest.raises(ValueError):
cli.restore(CT, CT_VOLID, 105, "zfs", rootfs_gb=bad)
assert fake.calls == []
def test_restore_rootfs_gb_with_force_raises_and_never_runs(fake):
# force=1 destroys the old volumes first; a failed extraction would lose the guest
with pytest.raises(ValueError):
cli.restore(CT, CT_VOLID, 100, "zfs", force=True, rootfs_gb=4)
assert fake.calls == []
def test_restore_rootfs_gb_on_a_vm_backup_raises_and_never_runs(fake):
with pytest.raises(ValueError):
cli.restore(VM, VM_VOLID, 109, "local-lvm", rootfs_gb=4)
assert fake.calls == []
Reference in new issue View Git Blame Copy Permalink