Christian Manivong
dd48d3c1e5
start_vm, stop_vm, reboot_vm, suspend_vm and get_vm_config existed only
as declarations. netOrk called Proxmox's own power_vm and read a VM's
raw config through _node_api(), so no other hypervisor could serve the
same endpoints. These let netOrk talk to every hypervisor alike.
The power methods accept a VM's name or vmid, wait for the Proxmox task,
and raise ValueError/RuntimeError as the contract says instead of
returning a result dict. A forced reboot of a container is stop + start,
since LXC has no reset; suspending a container is refused. power_vm is
unchanged for existing callers.
get_vm_config moves the config parsing netOrk did in
_parse_proxmox_hw_config into the driver and returns a VMConfigDict:
disks with storage and size, NICs with model, MAC, bridge and VLAN, CPU
topology, firmware, machine type and PCI/USB passthrough.
get_vms reports vmid as a string ("100"), following
napalm-device-types 2.0, still ordered numerically.
240 lines
8.3 KiB
Python
240 lines
8.3 KiB
Python
"""HypervisorDriver contract methods: VM lookup, power actions, get_vm_config.
|
|
|
|
netOrk used to call Proxmox's own ``power_vm`` and reach into ``_node_api()``
|
|
for a VM's hardware. Both are Proxmox-only, so a second hypervisor could not
|
|
serve the same endpoints. These pin the contract methods that replace them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from napalm_proxmox.vm_contract_mixin import parse_vm_config
|
|
|
|
QEMU_LIST = [{"vmid": 100, "name": "web01", "status": "running"}]
|
|
LXC_LIST = [{"vmid": 200, "name": "dns01", "status": "running"}]
|
|
|
|
|
|
@pytest.fixture
|
|
def api(driver):
|
|
node = driver._node_api()
|
|
node.qemu.get.return_value = QEMU_LIST
|
|
node.lxc.get.return_value = LXC_LIST
|
|
node.qemu.return_value.status.start.post.return_value = "UPID:start"
|
|
driver._wait_for_task = MagicMock()
|
|
return node
|
|
|
|
|
|
class TestGetVmsReportsStringIds:
|
|
def test_vmid_is_a_string(self, driver, api):
|
|
driver.get_vm_interfaces = MagicMock(return_value=({}, False, False))
|
|
driver._get_vm_disk_and_boot = MagicMock(return_value=([], False))
|
|
assert [vm["vmid"] for vm in driver.get_vms()] == ["100", "200"]
|
|
|
|
def test_ordered_numerically_not_lexically(self, driver, api):
|
|
api.qemu.get.return_value = [{"vmid": 1000, "name": "a"}, {"vmid": 99, "name": "b"}]
|
|
api.lxc.get.return_value = []
|
|
driver.get_vm_interfaces = MagicMock(return_value=({}, False, False))
|
|
driver._get_vm_disk_and_boot = MagicMock(return_value=([], False))
|
|
assert [vm["vmid"] for vm in driver.get_vms()] == ["99", "1000"]
|
|
|
|
|
|
class TestResolveVm:
|
|
def test_by_vmid_string(self, driver, api):
|
|
assert driver._resolve_vm("100") == (100, "vm")
|
|
|
|
def test_by_name(self, driver, api):
|
|
assert driver._resolve_vm("dns01") == (200, "container")
|
|
|
|
def test_unknown_raises_value_error(self, driver, api):
|
|
with pytest.raises(ValueError, match="nope"):
|
|
driver._resolve_vm("nope")
|
|
|
|
|
|
class TestPowerActions:
|
|
def test_start_posts_and_waits_for_the_task(self, driver, api):
|
|
driver.start_vm("web01")
|
|
api.qemu.return_value.status.start.post.assert_called_once()
|
|
driver._wait_for_task.assert_called_once_with("UPID:start", timeout=120)
|
|
|
|
@pytest.mark.parametrize(("force", "action"), [(False, "shutdown"), (True, "stop")])
|
|
def test_stop_graceful_or_forced(self, driver, api, force, action):
|
|
driver.stop_vm("100", force=force)
|
|
getattr(api.qemu.return_value.status, action).post.assert_called_once()
|
|
|
|
@pytest.mark.parametrize(("force", "action"), [(False, "reboot"), (True, "reset")])
|
|
def test_reboot_graceful_or_forced(self, driver, api, force, action):
|
|
driver.reboot_vm("100", force=force)
|
|
getattr(api.qemu.return_value.status, action).post.assert_called_once()
|
|
|
|
def test_forced_reboot_of_a_container_is_a_stop_and_start(self, driver, api):
|
|
"""LXC has no reset; stop + start is the closest thing to pulling the plug."""
|
|
driver.reboot_vm("200", force=True)
|
|
status = api.lxc.return_value.status
|
|
status.stop.post.assert_called_once()
|
|
status.start.post.assert_called_once()
|
|
|
|
def test_suspend_vm(self, driver, api):
|
|
driver.suspend_vm("100")
|
|
api.qemu.return_value.status.suspend.post.assert_called_once()
|
|
|
|
def test_suspend_container_is_refused(self, driver, api):
|
|
with pytest.raises(RuntimeError, match="container"):
|
|
driver.suspend_vm("200")
|
|
|
|
def test_api_error_becomes_runtime_error(self, driver, api):
|
|
api.qemu.return_value.status.start.post.side_effect = Exception("locked")
|
|
with pytest.raises(RuntimeError, match="locked"):
|
|
driver.start_vm("100")
|
|
|
|
|
|
QEMU_CONFIG = {
|
|
"name": "web01",
|
|
"cores": 2,
|
|
"sockets": 2,
|
|
"memory": "8192",
|
|
"ostype": "l26",
|
|
"cpu": "host,flags=+aes",
|
|
"bios": "ovmf",
|
|
"machine": "q35",
|
|
"boot": "order=scsi0;ide2;net0",
|
|
"scsi0": "local-lvm:vm-100-disk-0,size=32G,format=raw",
|
|
"virtio1": "tank:vm-100-disk-1,size=512M",
|
|
"ide2": "local:iso/debian.iso,media=cdrom",
|
|
"efidisk0": "local-lvm:vm-100-disk-2,size=4M",
|
|
"net0": "virtio=BC:24:11:AA:BB:CC,bridge=vmbr0,tag=10,firewall=1",
|
|
"net1": "e1000=BC:24:11:AA:BB:DD,bridge=vmbr1",
|
|
"hostpci0": "0000:01:00.0,pcie=1",
|
|
"usb0": "host=1234:5678",
|
|
"description": "Production web server",
|
|
"tags": "prod;web",
|
|
}
|
|
|
|
LXC_CONFIG = {
|
|
"hostname": "dns01",
|
|
"cores": 1,
|
|
"memory": 512,
|
|
"ostype": "debian",
|
|
"rootfs": "local-lvm:vm-200-disk-0,size=8G",
|
|
"mp0": "tank:subvol-200-disk-1,mp=/data,size=1T",
|
|
"net0": "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:00:00:01,ip=dhcp,tag=20,type=veth",
|
|
}
|
|
|
|
|
|
class TestParseQemuConfig:
|
|
@pytest.fixture
|
|
def cfg(self):
|
|
return parse_vm_config("100", "vm", QEMU_CONFIG)
|
|
|
|
def test_core_fields(self, cfg):
|
|
assert cfg["vmid"] == "100"
|
|
assert cfg["name"] == "web01"
|
|
assert cfg["vcpus"] == 4
|
|
assert cfg["memory"] == 8192
|
|
assert cfg["os_type"] == "l26"
|
|
assert cfg["description"] == "Production web server"
|
|
assert cfg["tags"] == ["prod", "web"]
|
|
|
|
def test_boot_order(self, cfg):
|
|
assert cfg["boot_order"] == ["scsi0", "ide2", "net0"]
|
|
|
|
def test_disks_skip_cdrom_and_normalise_size(self, cfg):
|
|
by_dev = {d["device"]: d for d in cfg["disks"]}
|
|
assert set(by_dev) == {"scsi0", "virtio1", "efidisk0"}
|
|
assert by_dev["scsi0"] == {
|
|
"device": "scsi0",
|
|
"storage": "local-lvm",
|
|
"size": 32,
|
|
"format": "raw",
|
|
"bootable": True,
|
|
}
|
|
assert by_dev["virtio1"]["size"] == 0 # 512M rounds down to 0 GB
|
|
assert by_dev["virtio1"]["bootable"] is False
|
|
|
|
def test_nics(self, cfg):
|
|
assert cfg["nics"] == [
|
|
{
|
|
"device": "net0",
|
|
"mac": "BC:24:11:AA:BB:CC",
|
|
"model": "virtio",
|
|
"bridge": "vmbr0",
|
|
"vlan_id": 10,
|
|
},
|
|
{
|
|
"device": "net1",
|
|
"mac": "BC:24:11:AA:BB:DD",
|
|
"model": "e1000",
|
|
"bridge": "vmbr1",
|
|
"vlan_id": 0,
|
|
},
|
|
]
|
|
|
|
def test_hardware_details(self, cfg):
|
|
assert cfg["cpu_type"] == "host"
|
|
assert cfg["sockets"] == 2
|
|
assert cfg["cores_per_socket"] == 2
|
|
assert cfg["firmware"] == "efi"
|
|
assert cfg["machine"] == "q35"
|
|
|
|
def test_passthrough(self, cfg):
|
|
assert cfg["passthrough"] == [
|
|
{"slot": "hostpci0", "kind": "pci", "config": "0000:01:00.0,pcie=1"},
|
|
{"slot": "usb0", "kind": "usb", "config": "host=1234:5678"},
|
|
]
|
|
|
|
def test_defaults_for_a_bare_config(self):
|
|
cfg = parse_vm_config("101", "vm", {})
|
|
assert cfg["name"] == "vm-101"
|
|
assert cfg["vcpus"] == 1
|
|
assert cfg["cpu_type"] == "kvm64"
|
|
assert cfg["firmware"] == "bios"
|
|
assert "machine" not in cfg
|
|
assert cfg["boot_order"] == []
|
|
|
|
def test_legacy_bootdisk(self):
|
|
cfg = parse_vm_config("101", "vm", {"boot": "cdn", "bootdisk": "scsi0"})
|
|
assert cfg["boot_order"] == ["scsi0"]
|
|
|
|
|
|
class TestParseLxcConfig:
|
|
@pytest.fixture
|
|
def cfg(self):
|
|
return parse_vm_config("200", "container", LXC_CONFIG)
|
|
|
|
def test_core_fields(self, cfg):
|
|
assert cfg["name"] == "dns01"
|
|
assert cfg["vcpus"] == 1
|
|
assert cfg["memory"] == 512
|
|
assert cfg["tags"] == []
|
|
|
|
def test_rootfs_and_mountpoint(self, cfg):
|
|
assert [(d["device"], d["storage"], d["size"]) for d in cfg["disks"]] == [
|
|
("rootfs", "local-lvm", 8),
|
|
("mp0", "tank", 1024),
|
|
]
|
|
|
|
def test_veth_nic(self, cfg):
|
|
assert cfg["nics"] == [
|
|
{
|
|
"device": "net0",
|
|
"mac": "BC:24:11:00:00:01",
|
|
"model": "veth",
|
|
"bridge": "vmbr0",
|
|
"vlan_id": 20,
|
|
}
|
|
]
|
|
|
|
def test_no_vm_only_hardware_fields(self, cfg):
|
|
assert "firmware" not in cfg
|
|
assert "sockets" not in cfg
|
|
|
|
|
|
class TestGetVmConfig:
|
|
def test_fetches_the_right_config(self, driver, api):
|
|
api.lxc.return_value.config.get.return_value = LXC_CONFIG
|
|
cfg = driver.get_vm_config("dns01")
|
|
assert cfg["vmid"] == "200"
|
|
assert cfg["name"] == "dns01"
|