napalm-proxmox/tests/test_vm_guest_os.py

"""Provisioning guests other than Linux: FreeBSD and OpenBSD (NetOrk/netork#794).

What each guest needs was found on FreeBSD 15.1 and OpenBSD 7.9 cloud images
(NetOrk/netork#793).
"""

from __future__ import annotations

import base64
from unittest.mock import MagicMock, patch

import pytest
import yaml
from napalm_device_types.provisioning import QEMU_GUEST_AGENTS

from napalm_proxmox.driver import ProxmoxDriver
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin

_FREEBSD_URL = (
    "https://download.freebsd.org/releases/VM-IMAGES/15.1-RELEASE/amd64/Latest/"
    "FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2.xz"
)
_DEBIAN_URL = (
    "https://cloud.debian.org/images/cloud/trixie/latest/debian-13-genericcloud-amd64.qcow2"
)


def _mixin(net0: str = "virtio=BC:24:11:AA:BB:02,bridge=vmbr0") -> tuple:
    """A mixin whose node answers like Proxmox; the VM config carries *net0*."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"
    api = MagicMock()
    api.cluster.nextid.get.return_value = 101
    api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
    node = MagicMock()
    node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
        {"storage": "local", "type": "dir", "content": "snippets,iso", "enabled": 1},
    ]
    vm = MagicMock()
    node.qemu.return_value = vm
    vm.config.get.return_value = {"unused0": "local-lvm:vm-101-disk-0", "net0": net0}
    vm.status.start.post.return_value = "UPID:pve1:start"
    node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    mixin._api = api
    mixin._node_api = MagicMock(return_value=node)
    mixin._download_cloud_image = MagicMock(
        return_value="/var/lib/vz/template/netork-images/x.qcow2"
    )
    mixin._run_node_command = MagicMock(return_value="")
    return mixin, node, vm


def _create(mixin, guest_os: str, image_url: str = _DEBIAN_URL, **kwargs):
    with patch("time.sleep"):
        return mixin.create_vm_from_cloud_init(
            name="bsd-vm",
            image_url=image_url,
            cpu=2,
            memory=2048,
            nics=[{"bridge": "vmbr0"}],
            cloud_init_config={"hostname": "bsd-vm"},
            guest_os=guest_os,
            **kwargs,
        )


def _all_config_posts(vm) -> dict:
    merged: dict = {}
    for call in vm.config.post.call_args_list:
        merged.update(call.kwargs)
    return merged


def _written_snippet(mixin, name: str) -> str | None:
    """The content of the snippet *name* the driver wrote over SSH, if any."""
    for call in mixin._run_node_command.call_args_list:
        command = call.args[0]
        if f"snippets/{name}" in command and "base64 -d" in command:
            encoded = command.split("echo ", 1)[1].split(" |", 1)[0]
            return base64.b64decode(encoded).decode()
    return None


class TestWhichGuestsProxmoxProvisions:
    def test_all_qemu_guests(self):
        assert ProxmoxDriver.GUEST_AGENTS == QEMU_GUEST_AGENTS

    def test_an_unknown_guest_is_refused_before_anything_is_created(self):
        mixin, node, _ = _mixin()
        with pytest.raises(ValueError, match="windows"):
            _create(mixin, "windows")
        mixin._api.cluster.nextid.get.assert_not_called()
        node.qemu.post.assert_not_called()


class TestVmShellPerGuest:
    @pytest.mark.parametrize(
        ("guest_os", "ostype"), [("linux", "l26"), ("freebsd", "other"), ("openbsd", "other")]
    )
    def test_ostype(self, guest_os, ostype):
        mixin, node, _ = _mixin()
        _create(mixin, guest_os)
        assert node.qemu.post.call_args.kwargs["ostype"] == ostype

    @pytest.mark.parametrize("guest_os", ["linux", "freebsd"])
    def test_agent_over_virtio_serial(self, guest_os):
        mixin, node, _ = _mixin()
        _create(mixin, guest_os)
        shell = node.qemu.post.call_args.kwargs
        assert shell["agent"] == "1"
        assert "serial0" not in shell

    def test_openbsd_agent_on_the_second_isa_serial_port(self):
        """OpenBSD's qemu-ga cannot use virtio-serial, and the image keeps its
        console on com0: the agent only answers as cua01, behind serial0."""
        mixin, node, _ = _mixin()
        _create(mixin, "openbsd")
        shell = node.qemu.post.call_args.kwargs
        assert shell["agent"] == "1,type=isa"
        assert shell["serial0"] == "socket"


class TestNetworkConfigPerGuest:
    def test_linux_keeps_proxmoxs_own_network_config(self):
        mixin, _, vm = _mixin()
        _create(mixin, "linux")
        config = _all_config_posts(vm)
        assert config["cicustom"] == "user=local:snippets/101-user-data.yaml"
        assert config["ipconfig0"] == "ip=dhcp"
        assert _written_snippet(mixin, "101-network-config.yaml") is None

    @pytest.mark.parametrize("guest_os", ["freebsd", "openbsd"])
    def test_a_bsd_guest_gets_a_v2_network_config_matched_by_its_mac(self, guest_os):
        """FreeBSD's nuageinit fails on the v1 Proxmox writes and then skips
        runcmd, which is what starts the guest agent."""
        mixin, _, vm = _mixin(net0="virtio=BC:24:11:AA:BB:02,bridge=vmbr0")
        _create(mixin, guest_os)
        config = _all_config_posts(vm)
        assert config["cicustom"] == (
            "user=local:snippets/101-user-data.yaml,network=local:snippets/101-network-config.yaml"
        )
        assert yaml.safe_load(_written_snippet(mixin, "101-network-config.yaml")) == {
            "version": 2,
            "ethernets": {"nic0": {"match": {"macaddress": "bc:24:11:aa:bb:02"}, "dhcp4": True}},
        }


class TestPackedImages:
    """Proxmox's download-url unpacks ISOs only, so a packed image goes over SSH."""

    def test_an_xz_image_is_unpacked_on_the_node_and_the_unpacked_file_imported(self):
        mixin, _, _ = _mixin()
        mixin._run_node_command = MagicMock(
            side_effect=lambda cmd, timeout: "MISSING" if "test -f" in cmd else ""
        )
        mixin._download_cloud_image = MagicMock(
            return_value="/var/lib/vz/template/netork-images/k-FreeBSD-15.1-ufs.qcow2.xz"
        )
        _create(mixin, "freebsd", image_url=_FREEBSD_URL, image_checksum="sha256:abc")

        mixin._download_cloud_image.assert_called_once()
        assert mixin._download_cloud_image.call_args.args[1] == "sha256:abc"  # the .xz is verified
        commands = [c.args[0] for c in mixin._run_node_command.call_args_list]
        unpack = next(c for c in commands if c.startswith("xz -dc "))
        assert "k-FreeBSD-15.1-ufs.qcow2.xz" in unpack
        importdisk = next(c for c in commands if c.startswith("qm importdisk"))
        assert ".qcow2.xz" not in importdisk
        assert ".qcow2 " in importdisk

    def test_an_unpacked_image_already_on_the_node_is_not_downloaded_again(self):
        mixin, _, _ = _mixin()
        mixin._run_node_command = MagicMock(
            side_effect=lambda cmd, timeout: "EXISTS" if "test -f" in cmd else ""
        )
        _create(mixin, "freebsd", image_url=_FREEBSD_URL)

        mixin._download_cloud_image.assert_not_called()
        commands = [c.args[0] for c in mixin._run_node_command.call_args_list]
        assert not any(c.startswith("xz -dc ") for c in commands)

    def test_a_packed_image_never_goes_through_an_import_storage(self):
        mixin, node, _ = _mixin()
        node.storage.get.return_value = [
            {"storage": "local-lvm", "content": "images,rootdir", "enabled": 1},
            {"storage": "local", "content": "snippets,import", "enabled": 1},
        ]
        mixin._run_node_command = MagicMock(
            side_effect=lambda cmd, timeout: "MISSING" if "test -f" in cmd else ""
        )
        _create(mixin, "freebsd", image_url=_FREEBSD_URL)
        # storage(name)("download-url").post(...) is the import-storage download.
        node.storage.return_value.return_value.post.assert_not_called()
        mixin._download_cloud_image.assert_called_once()


