Duckboot/cv4pve-cli-tui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0c3669271c1bc9bd09ad57097f9fea3be29ec2ca
cv4pve-cli-tui/tests/test_cli.py
T
DuckbootandClaude Opus 5.5 5038148686 Add network editing with VLAN, firewall, IPv4/IPv6 and undo 
Press n in the info screen to edit one interface: bridge, VLAN tag,
firewall, IPv4 (DHCP, static or none) and IPv6 (SLAAC, DHCP, static or
none). The netN string is parsed, the chosen fields are changed and
the string is written back, so hwaddr, model=MAC, mtu, rate, trunks
and type are kept. For VMs, IP settings go to cloud-init ipconfigN and
only when the VM has a cloud-init drive.

A confirm shows the old and new raw values before anything is sent.
Press u to restore the values from before the last change to that
guest in this session.

Review fixes:
- Undo entries are cleared on a context switch, because VMIDs repeat
  across clusters.
- has_cloudinit only looks at disks.
- CIDR needs a decimal prefix: no netmask form and no /0.
- Fields you did not touch are not rewritten.

Live test on CT 104: firewall, VLAN 99, IPv6 DHCP and static IPv4 were
each applied and undone, and net0 ended identical to the baseline.
Going from static back to DHCP left the CT without IPv4 until a
reboot, so the confirm now warns that IP changes on a running
container may need one.

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

988 lines
31 KiB
Python
Raw Blame History

"""Tests for cv4pve_tui.cli; spec: docs/plans/guest-tui-mvp.md ("Module cv4pve_tui/cli.py").
subprocess.run is always mocked: cv4pve-cli is never executed.
"""
import json
import subprocess
import pytest
from 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",
]
# --- 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)
Reference in new issue View Git Blame Copy Permalink