DuckbootandClaude Opus 5.5
0df928ab11
r on the info screen restores the backup under the cursor of the backup list. - New guest (default): the next free VMID from `get cluster nextid`, editable. The original is left untouched. An id already used by a listed guest is refused. - Overwrite: replaces the guest itself. The guest must be stopped; its live status is read at apply and again right before the call. The VMID must be typed to confirm, then an "Overwrite" confirm follows. --storage is always passed. It is preselected to the storage of the guest's rootfs or first disk, and the choices are the node's storages that can hold CT (rootdir) or VM (images) disks. Start after restore is off by default, because a copy keeps the original's MAC address and IP. Safety: - cli.restore refuses force (overwrite) for any VMID other than the guest's own (explicit raise). A hypothesis property checks that no other VMID ever reaches the CLI with force. - The volid and storage are validated before they become arguments. - A backup of the other guest type is refused. A reused VMID can carry a qemu backup for a CT or the other way round, and whether PVE rejects a mismatched archive before removing the old guest is not verified. - An earlier guest's backup (visible with a) is labelled with that guest's name in the dialog and the confirm. - r before the info screen has loaded says "still loading". - Restore gets a 1-hour CLI timeout. Not run live: a live test would create a guest that then has to be deleted, which the issue reserves for the user's approval. A read-only check opened the dialog for CT 104 (next id 105, local-lvm preselected) and closed it without any call. Closes #15 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1770 lines
56 KiB
Python
1770 lines
56 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")
|
|
|
|
|
|
# --- #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 == []
|