DuckbootandClaude Opus 5.5
364f8e2137
U opens "Package updates": the pending apt updates of every online node, with node, package, current and new version, origin and priority. A node that fails is shown as "<node>: unavailable", and the other nodes are still listed. A hint says the list is as of the node's last apt update; this call does not refresh it. The main frame's bottom border shows "· N package updates" after the backup count, with a singular for 1. The count is fetched on mount, after a context switch and whenever the updates screen loads, not on every guest reload, since it costs one CLI call per node. It uses the same generation guard as the guest list, so a count from the old cluster is dropped after a switch. Running apt update or upgrade stays out of scope; it is a write. Read-only live check: "mainframe · 23 package updates", and 23 rows in the screen. Closes #13 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1378 lines
44 KiB
Python
1378 lines
44 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 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")
|