napalm-vmware/tests/test_drivers.py

"""The two driver classes: identity, connection handling and getters."""

from __future__ import annotations

from importlib.metadata import entry_points
from unittest.mock import MagicMock, patch

import pytest
from napalm.base.exceptions import ConnectionException
from napalm_device_types.roles import role_keys_of

from napalm_vmware import VmwareEsxiDriver, VmwareVcenterDriver
from tests.conftest import FakeInventory

DRIVERS = [VmwareEsxiDriver, VmwareVcenterDriver]


def _driver(cls, data, hostname="esx01.example.lan"):
    driver = cls(hostname, "root", "secret")
    driver._si = MagicMock()
    driver._inventory = FakeInventory(data)
    return driver


@pytest.fixture
def esxi(esxi_data):
    return _driver(VmwareEsxiDriver, esxi_data)


@pytest.fixture
def vcenter(vcenter_data):
    return _driver(VmwareVcenterDriver, vcenter_data, hostname="vc01.example.lan")


class TestIdentity:
    @pytest.mark.parametrize("cls", DRIVERS)
    def test_entry_point_name_is_the_driver_name(self, cls):
        (ep,) = [e for e in entry_points(group="napalm.drivers") if e.name == cls.DRIVER_NAME]
        assert ep.load() is cls

    @pytest.mark.parametrize("cls", DRIVERS)
    def test_is_a_hypervisor_and_nothing_else(self, cls):
        assert role_keys_of(cls) == ["hypervisor"]

    @pytest.mark.parametrize("cls", DRIVERS)
    def test_talks_api_not_ssh(self, cls):
        assert cls.USES_SSH is False

    @pytest.mark.parametrize("cls", DRIVERS)
    def test_never_fingerprints_on_the_vmware_oui(self, cls):
        """00:50:56 is the OUI of *guest* NICs; every VM would score as a host."""
        assert not any(p.startswith("00:50:56") for p in cls.OUI_PREFIXES)

    def test_fingerprints_exclude_each_other(self):
        esxi = {r.pattern: r for r in VmwareEsxiDriver.HTTP_FINGERPRINT}
        vc = {r.pattern: r for r in VmwareVcenterDriver.HTTP_FINGERPRINT}
        assert esxi["vmware esxi"].mandatory and vc["vmware esxi"].negative
        assert vc["vcenter"].mandatory and esxi["vcenter"].negative

    @pytest.mark.parametrize(
        "method",
        [
            "get_vms",
            "get_vm_config",
            "start_vm",
            "stop_vm",
            "reboot_vm",
            "suspend_vm",
            "get_vm_snapshots",
            "create_vm_snapshot",
            "delete_vm_snapshot",
            "rollback_vm_snapshot",
            "get_vm_storage_pools",
            "get_virtual_networks",
            "get_device_warnings",
        ],
    )
    @pytest.mark.parametrize("cls", DRIVERS)
    def test_implements_the_v1_contract(self, cls, method):
        assert callable(getattr(cls, method, None))

    @pytest.mark.parametrize("cls", DRIVERS)
    @pytest.mark.parametrize(
        "method",
        [
            "create_vm_from_cloud_init",
            "destroy_vm",
            "get_vm_status",
            "get_network_targets",
            "get_image_storages",
        ],
    )
    def test_implements_provisioning(self, cls, method):
        assert callable(getattr(cls, method, None))

    @pytest.mark.parametrize("cls", DRIVERS)
    def test_declares_its_guest_agent(self, cls):
        assert cls.GUEST_AGENT_PACKAGES == ("open-vm-tools",)


class TestOptionalArgs:
    def test_defaults(self):
        d = VmwareEsxiDriver("h", "u", "p")
        assert (d._port, d._verify_ssl) == (443, True)

    def test_netork_style_args(self):
        d = VmwareEsxiDriver("h", "u", "p", optional_args={"port": "8443", "ssl_verify": False})
        assert (d._port, d._verify_ssl) == (8443, False)

    def test_verify_ssl_wins_over_ssl_verify(self):
        d = VmwareEsxiDriver("h", "u", "p", optional_args={"ssl_verify": True, "verify_ssl": False})
        assert d._verify_ssl is False

    def test_ssh_keys_netork_adds_are_ignored(self):
        VmwareEsxiDriver("h", "u", "p", optional_args={"ssh_username": "x", "ssh_port": 22})


