Christian Manivong
c8e472b4c6
Two drivers from one package, both in the hypervisor role: - vmware_esxi talks to one host directly: facts, vmnics and vmkernel NICs, CDP/LLDP neighbours, sensors, VMs, datastores and port groups. - vmware_vcenter talks to a vCenter: every VM of every host it manages, with the host as the VM's node, plus distributed port groups. It reports no interfaces of its own; the hosts' NICs belong to the hosts. Both implement the HypervisorDriver VM contract: get_vms, get_vm_config, start/stop/reboot/suspend_vm and the four snapshot methods, and emit raw device warnings (maintenance mode, disconnected host, config issues, host managed by a vCenter, free license making the API read-only). Every read is a PropertyCollector query for the explicit paths in paths.py, converted by to_plain() into dicts and lists; the parsers only ever see that. tools/harvest.py dumps exactly those paths to JSON and tools/sanitize.py scrubs the dump, so a real host can become a test fixture without code changes. A VM's vmid is its instance UUID, which survives vMotion and re-registration; a MoRef does not. Tested against govmomi's vcsim in ESXi and vCenter mode, including real power and snapshot tasks. Not yet tested against real hardware.
132 lines
5.1 KiB
Python
132 lines
5.1 KiB
Python
"""get_vms: VMDict plus the extra keys netOrk's VM sync reads."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from napalm_vmware.parse.networks import network_index
|
|
from napalm_vmware.parse.vms import vm_list
|
|
|
|
|
|
def _vms(data):
|
|
objs = data["objects"]
|
|
index = network_index(objs["HostSystem"], objs.get("DistributedVirtualPortgroup", []))
|
|
return vm_list(objs["VirtualMachine"], objs["HostSystem"], index)
|
|
|
|
|
|
@pytest.fixture
|
|
def vm0(esxi_data):
|
|
return esxi_data["objects"]["VirtualMachine"][0]
|
|
|
|
|
|
class TestVmListFromVcsim:
|
|
def test_one_entry_per_vm(self, esxi_data):
|
|
assert [vm["name"] for vm in _vms(esxi_data)] == ["ha-host_VM0", "ha-host_VM1"]
|
|
|
|
def test_contract_fields(self, esxi_data, vm0):
|
|
vm = _vms(esxi_data)[0]
|
|
assert vm["vmid"] == vm0["config.instanceUuid"]
|
|
assert vm["moref"] == vm0["_moref"]
|
|
assert vm["status"] == "running"
|
|
assert vm["vcpus"] == vm0["config.hardware.numCPU"]
|
|
assert vm["memory"] == vm0["config.hardware.memoryMB"]
|
|
assert vm["node"] == "localhost.localdomain"
|
|
assert vm["type"] == "vm"
|
|
|
|
def test_vcenter_node_is_the_esxi_host(self, vcenter_data):
|
|
nodes = {vm["name"]: vm["node"] for vm in _vms(vcenter_data)}
|
|
assert nodes["DC0_H0_VM0"] == "DC0_H0"
|
|
|
|
def test_nic_on_a_distributed_portgroup(self, vcenter_data):
|
|
vm = _vms(vcenter_data)[0]
|
|
(iface,) = vm["interfaces"].values()
|
|
assert iface["bridge"] == "DC0_DVPG0"
|
|
assert iface["tag"] is None
|
|
assert iface["mac_address"] == "00:0C:29:63:65:61"
|
|
|
|
def test_disks_in_megabytes(self, vcenter_data):
|
|
vm = _vms(vcenter_data)[0]
|
|
assert vm["disks"] == [{"name": "disk-202-0", "size_mb": 10240}]
|
|
|
|
def test_tools_not_installed(self, esxi_data):
|
|
vm = _vms(esxi_data)[0]
|
|
assert vm["agent_enabled"] is False
|
|
assert vm["agent_running"] is False
|
|
|
|
|
|
class TestVmListBranches:
|
|
def test_templates_are_not_vms(self, esxi_data, vm0):
|
|
vm0["config.template"] = True
|
|
assert [vm["name"] for vm in _vms(esxi_data)] == ["ha-host_VM1"]
|
|
|
|
def test_vm_without_config_is_skipped(self, esxi_data, vm0):
|
|
"""An orphaned or inaccessible VM has no instanceUuid to identify it by."""
|
|
del vm0["config.instanceUuid"]
|
|
assert len(_vms(esxi_data)) == 1
|
|
|
|
@pytest.mark.parametrize(
|
|
("power", "status"),
|
|
[("poweredOn", "running"), ("poweredOff", "stopped"), ("suspended", "suspended")],
|
|
)
|
|
def test_power_state(self, esxi_data, vm0, power, status):
|
|
vm0["runtime.powerState"] = power
|
|
assert _vms(esxi_data)[0]["status"] == status
|
|
|
|
def test_usage_from_quickstats(self, esxi_data, esxi_host, vm0):
|
|
esxi_host["summary.hardware"]["cpuMhz"] = 2000
|
|
vm0["config.hardware.numCPU"] = 2
|
|
vm0["summary.quickStats"].update(
|
|
overallCpuUsage=1000, guestMemoryUsage=512, uptimeSeconds=3600
|
|
)
|
|
vm = _vms(esxi_data)[0]
|
|
assert vm["cpu_usage"] == 0.25
|
|
assert vm["memory_usage"] == 512
|
|
assert vm["uptime"] == 3600
|
|
|
|
def test_cpu_usage_is_capped_at_one(self, esxi_data, esxi_host, vm0):
|
|
esxi_host["summary.hardware"]["cpuMhz"] = 1000
|
|
vm0["config.hardware.numCPU"] = 1
|
|
vm0["summary.quickStats"]["overallCpuUsage"] = 1500
|
|
assert _vms(esxi_data)[0]["cpu_usage"] == 1.0
|
|
|
|
def test_tools_running_and_guest_ip(self, esxi_data, vm0):
|
|
vm0["guest.toolsStatus"] = "toolsOk"
|
|
vm0["guest.toolsRunningStatus"] = "guestToolsRunning"
|
|
vm0["guest.net"][0]["ipAddress"] = ["fe80::1", "192.0.2.50"]
|
|
vm = _vms(esxi_data)[0]
|
|
assert vm["agent_enabled"] is True
|
|
assert vm["agent_running"] is True
|
|
assert vm["ipv4"] == "192.0.2.50"
|
|
(iface,) = vm["interfaces"].values()
|
|
assert iface["ipv4"] == "192.0.2.50"
|
|
assert iface["is_up"] is True
|
|
|
|
def test_guest_primary_ip_wins(self, esxi_data, vm0):
|
|
vm0["guest.ipAddress"] = "192.0.2.99"
|
|
vm0["guest.net"][0]["ipAddress"] = ["192.0.2.50"]
|
|
assert _vms(esxi_data)[0]["ipv4"] == "192.0.2.99"
|
|
|
|
def test_ipv6_primary_ip_is_not_an_ipv4(self, esxi_data, vm0):
|
|
vm0["guest.ipAddress"] = "2001:db8::5"
|
|
assert _vms(esxi_data)[0]["ipv4"] == ""
|
|
|
|
def test_vlan_of_a_standard_portgroup_becomes_the_tag(self, esxi_data, esxi_host):
|
|
esxi_host["config.network.portgroup"][0]["spec"]["vlanId"] = 30
|
|
vm = _vms(esxi_data)[0]
|
|
(iface,) = vm["interfaces"].values()
|
|
assert iface["bridge"] == "VM Network"
|
|
assert iface["tag"] == 30
|
|
|
|
def test_autostart(self, esxi_data, esxi_host, vm0):
|
|
esxi_host["config.autoStart"]["defaults"]["enabled"] = True
|
|
esxi_host["config.autoStart"]["powerInfo"] = [
|
|
{"key": vm0["_moref"], "startAction": "powerOn", "startOrder": 1}
|
|
]
|
|
assert [vm["onboot"] for vm in _vms(esxi_data)] == [True, False]
|
|
|
|
def test_autostart_entry_ignored_while_the_manager_is_off(self, esxi_data, esxi_host, vm0):
|
|
esxi_host["config.autoStart"]["powerInfo"] = [
|
|
{"key": vm0["_moref"], "startAction": "powerOn", "startOrder": 1}
|
|
]
|
|
assert _vms(esxi_data)[0]["onboot"] is False
|