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.
135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
"""Power and snapshot actions: which vSphere call each one makes, and refusals."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from napalm_vmware import VmwareEsxiDriver
|
|
from tests.conftest import FakeInventory
|
|
|
|
|
|
@pytest.fixture
|
|
def vm0(esxi_data):
|
|
return esxi_data["objects"]["VirtualMachine"][0]
|
|
|
|
|
|
@pytest.fixture
|
|
def driver(esxi_data):
|
|
d = VmwareEsxiDriver("esx01", "root", "secret")
|
|
d._si = MagicMock()
|
|
d._inventory = FakeInventory(esxi_data)
|
|
d.mo = MagicMock(name="vm")
|
|
d._mo = MagicMock(return_value=d.mo)
|
|
return d
|
|
|
|
|
|
@pytest.fixture
|
|
def waited():
|
|
with patch("napalm_vmware.actions.wait_for_task") as wait:
|
|
yield wait
|
|
|
|
|
|
class TestPower:
|
|
def test_start(self, driver, vm0, waited):
|
|
vm0["runtime.powerState"] = "poweredOff"
|
|
driver.start_vm(vm0["name"])
|
|
driver.mo.PowerOnVM_Task.assert_called_once_with()
|
|
assert waited.call_args.args[1] is driver.mo.PowerOnVM_Task.return_value
|
|
|
|
def test_forced_stop_powers_off(self, driver, vm0, waited):
|
|
driver.stop_vm(vm0["name"], force=True)
|
|
driver.mo.PowerOffVM_Task.assert_called_once_with()
|
|
|
|
def test_graceful_stop_asks_tools_and_waits_for_power_off(self, driver, vm0):
|
|
vm0["guest.toolsRunningStatus"] = "guestToolsRunning"
|
|
with patch("napalm_vmware.actions.wait_until") as until:
|
|
driver.stop_vm(vm0["name"])
|
|
driver.mo.ShutdownGuest.assert_called_once_with()
|
|
driver._inventory.data["objects"]["VirtualMachine"][0]["runtime.powerState"] = "poweredOff"
|
|
driver._inventory.properties = lambda obj, paths: {"runtime.powerState": "poweredOff"}
|
|
check = until.call_args.args[0]
|
|
assert check() is True
|
|
|
|
def test_graceful_stop_without_tools_is_refused(self, driver, vm0):
|
|
with pytest.raises(RuntimeError, match="VMware Tools"):
|
|
driver.stop_vm(vm0["name"])
|
|
driver.mo.ShutdownGuest.assert_not_called()
|
|
|
|
def test_stopping_a_stopped_vm_is_refused(self, driver, vm0):
|
|
vm0["runtime.powerState"] = "poweredOff"
|
|
with pytest.raises(RuntimeError, match="not running"):
|
|
driver.stop_vm(vm0["name"], force=True)
|
|
|
|
def test_graceful_reboot(self, driver, vm0):
|
|
vm0["guest.toolsRunningStatus"] = "guestToolsRunning"
|
|
driver.reboot_vm(vm0["name"])
|
|
driver.mo.RebootGuest.assert_called_once_with()
|
|
|
|
def test_forced_reboot_resets(self, driver, vm0, waited):
|
|
driver.reboot_vm(vm0["name"], force=True)
|
|
driver.mo.ResetVM_Task.assert_called_once_with()
|
|
|
|
def test_suspend(self, driver, vm0, waited):
|
|
driver.suspend_vm(vm0["name"])
|
|
driver.mo.SuspendVM_Task.assert_called_once_with()
|
|
|
|
def test_suspend_requires_a_running_vm(self, driver, vm0):
|
|
vm0["runtime.powerState"] = "suspended"
|
|
with pytest.raises(RuntimeError, match="not running"):
|
|
driver.suspend_vm(vm0["name"])
|
|
|
|
def test_free_license_refusal_is_explained(self, driver, vm0):
|
|
from pyVmomi import vim
|
|
|
|
vm0["runtime.powerState"] = "poweredOff"
|
|
driver.mo.PowerOnVM_Task.side_effect = vim.fault.RestrictedVersion(msg="prohibited")
|
|
with pytest.raises(RuntimeError, match="free vSphere Hypervisor"):
|
|
driver.start_vm(vm0["name"])
|
|
|
|
def test_unknown_vm(self, driver):
|
|
with pytest.raises(ValueError):
|
|
driver.start_vm("nope")
|
|
|
|
|
|
def _with_snapshot(vm, name="before-upgrade", moref="snapshot-1"):
|
|
vm["snapshot"] = {
|
|
"rootSnapshotList": [
|
|
{"name": name, "snapshot": moref, "createTime": 1.0, "state": "poweredOff"}
|
|
]
|
|
}
|
|
|
|
|
|
class TestSnapshots:
|
|
def test_create(self, driver, vm0, waited):
|
|
driver.create_vm_snapshot(vm0["name"], "pre", description="d", include_memory=True)
|
|
driver.mo.CreateSnapshot_Task.assert_called_once_with(
|
|
name="pre", description="d", memory=True, quiesce=False
|
|
)
|
|
|
|
def test_memory_is_only_captured_from_a_running_vm(self, driver, vm0, waited):
|
|
vm0["runtime.powerState"] = "poweredOff"
|
|
driver.create_vm_snapshot(vm0["name"], "pre", include_memory=True)
|
|
assert driver.mo.CreateSnapshot_Task.call_args.kwargs["memory"] is False
|
|
|
|
def test_create_refuses_a_duplicate_name(self, driver, vm0):
|
|
_with_snapshot(vm0, "pre")
|
|
with pytest.raises(ValueError, match="already"):
|
|
driver.create_vm_snapshot(vm0["name"], "pre")
|
|
|
|
def test_delete_keeps_the_children(self, driver, vm0, waited):
|
|
_with_snapshot(vm0)
|
|
driver.delete_vm_snapshot(vm0["name"], "before-upgrade")
|
|
assert driver._mo.call_args.args[1] == "snapshot-1"
|
|
driver.mo.RemoveSnapshot_Task.assert_called_once_with(removeChildren=False)
|
|
|
|
def test_rollback(self, driver, vm0, waited):
|
|
_with_snapshot(vm0)
|
|
driver.rollback_vm_snapshot(vm0["name"], "before-upgrade")
|
|
driver.mo.RevertToSnapshot_Task.assert_called_once_with()
|
|
|
|
def test_unknown_snapshot(self, driver, vm0):
|
|
with pytest.raises(ValueError, match="no snapshot"):
|
|
driver.rollback_vm_snapshot(vm0["name"], "nope")
|