"""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") == []