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.
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
"""Waiting for vSphere tasks, and turning vSphere faults into RuntimeError."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from pyVmomi import vim
|
|
|
|
from napalm_vmware._tasks import fault_message, invoke, wait_for_task
|
|
|
|
|
|
class _Clock:
|
|
def __init__(self):
|
|
self.now = 0.0
|
|
|
|
def __call__(self):
|
|
return self.now
|
|
|
|
def sleep(self, seconds):
|
|
self.now += seconds
|
|
|
|
|
|
def _inventory(*states):
|
|
inv = MagicMock()
|
|
inv.properties.side_effect = list(states)
|
|
return inv
|
|
|
|
|
|
class TestWaitForTask:
|
|
def test_returns_once_the_task_succeeds(self):
|
|
inv = _inventory({"info.state": "running"}, {"info.state": "success"})
|
|
clock = _Clock()
|
|
wait_for_task(inv, "task", timeout=10, clock=clock, sleep=clock.sleep)
|
|
assert inv.properties.call_count == 2
|
|
|
|
def test_error_becomes_runtime_error_with_the_fault_text(self):
|
|
inv = _inventory(
|
|
{
|
|
"info.state": "error",
|
|
"info.error": {
|
|
"_type": "InvalidPowerState",
|
|
"msg": "The attempted operation cannot be performed "
|
|
"in the current state (Powered off).",
|
|
},
|
|
}
|
|
)
|
|
with pytest.raises(RuntimeError, match="current state"):
|
|
wait_for_task(inv, "task", timeout=10)
|
|
|
|
def test_times_out(self):
|
|
clock = _Clock()
|
|
inv = MagicMock()
|
|
inv.properties.return_value = {"info.state": "running"}
|
|
with pytest.raises(RuntimeError, match="did not finish within 5s"):
|
|
wait_for_task(inv, "task", timeout=5, clock=clock, sleep=clock.sleep)
|
|
|
|
def test_vanished_task_is_an_error(self):
|
|
with pytest.raises(RuntimeError, match="disappeared"):
|
|
wait_for_task(_inventory({}), "task", timeout=5)
|
|
|
|
|
|
class TestFaultMessage:
|
|
def test_free_license(self):
|
|
msg = fault_message(
|
|
{
|
|
"_type": "RestrictedVersion",
|
|
"msg": "Current license or ESXi version prohibits "
|
|
"execution of the requested operation.",
|
|
}
|
|
)
|
|
assert "free" in msg.lower()
|
|
assert "read-only" in msg
|
|
|
|
def test_localized_message_preferred(self):
|
|
assert fault_message({"_type": "X", "localizedMessage": "L", "msg": "M"}) == "L"
|
|
|
|
def test_bare_type(self):
|
|
assert fault_message({"_type": "NotSupported"}) == "NotSupported"
|
|
|
|
|
|
class TestInvoke:
|
|
def test_passes_the_result_through(self):
|
|
assert invoke(lambda x: x + 1, 1) == 2
|
|
|
|
def test_method_fault_becomes_runtime_error(self):
|
|
def boom():
|
|
raise vim.fault.RestrictedVersion(msg="prohibited")
|
|
|
|
with pytest.raises(RuntimeError, match="free"):
|
|
invoke(boom)
|
|
|
|
def test_other_exceptions_are_left_alone(self):
|
|
def boom():
|
|
raise KeyError("x")
|
|
|
|
with pytest.raises(KeyError):
|
|
invoke(boom)
|
|
|
|
|
|
class TestWaitUntil:
|
|
def test_returns_when_the_check_passes(self):
|
|
from napalm_vmware._tasks import wait_until
|
|
|
|
answers = iter([False, False, True])
|
|
clock = _Clock()
|
|
wait_until(lambda: next(answers), timeout=10, what="x", clock=clock, sleep=clock.sleep)
|
|
assert clock.now == 1.0
|
|
|
|
def test_times_out_naming_what_it_waited_for(self):
|
|
from napalm_vmware._tasks import wait_until
|
|
|
|
clock = _Clock()
|
|
with pytest.raises(RuntimeError, match="guest shutdown did not finish within 3s"):
|
|
wait_until(
|
|
lambda: False, timeout=3, what="guest shutdown", clock=clock, sleep=clock.sleep
|
|
)
|