DuckbootandClaude Opus 5.5
1ed5e5069b
l on the main screen opens "Tasks", the recent cluster tasks newest
first. l on the info screen shows only that guest's tasks. Columns:
start, duration (or "running"), node, type, VMID ("-" for node
tasks), user and status. A failed task gets a "!" in the leading
column. Enter opens the task's log in a plain-text widget, so the
output is never parsed as markup.
`task list -o json` has no start or end times, so the list comes from
`/cluster/tasks`. The upid is validated (it must start with "UPID:")
before it goes in as a positional argument.
Review fixes:
- A finished task without a start time no longer crashes the app; its
duration shows "-".
- "running" now means there is no endtime. A finished task with an
empty status shows "unknown" and is flagged.
Read-only live check: 25 tasks listed, and the log of the newest
(vzdelsnapshot 104) ends with TASK OK.
Closes #10
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1343 lines
43 KiB
Python
1343 lines
43 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 == []
|