napalm-vmware/tests/test_parse_vm_config.py

"""get_vm_config and snapshot trees."""

from __future__ import annotations

import pytest

from napalm_vmware.parse.networks import NetworkIndex, network_index
from napalm_vmware.parse.snapshots import find_snapshot, snapshot_list
from napalm_vmware.parse.vm_config import vm_config


@pytest.fixture
def vm(vcenter_data):
    return vcenter_data["objects"]["VirtualMachine"][0]


@pytest.fixture
def index(vcenter_data):
    objs = vcenter_data["objects"]
    return network_index(objs["HostSystem"], objs["DistributedVirtualPortgroup"])


class TestVmConfig:
    def test_core_fields_from_vcsim(self, vm, index):
        cfg = vm_config(vm, index)
        assert cfg["name"] == "DC0_H0_VM0"
        assert cfg["vmid"] == vm["config.instanceUuid"]
        assert cfg["vcpus"] == vm["config.hardware.numCPU"]
        assert cfg["memory"] == vm["config.hardware.memoryMB"]
        assert cfg["os_type"] == vm["config.guestId"]
        assert cfg["os_name"] == vm["config.guestFullName"]
        assert cfg["description"] == ""
        assert cfg["tags"] == []
        assert cfg["firmware"] == "bios"
        assert cfg["machine"] == "vmx-13"
        assert cfg["passthrough"] == []

    def test_disk(self, vm, index):
        assert vm_config(vm, index)["disks"] == [
            {
                "device": "disk-202-0",
                "storage": "LocalDS_0",
                "size": 10,
                "format": "vmdk",
                "bootable": True,  # no boot order: the BIOS boots the first disk
            }
        ]

    def test_nic(self, vm, index):
        assert vm_config(vm, index)["nics"] == [
            {
                "device": "ethernet-0",
                "mac": "00:0C:29:63:65:61",
                "model": "e1000",
                "bridge": "DC0_DVPG0",
                "vlan_id": 0,
            }
        ]

    def test_cpu_topology(self, vm, index):
        vm["config.hardware.numCPU"] = 8
        vm["config.hardware.numCoresPerSocket"] = 4
        cfg = vm_config(vm, index)
        assert (cfg["sockets"], cfg["cores_per_socket"]) == (2, 4)

    def test_explicit_boot_order(self, vm, index):
        disk_key = next(
            d["key"] for d in vm["config.hardware.device"] if d["_type"] == "VirtualDisk"
        )
        nic_key = next(
            d["key"] for d in vm["config.hardware.device"] if d["_type"] == "VirtualE1000"
        )
        vm["config.bootOptions.bootOrder"] = [
            {"_type": "VirtualMachineBootOptionsBootableCdromDevice"},
            {"_type": "VirtualMachineBootOptionsBootableEthernetDevice", "deviceKey": nic_key},
            {"_type": "VirtualMachineBootOptionsBootableDiskDevice", "deviceKey": disk_key},
        ]
        cfg = vm_config(vm, index)
        assert cfg["boot_order"] == ["cdrom", "ethernet-0", "disk-202-0"]
        assert cfg["disks"][0]["bootable"] is True

    def test_disk_outside_the_boot_order_is_not_bootable(self, vm, index):
        nic_key = next(
            d["key"] for d in vm["config.hardware.device"] if d["_type"] == "VirtualE1000"
        )
        vm["config.bootOptions.bootOrder"] = [
            {"_type": "VirtualMachineBootOptionsBootableEthernetDevice", "deviceKey": nic_key},
        ]
        assert vm_config(vm, index)["disks"][0]["bootable"] is False

    def test_passthrough_and_annotation(self, vm, index):
        vm["config.annotation"] = "Production web server"
        vm["config.hardware.device"].append(
            {
                "_type": "VirtualPCIPassthrough",
                "key": 13000,
                "deviceInfo": {"label": "PCI device 0", "summary": "NVIDIA T4"},
            }
        )
        cfg = vm_config(vm, index)
        assert cfg["description"] == "Production web server"
        assert cfg["passthrough"] == [
            {"slot": "PCI device 0", "kind": "pci", "config": "NVIDIA T4"}
        ]

    def test_config_less_vm_is_refused(self, vm):
        del vm["config.instanceUuid"]
        with pytest.raises(ValueError):
            vm_config(vm, NetworkIndex())


def _snap(name, moref, created, children=(), state="poweredOff", description=""):
    return {
        "name": name,
        "snapshot": moref,
        "createTime": created,
        "state": state,
        "description": description,
        "childSnapshotList": list(children),
    }


@pytest.fixture
def tree_vm():
    return {
        "name": "web01",
        "snapshot": {
            "currentSnapshot": "snapshot-3",
            "rootSnapshotList": [
                _snap(
                    "base",
                    "snapshot-1",
                    100.0,
                    [
                        _snap("upgrade", "snapshot-2", 200.0, state="poweredOn", description="d"),
                        _snap("again", "snapshot-3", 300.0),
                    ],
                ),
                _snap("again", "snapshot-4", 400.0),
            ],
        },
    }


class TestSnapshots:
    def test_flattened_depth_first_with_parents(self, tree_vm):
        assert snapshot_list(tree_vm) == [
            {
                "name": "base",
                "vm": "web01",
                "created": 100.0,
                "description": "",
                "has_memory": False,
                "parent": "",
            },
            {
                "name": "upgrade",
                "vm": "web01",
                "created": 200.0,
                "description": "d",
                "has_memory": True,
                "parent": "base",
            },
            {
                "name": "again",
                "vm": "web01",
                "created": 300.0,
                "description": "",
                "has_memory": False,
                "parent": "base",
            },
            {
                "name": "again",
                "vm": "web01",
                "created": 400.0,
                "description": "",
                "has_memory": False,
                "parent": "",
            },
        ]

    def test_vm_without_snapshots(self):
        assert snapshot_list({"name": "x"}) == []

    def test_find_by_name(self, tree_vm):
        assert find_snapshot(tree_vm, "upgrade") == "snapshot-2"

    def test_unknown_name(self, tree_vm):
        with pytest.raises(ValueError, match="no snapshot"):
            find_snapshot(tree_vm, "nope")

    def test_duplicate_names_are_refused_not_guessed(self, tree_vm):
        """vSphere allows two snapshots with one name. Picking one could roll
        back or delete the wrong state."""
        with pytest.raises(ValueError, match="2 snapshots"):
            find_snapshot(tree_vm, "again")