class TestLoopbackIsNeverTheVmsAddress:
    """OpenBSD's agent lists lo0 first (#793); Linux names it lo."""

    def _status(self, interfaces: list) -> dict:
        mixin = ProxmoxVMProvisionMixin()
        node = MagicMock()
        mixin._node_api = MagicMock(return_value=node)
        node.qemu.return_value.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}
        node.qemu.return_value.agent.return_value.get.return_value = {"result": interfaces}
        with patch("time.sleep"):
            return mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)

    def test_openbsd_answer(self):
        """The raw answer of OpenBSD 7.9's qemu-ga, from the spike."""
        result = self._status(
            [
                {
                    "name": "lo0",
                    "ip-addresses": [
                        {"ip-address-type": "ipv6", "ip-address": "::1", "prefix": 128},
                        {"ip-address-type": "ipv6", "ip-address": "fe80:3::1", "prefix": 64},
                        {"ip-address-type": "ipv4", "ip-address": "127.0.0.1", "prefix": 8},
                    ],
                    "hardware-address": "00:00:00:00:00:00",
                },
                {
                    "name": "vio0",
                    "ip-addresses": [
                        {"ip-address-type": "ipv4", "ip-address": "10.0.2.15", "prefix": 24}
                    ],
                    "hardware-address": "bc:24:11:aa:bb:04",
                },
                {"name": "enc0", "hardware-address": "00:00:00:00:00:00"},
                {"name": "pflog0", "hardware-address": "00:00:00:00:00:00"},
            ]
        )
        assert result["ip_address"] == "10.0.2.15"
        assert result["mac_address"] == "bc:24:11:aa:bb:04"

    def test_any_loopback_address_is_skipped_whatever_the_interface_is_called(self):
        result = self._status(
            [
                {
                    "name": "lo1",
                    "ip-addresses": [{"ip-address-type": "ipv4", "ip-address": "127.0.0.2"}],
                },
                {
                    "name": "vtnet0",
                    "ip-addresses": [{"ip-address-type": "ipv4", "ip-address": "10.0.0.5"}],
                },
            ]
        )
        assert result["ip_address"] == "10.0.0.5"


class TestDestroyRemovesEverySnippet:
    def test_user_and_network_snippets_are_read_before_the_vm_is_deleted(self):
        """After the delete the config is gone, and the snippets stayed behind
        (napalm-proxmox#15)."""
        mixin = ProxmoxVMProvisionMixin()
        node = MagicMock()
        mixin._node_api = MagicMock(return_value=node)
        vm = node.qemu.return_value
        order: list[str] = []
        vm.config.get.side_effect = lambda: (
            order.append("config")
            or {
                "cicustom": "user=local:snippets/101-user-data.yaml,"
                "network=local:snippets/101-network-config.yaml"
            }
        )
        vm.delete.side_effect = lambda **kw: order.append("delete")
        node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}

        with patch("time.sleep"):
            mixin.destroy_vm("101")

        assert order.index("config") < order.index("delete")
        deleted = [c.args[0] for c in node.storage.return_value.content.call_args_list]
        assert deleted == ["snippets/101-user-data.yaml", "snippets/101-network-config.yaml"]