class TestConnection:
    def _open(self, cls, data, connect_error=None):
        driver = cls("host", "u", "p")
        si = MagicMock()
        with (
            patch(
                "napalm_vmware.base._session.connect", return_value=si, side_effect=connect_error
            ) as connect,
            patch("napalm_vmware.base._session.disconnect") as disconnect,
            patch("napalm_vmware.base.Inventory", return_value=FakeInventory(data)),
        ):
            try:
                driver.open()
            finally:
                self.connect, self.disconnect = connect, disconnect
        return driver, si

    def test_open_esxi(self, esxi_data):
        driver, si = self._open(VmwareEsxiDriver, esxi_data)
        assert driver._si is si
        self.connect.assert_called_once_with("host", 443, "u", "p", verify_ssl=True, timeout=60)

    def test_esxi_driver_refuses_a_vcenter(self, vcenter_data):
        with pytest.raises(ConnectionException, match="vmware_vcenter"):
            self._open(VmwareEsxiDriver, vcenter_data)
        self.disconnect.assert_called_once()

    def test_vcenter_driver_refuses_an_esxi_host(self, esxi_data):
        with pytest.raises(ConnectionException, match="vmware_esxi"):
            self._open(VmwareVcenterDriver, esxi_data)

    def test_connect_failure_is_a_connection_exception(self, esxi_data):
        with pytest.raises(ConnectionException, match="host"):
            self._open(VmwareEsxiDriver, esxi_data, connect_error=OSError("refused"))

    def test_close(self, esxi):
        si = esxi._si
        with patch("napalm_vmware.base._session.disconnect") as disconnect:
            esxi.close()
            esxi.close()  # idempotent
        disconnect.assert_called_once_with(si)
        assert esxi._inventory is None

    def test_is_alive(self, esxi):
        with patch("napalm_vmware.base._session.alive", return_value=True):
            assert esxi.is_alive() == {"is_alive": True}
        esxi._si = None
        assert esxi.is_alive() == {"is_alive": False}


class TestEsxiGetters:
    def test_facts(self, esxi):
        assert esxi.get_facts()["hostname"] == "localhost"

    def test_interfaces(self, esxi):
        assert set(esxi.get_interfaces()) == {"vmnic0", "vmnic1", "vmk0"}
        assert esxi.get_interfaces_ip() == {}

    def test_vms(self, esxi):
        assert [vm["name"] for vm in esxi.get_vms()] == ["ha-host_VM0", "ha-host_VM1"]

    def test_lldp_via_query_network_hint(self, esxi, esxi_host):
        esxi_host["configManager.networkSystem"] = "networkSystem"
        network_system = MagicMock()
        network_system.QueryNetworkHint.return_value = []
        esxi._mo = MagicMock(return_value=network_system)
        assert esxi.get_lldp_neighbors() == {}
        esxi._mo.assert_called_once()

    def test_lldp_unsupported_is_empty_not_an_error(self, esxi, esxi_host):
        from pyVmomi import vmodl

        esxi_host["configManager.networkSystem"] = "networkSystem"
        network_system = MagicMock()
        network_system.QueryNetworkHint.side_effect = vmodl.fault.NotSupported()
        esxi._mo = MagicMock(return_value=network_system)
        assert esxi.get_lldp_neighbors() == {}

    def test_no_host_visible(self, esxi):
        esxi._inventory.objects("HostSystem").clear()
        with pytest.raises(RuntimeError, match="no host"):
            esxi.get_facts()

    def test_warnings(self, esxi, esxi_host):
        esxi_host["runtime.inMaintenanceMode"] = True
        assert [w["code"] for w in esxi.get_device_warnings()] == ["vmware_host_maintenance_mode"]

    def test_storage_and_networks(self, esxi):
        assert list(esxi.get_vm_storage_pools())
        assert "VM Network" in esxi.get_virtual_networks()

    def test_environment(self, esxi):
        assert "0" in esxi.get_environment()["cpu"]


class TestVcenterGetters:
    def test_facts(self, vcenter):
        assert vcenter.get_facts()["hostname"] == "vc01"

    def test_host_nics_are_not_the_vcenters(self, vcenter):
        """Reporting every host's vmnics on the vCenter device would attach
        their MACs to the wrong device in netOrk's topology."""
        assert vcenter.get_interfaces() == {}
        assert vcenter.get_interfaces_ip() == {}
        assert vcenter.get_lldp_neighbors() == {}

    def test_vms_span_all_hosts(self, vcenter):
        assert len({vm["node"] for vm in vcenter.get_vms()}) > 1

    def test_distributed_portgroups(self, vcenter):
        assert vcenter.get_virtual_networks()["DC0_DVPG0"]["type"] == "dvportgroup"


class TestFindVm:
    def test_by_name_vmid_or_moref(self, esxi, esxi_data):
        vm0 = esxi_data["objects"]["VirtualMachine"][0]
        for key in (vm0["name"], vm0["config.instanceUuid"], vm0["_moref"]):
            assert esxi.get_vm_config(key)["vmid"] == vm0["config.instanceUuid"]

    def test_unknown(self, esxi):
        with pytest.raises(ValueError, match="no VM"):
            esxi.get_vm_config("nope")

    def test_ambiguous_name(self, esxi, esxi_data):
        vms = esxi_data["objects"]["VirtualMachine"]
        vms[1]["name"] = vms[0]["name"]
        with pytest.raises(ValueError, match="2 VMs"):
            esxi.get_vm_config(vms[0]["name"])

    def test_templates_are_not_found(self, esxi, esxi_data):
        vm0 = esxi_data["objects"]["VirtualMachine"][0]
        vm0["config.template"] = True
        with pytest.raises(ValueError):
            esxi.get_vm_config(vm0["name"])

    def test_snapshots(self, esxi):
        assert esxi.get_vm_snapshots("ha-host_VM0") == []