Christian Manivong
3b3f5d9d02
driver / standard (pull_request) Successful in 18s
driver / lint (pull_request) Successful in 14s
driver / test (3.13) (pull_request) Successful in 55s
driver / types (pull_request) Successful in 1m7s
driver / test (3.12) (pull_request) Successful in 1m9s
driver / lowest (pull_request) Failing after 5s
driver / build (pull_request) Successful in 35s
driver / security (pull_request) Successful in 1m12s
CI / driver (pull_request) Failing after 2m51s
driver / publish (pull_request) Skipped
netOrk runs every driver in its own containers on Python 3.12, so 3.10 and 3.11 are no longer supported or tested. requires-python, mypy and ruff move to 3.12, along with what ruff's pyupgrade rules then rewrite.
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
"""vSphere tasks and faults, in the terms of the HypervisorDriver contract.
|
|
|
|
Contract methods raise ``RuntimeError`` when the platform refuses an
|
|
operation. vSphere reports refusals two ways: a ``MethodFault`` raised by the
|
|
call itself, or a task that ends in state ``error``. Both end up here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from pyVmomi import vmodl
|
|
|
|
from napalm_vmware._plain import to_plain
|
|
|
|
_POLL_SECONDS = 0.5
|
|
_FREE_LICENSE = (
|
|
"the host runs the free vSphere Hypervisor license, which makes the "
|
|
"vSphere API read-only; assign a paid license to change VMs from netOrk"
|
|
)
|
|
|
|
|
|
def fault_message(fault: dict[str, Any]) -> str:
|
|
"""A readable sentence for a (plain) vSphere fault."""
|
|
if fault.get("_type") == "RestrictedVersion":
|
|
return _FREE_LICENSE
|
|
return fault.get("localizedMessage") or fault.get("msg") or fault.get("_type", "unknown fault")
|
|
|
|
|
|
def invoke[T](call: Callable[..., T], *args: Any, **kwargs: Any) -> T:
|
|
"""Run a vSphere call, re-raising any ``MethodFault`` as ``RuntimeError``."""
|
|
try:
|
|
return call(*args, **kwargs)
|
|
except vmodl.MethodFault as fault:
|
|
raise RuntimeError(fault_message(to_plain(fault))) from fault
|
|
|
|
|
|
def wait_for_task(
|
|
inventory: Any,
|
|
task: Any,
|
|
timeout: float,
|
|
*,
|
|
clock: Callable[[], float] = time.monotonic,
|
|
sleep: Callable[[float], None] = time.sleep,
|
|
) -> None:
|
|
"""Block until ``task`` succeeds; raise ``RuntimeError`` otherwise."""
|
|
deadline = clock() + timeout
|
|
while True:
|
|
info = inventory.properties(task, ["info.state", "info.error"])
|
|
if not info:
|
|
raise RuntimeError("the vSphere task disappeared before it finished")
|
|
state = info.get("info.state")
|
|
if state == "success":
|
|
return
|
|
if state == "error":
|
|
raise RuntimeError(fault_message(info.get("info.error") or {}))
|
|
if clock() >= deadline:
|
|
raise RuntimeError(f"the vSphere task did not finish within {timeout:g}s")
|
|
sleep(_POLL_SECONDS)
|
|
|
|
|
|
def wait_until(
|
|
check: Callable[[], bool],
|
|
timeout: float,
|
|
what: str,
|
|
*,
|
|
clock: Callable[[], float] = time.monotonic,
|
|
sleep: Callable[[float], None] = time.sleep,
|
|
) -> None:
|
|
"""Poll ``check`` until it is true; for operations vSphere runs without a task.
|
|
|
|
``ShutdownGuest`` only *asks* VMware Tools to shut the guest down and
|
|
returns at once, so completion has to be observed from the power state.
|
|
"""
|
|
deadline = clock() + timeout
|
|
while not check():
|
|
if clock() >= deadline:
|
|
raise RuntimeError(f"{what} did not finish within {timeout:g}s")
|
|
sleep(_POLL_SECONDS)
|