feat!: report hosts and licenses as facts, not warnings (0.2.0) #6

Merged
christianmanivong merged 1 commits from feat/facts-not-warnings into main 2026-10-17 08:34:58 +00:00
10 changed files with 154 additions and 100 deletions

No files matched your search

+6
View File
@@ -16,6 +16,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Declares `open-vm-tools` as the guest agent for provisioned VMs.
### Changed
- **Breaking:** `get_device_warnings` is gone; the driver reports facts and the
caller names the warnings (netOrk #1111). `get_hypervisor_hosts()` gives each
host's connection state (`None` on ESXi, which cannot tell), maintenance mode,
configuration issues and managing vCenter (ESXi only); `get_licenses()` gives
the assigned licenses with their edition key as `edition`. Requires
napalm-device-types 4.14.0; version 0.2.0.
- The guest agent is declared per guest OS (`GUEST_AGENTS`, napalm-device-types
3.0); VMware provisions Linux guests only and refuses any other `guest_os`.
- The NoCloud network-config comes from napalm-device-types
+8 -3
View File
@@ -29,7 +29,8 @@ hardware**: see [Harvesting fixtures](#harvesting-fixtures).
| `get_vm_snapshots`, `create/delete/rollback_vm_snapshot` | ✅ | ✅ | snapshot tree |
| `get_vm_storage_pools` | ✅ | ✅ | datastores |
| `get_virtual_networks` | ✅ | ✅ (+ dvPortgroups) | port groups |
| `get_device_warnings` | ✅ | ✅ | raw `{code, meta}` |
| `get_hypervisor_hosts` | ✅ | ✅ | connection state, maintenance mode, config issues, managing vCenter |
| `get_licenses` | ✅ | ✅ | license manager: name and edition key, never the key |
| `reboot_host` | ✅ (maintenance mode only) | — | `RebootHost_Task` |
| `create_vm_from_cloud_init`, `destroy_vm`, `get_vm_status` | ✅ | ✅ | see [Provisioning](#provisioning) |
| `get_network_targets`, `get_image_storages` | ✅ | ✅ | port groups, datastores |
@@ -148,5 +149,9 @@ The MoRef is reported alongside as `moref`.
hosts. Reporting them on the vCenter would attach their MACs to the wrong
device. Add a host with `vmware_esxi` to see its NICs.
**Warnings are raw.** `get_device_warnings()` returns `{code, meta}` only;
what a code means is decided in netOrk's `WARNING_CATALOG`.
**Facts, not warnings.** The driver names no warnings. `get_hypervisor_hosts()`
and `get_licenses()` report what the endpoint says; whether a host in
maintenance mode or the free `esxBasic` edition is a problem is the caller's
to decide. The ESXi driver reports `connected` as `None` (a disconnected host
could not have answered) and the managing vCenter's address as `managed_by`;
under the vCenter driver `managed_by` is `None`, since every host names it.
+11 -5
View File
@@ -6,8 +6,10 @@ from pathlib import Path
from typing import Any, ClassVar
from napalm.base.exceptions import ConnectionException
from napalm_device_types import HypervisorDriver
from napalm_device_types import HypervisorDriver, LicenseMixin
from napalm_device_types.models import (
HypervisorHostDict,
LicenseDict,
SnapshotDict,
StorageVolumeDict,
VirtualNetworkDict,
@@ -17,12 +19,13 @@ from napalm_device_types.models import (
from napalm_vmware import _session, paths
from napalm_vmware._inventory import Inventory
from napalm_vmware.parse.environment import environment
from napalm_vmware.parse.hosts import hypervisor_hosts
from napalm_vmware.parse.licenses import license_list
from napalm_vmware.parse.networks import NetworkIndex, network_index, virtual_networks
from napalm_vmware.parse.snapshots import snapshot_list
from napalm_vmware.parse.storage import storage_pools
from napalm_vmware.parse.vm_config import vm_config
from napalm_vmware.parse.vms import vm_list
from napalm_vmware.parse.warnings import host_warnings
from napalm_vmware.provision.image import DEFAULT_CACHE_DIR
_DEFAULT_PORT = 443
@@ -30,7 +33,7 @@ _DEFAULT_PORT = 443
_DRIVER_FOR_API = {"HostAgent": "vmware_esxi", "VirtualCenter": "vmware_vcenter"}
class VmwareBaseDriver(HypervisorDriver):
class VmwareBaseDriver(LicenseMixin, HypervisorDriver):
"""Shared implementation. Concrete drivers set ``API_TYPE`` and ``STANDALONE``."""
VENDOR = "VMware"
@@ -154,5 +157,8 @@ class VmwareBaseDriver(HypervisorDriver):
def get_environment(self) -> dict[str, Any]:
return environment(self._hosts())
def get_device_warnings(self) -> list[dict[str, Any]]:
return host_warnings(self._hosts(), self._inventory.licenses(), standalone=self.STANDALONE)
def get_hypervisor_hosts(self) -> list[HypervisorHostDict]:
return hypervisor_hosts(self._hosts(), standalone=self.STANDALONE)
def get_licenses(self) -> list[LicenseDict]:
return license_list(self._inventory.licenses())
+39
View File
@@ -0,0 +1,39 @@
"""The hosts a driver sees, as facts (``HypervisorDriver.get_hypervisor_hosts``).
Whether a host in maintenance mode or one managed by a vCenter is a problem is
decided by the caller, never here.
"""
from __future__ import annotations
from typing import Any
from napalm_device_types.models import HypervisorHostDict
_CONNECTED = "connected"
_UNMANAGED = {None, "", "0.0.0.0"}
def _host(host: dict[str, Any], standalone: bool) -> HypervisorHostDict:
manager = host.get("summary.managementServerIp")
return {
"name": host.get("name", host.get("_moref", "")),
# A disconnected host could not have answered the ESXi driver.
"connected": None
if standalone
else host.get("runtime.connectionState", _CONNECTED) == _CONNECTED,
"maintenance": bool(host.get("runtime.inMaintenanceMode")),
"config_issues": [
i["fullFormattedMessage"]
for i in host.get("configIssue") or []
if i.get("fullFormattedMessage")
],
# Under a vCenter every host names that vCenter; only a host the ESXi
# driver reached directly can be managed by someone else.
"managed_by": manager if standalone and manager not in _UNMANAGED else None,
}
def hypervisor_hosts(hosts: list[dict[str, Any]], *, standalone: bool) -> list[HypervisorHostDict]:
"""``standalone`` is True for the ESXi driver, False for vCenter."""
return [_host(host, standalone) for host in hosts]
+19
View File
@@ -0,0 +1,19 @@
"""Licenses assigned on an endpoint, as facts (``LicenseMixin.get_licenses``).
The edition key says which edition runs -- ``esxBasic`` is the free vSphere
Hypervisor, whose API refuses every write; what follows from that is the
caller's to decide. The license key itself is never reported.
"""
from __future__ import annotations
from typing import Any
from napalm_device_types.models import LicenseDict
def license_list(licenses: list[dict[str, Any]]) -> list[LicenseDict]:
return [
{"name": lic.get("name", ""), "status": "active", "edition": str(lic.get("editionKey", ""))}
for lic in licenses
]
-53
View File
@@ -1,53 +0,0 @@
"""Raw device warnings: ``{"code", "meta"}`` only.
What a code means to a user -- severity, wording, the fix -- is decided in
netOrk's ``WARNING_CATALOG``, never here.
"""
from __future__ import annotations
from typing import Any
_CONNECTED = "connected"
_UNMANAGED = {None, "", "0.0.0.0"}
#: License edition of the free vSphere Hypervisor. The vSphere API refuses
#: every write on it (power, snapshots) with a RestrictedVersion fault.
_FREE_EDITIONS = ("esxBasic",)
def _host_codes(host: dict[str, Any], standalone: bool) -> list[dict[str, Any]]:
name = host.get("name", host.get("_moref", ""))
found: list[dict[str, Any]] = []
state = host.get("runtime.connectionState", _CONNECTED)
if not standalone and state != _CONNECTED:
found.append({"code": "vmware_host_disconnected", "meta": {"host": name, "state": state}})
if host.get("runtime.inMaintenanceMode"):
found.append({"code": "vmware_host_maintenance_mode", "meta": {"host": name}})
issues = [
i["fullFormattedMessage"]
for i in host.get("configIssue") or []
if i.get("fullFormattedMessage")
]
if issues:
found.append({"code": "vmware_host_config_issue", "meta": {"host": name, "issues": issues}})
manager = host.get("summary.managementServerIp")
if standalone and manager not in _UNMANAGED:
found.append({"code": "vmware_host_managed_by_vcenter", "meta": {"vcenter": manager}})
return found
def host_warnings(
hosts: list[dict[str, Any]], licenses: list[dict[str, Any]], *, standalone: bool
) -> list[dict[str, Any]]:
"""Warnings for the hosts this driver sees.
``standalone`` is True for the ESXi driver. It alone reports a host being
managed by a vCenter -- a hint to add the vCenter instead -- and it never
reports disconnection, since a disconnected host could not have answered.
"""
found = [w for host in hosts for w in _host_codes(host, standalone)]
for lic in licenses:
if str(lic.get("editionKey", "")).startswith(_FREE_EDITIONS):
found.append({"code": "vmware_api_read_only", "meta": {"edition": lic.get("name", "")}})
break
return found
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-vmware"
version = "0.1.0"
version = "0.2.0"
description = "NAPALM drivers for VMware ESXi hosts and vCenter Server (vSphere API via pyVmomi)"
readme = "README.md"
requires-python = ">=3.10"
@@ -29,7 +29,7 @@ classifiers = [
]
dependencies = [
"napalm>=5.0.0",
"napalm-device-types>=3.0.0",
"napalm-device-types>=4.14.0",
"pyvmomi>=8.0.1",
# Provisioning: download the image, write the NoCloud seed ISO and its YAML.
"requests>=2.31",
+17 -3
View File
@@ -72,7 +72,8 @@ class TestIdentity:
"rollback_vm_snapshot",
"get_vm_storage_pools",
"get_virtual_networks",
"get_device_warnings",
"get_hypervisor_hosts",
"get_licenses",
],
)
@pytest.mark.parametrize("cls", DRIVERS)
@@ -204,9 +205,22 @@ class TestEsxiGetters:
with pytest.raises(RuntimeError, match="no host"):
esxi.get_facts()
def test_warnings(self, esxi, esxi_host):
def test_host_facts(self, esxi, esxi_host):
esxi_host["runtime.inMaintenanceMode"] = True
assert [w["code"] for w in esxi.get_device_warnings()] == ["vmware_host_maintenance_mode"]
assert [h["maintenance"] for h in esxi.get_hypervisor_hosts()] == [True]
def test_license_facts(self, esxi):
esxi._inventory.data["licenses"] = [{"editionKey": "esxBasic", "name": "Free"}]
assert esxi.get_licenses() == [{"name": "Free", "status": "active", "edition": "esxBasic"}]
@pytest.mark.parametrize("cls", DRIVERS)
def test_no_warnings_of_their_own(self, cls):
"""Hosts and licenses are facts; the caller names the warnings (netOrk #1111)."""
assert not any(
"get_device_warnings" in vars(c)
for c in cls.__mro__
if c.__module__.startswith("napalm_vmware")
)
def test_storage_and_networks(self, esxi):
assert list(esxi.get_vm_storage_pools())
+50 -33
View File
@@ -1,13 +1,14 @@
"""Datastores, sensors/utilisation, CDP/LLDP neighbours and raw device warnings."""
"""Datastores, sensors/utilisation, CDP/LLDP neighbours, host and license facts."""
from __future__ import annotations
import pytest
from napalm_vmware.parse.environment import environment
from napalm_vmware.parse.hosts import hypervisor_hosts
from napalm_vmware.parse.licenses import license_list
from napalm_vmware.parse.lldp import lldp_neighbors
from napalm_vmware.parse.storage import storage_pools
from napalm_vmware.parse.warnings import host_warnings
class TestStoragePools:
@@ -125,60 +126,76 @@ class TestLldpNeighbors:
assert lldp_neighbors([]) == {}
class TestHostWarnings:
class TestHypervisorHosts:
"""Facts only (netOrk #1111): which of them is a warning is the caller's call."""
def test_healthy_vcsim_host(self, esxi_host):
assert host_warnings([esxi_host], [], standalone=True) == []
assert hypervisor_hosts([esxi_host], standalone=True) == [
{
"name": "localhost.localdomain",
"connected": None,
"maintenance": False,
"config_issues": [],
"managed_by": None,
}
]
def test_maintenance_mode(self, esxi_host):
esxi_host["runtime.inMaintenanceMode"] = True
assert host_warnings([esxi_host], [], standalone=True) == [
{"code": "vmware_host_maintenance_mode", "meta": {"host": "localhost.localdomain"}}
]
assert hypervisor_hosts([esxi_host], standalone=True)[0]["maintenance"] is True
@pytest.mark.parametrize("state", ["disconnected", "notResponding"])
def test_disconnected_host_under_vcenter(self, esxi_host, state):
esxi_host["runtime.connectionState"] = state
assert host_warnings([esxi_host], [], standalone=False) == [
{
"code": "vmware_host_disconnected",
"meta": {"host": "localhost.localdomain", "state": state},
}
]
assert hypervisor_hosts([esxi_host], standalone=False)[0]["connected"] is False
def test_connected_host_under_vcenter(self, esxi_host):
assert hypervisor_hosts([esxi_host], standalone=False)[0]["connected"] is True
def test_a_standalone_host_cannot_tell(self, esxi_host):
"""A disconnected host could not have answered the ESXi driver."""
esxi_host["runtime.connectionState"] = "disconnected"
assert hypervisor_hosts([esxi_host], standalone=True)[0]["connected"] is None
def test_config_issues(self, esxi_host):
esxi_host["configIssue"] = [
{"fullFormattedMessage": "SSH for the host has been enabled"},
{"fullFormattedMessage": ""},
]
assert host_warnings([esxi_host], [], standalone=True) == [
{
"code": "vmware_host_config_issue",
"meta": {
"host": "localhost.localdomain",
"issues": ["SSH for the host has been enabled"],
},
}
assert hypervisor_hosts([esxi_host], standalone=True)[0]["config_issues"] == [
"SSH for the host has been enabled"
]
def test_standalone_driver_on_a_vcenter_managed_host(self, esxi_host):
esxi_host["summary.managementServerIp"] = "192.0.2.5"
assert host_warnings([esxi_host], [], standalone=True) == [
{"code": "vmware_host_managed_by_vcenter", "meta": {"vcenter": "192.0.2.5"}}
]
assert hypervisor_hosts([esxi_host], standalone=True)[0]["managed_by"] == "192.0.2.5"
@pytest.mark.parametrize("ip", ["", "0.0.0.0", None])
def test_unmanaged(self, esxi_host, ip):
esxi_host["summary.managementServerIp"] = ip
assert hypervisor_hosts([esxi_host], standalone=True)[0]["managed_by"] is None
def test_vcenter_driver_does_not_report_its_own_management(self, esxi_host):
esxi_host["summary.managementServerIp"] = "192.0.2.5"
assert host_warnings([esxi_host], [], standalone=False) == []
assert hypervisor_hosts([esxi_host], standalone=False)[0]["managed_by"] is None
def test_free_license_makes_the_api_read_only(self, esxi_host):
licenses = [{"editionKey": "esxBasic", "name": "VMware vSphere 8 Hypervisor"}]
assert host_warnings([esxi_host], licenses, standalone=True) == [
class TestLicenseList:
def test_the_edition_key_is_the_edition(self):
licenses = [
{
"code": "vmware_api_read_only",
"meta": {"edition": "VMware vSphere 8 Hypervisor"},
"editionKey": "esxBasic",
"name": "VMware vSphere 8 Hypervisor",
"licenseKey": "AAAAA-BBBBB-CCCCC-DDDDD-EEEEE",
}
]
assert license_list(licenses) == [
{"name": "VMware vSphere 8 Hypervisor", "status": "active", "edition": "esxBasic"}
]
def test_evaluation_license_is_not_read_only(self, esxi_host):
licenses = [{"editionKey": "eval", "name": "Evaluation Mode"}]
assert host_warnings([esxi_host], licenses, standalone=True) == []
def test_the_key_is_never_reported(self):
licenses = [{"editionKey": "eval", "name": "Evaluation Mode", "licenseKey": "00000"}]
assert "00000" not in str(license_list(licenses))
def test_none(self):
assert license_list([]) == []
+2 -1
View File
@@ -50,7 +50,8 @@ def test_esxi_reads(esxi):
assert esxi.get_vm_storage_pools()
assert esxi.get_virtual_networks()
assert "cpu" in esxi.get_environment()
assert isinstance(esxi.get_device_warnings(), list)
assert isinstance(esxi.get_hypervisor_hosts(), list)
assert isinstance(esxi.get_licenses(), list)
assert isinstance(esxi.get_lldp_neighbors(), dict)